diff --git a/Dockerfile.web-rust b/Dockerfile.web-rust index 786b0577c..9f7d0b28b 100644 --- a/Dockerfile.web-rust +++ b/Dockerfile.web-rust @@ -149,6 +149,11 @@ COPY --from=builder --chown=10001:10001 /out/web-bundle /app/web-bundle # The daemon writes nothing outside /tmp in online mode (settings persistence # is refused there), so it runs unprivileged and the image can be mounted # read-only by the orchestrator. +# The online data directory, created with the runtime owner so a fresh named +# volume inherits that ownership and mode on its first mount. The daemon probes +# it at start-up and refuses to run if it is not writable. +RUN install -d -o 10001 -g 10001 -m 0700 /data + USER 10001:10001 EXPOSE ${SERVE_PORT} diff --git a/crates/op-collab-host/src/lib.rs b/crates/op-collab-host/src/lib.rs index 182061376..64266c68e 100644 --- a/crates/op-collab-host/src/lib.rs +++ b/crates/op-collab-host/src/lib.rs @@ -21,6 +21,7 @@ mod runtime; pub use blocking::{install_blocking_executor, BlockingExecutor}; pub use host::{CollabHost, CollabWakeNotifier, HeadlessCollabHost}; +pub use runtime::local_edit::LocalEditOutcome; pub use runtime::types::{CollabRuntimeFailure, CollabStatusEvent}; pub use runtime::CollabRuntime; diff --git a/crates/op-collab-host/src/runtime.rs b/crates/op-collab-host/src/runtime.rs index 018aec704..c5a4d579d 100644 --- a/crates/op-collab-host/src/runtime.rs +++ b/crates/op-collab-host/src/runtime.rs @@ -8,7 +8,7 @@ mod effects_wire; mod failure; mod guest_confirmation; mod guest_routes; -mod local_edit; +pub(crate) mod local_edit; mod network; mod poll; mod region_pref; @@ -63,6 +63,14 @@ const MAX_STATUS_EVENTS: usize = 64; /// `refresh_availability` → `drain_ui_action` → `poll`, plus the local-edit /// capture pair around every mutating gesture. pub struct CollabRuntime { + /// What the last completed local-edit capture resolved to, recorded by + /// the effect-routing layer and consumed by `finish_local_edit`. + /// + /// A field rather than a return value because the resolution is decided + /// several frames down inside `route_owner_output` / `route_guest_output`, + /// which are also reached from `poll` — threading it back through every + /// effect signature would touch far more than the one decision. + pub(crate) last_local_edit: Option, events: Receiver, event_sender: SyncSender, terminal_event_sender: TerminalEventLane, @@ -137,6 +145,7 @@ impl CollabRuntime { let (event_sender, events, terminal_event_sender, terminal_events) = network::event_channel(); Self { + last_local_edit: None, events, event_sender, terminal_event_sender, diff --git a/crates/op-collab-host/src/runtime/conflict_tests.rs b/crates/op-collab-host/src/runtime/conflict_tests.rs index be884a38b..1bda7623c 100644 --- a/crates/op-collab-host/src/runtime/conflict_tests.rs +++ b/crates/op-collab-host/src/runtime/conflict_tests.rs @@ -83,7 +83,10 @@ fn property_conflict_stashes_discarded_edit_and_reapply_resubmits_it() { // Guest optimistically renames the shared node. assert!(runtime.begin_local_edit(&mut host)); host.editor_state_mut().doc = document_named("Guest intent"); - assert!(runtime.finish_local_edit(&mut host)); + assert_ne!( + runtime.finish_local_edit(&mut host), + crate::runtime::local_edit::LocalEditOutcome::Failed + ); let _submit = commands.recv_timeout(Duration::from_secs(1)).unwrap(); // The owner concurrently renames the same field and wins seq 1. diff --git a/crates/op-collab-host/src/runtime/effects.rs b/crates/op-collab-host/src/runtime/effects.rs index cc7fefc34..0fe8a87eb 100644 --- a/crates/op-collab-host/src/runtime/effects.rs +++ b/crates/op-collab-host/src/runtime/effects.rs @@ -1,6 +1,8 @@ use std::sync::Arc; use std::time::{Duration, Instant}; +use crate::runtime::local_edit::LocalEditOutcome; + use op_collab::{ Bye, ByeReason, CollabMessage, ConnectionKey, GuestEffect, OwnerEffect, ParticipantPresence, Point, Presence, UndoOutcome, UndoResult, Viewport, @@ -208,15 +210,20 @@ impl CollabRuntime { self.close_failed_owner_peer(owner, connection, host)?; } if let Some(local) = local_edit { - match local { - LocalEditResolution::NoChange | LocalEditResolution::Committed(_) => {} + // Recorded so `finish_local_edit` can report what actually + // happened. Swallowing the rejection here is what let a rolled-back + // push be answered 200. + self.last_local_edit = Some(match local { + LocalEditResolution::NoChange => LocalEditOutcome::NoChange, + LocalEditResolution::Committed(_) => LocalEditOutcome::Committed, LocalEditResolution::Rejected(_) => { self.set_notice( host, CollabNoticeKind::Reject(CollabRejectUiCode::Unsupported), ); + LocalEditOutcome::Rejected } - } + }); } for effect in effects { self.route_owner_effect(owner, effect, host)?; @@ -323,19 +330,23 @@ impl CollabRuntime { deliver_outbound: bool, ) -> Result<(), CollabRuntimeError> { if let Some(local) = output.local_edit { - match local { - GuestLocalEditResolution::NoChange => {} + self.last_local_edit = Some(match local { + GuestLocalEditResolution::NoChange => LocalEditOutcome::NoChange, GuestLocalEditResolution::Submitted => { host.editor_state_mut().editor_ui.collab.pending_edit = CollabPendingEditUi::Submitting; + // Submitted, not yet acknowledged — but it IS on the wire + // and the local document holds it, so the push landed. + LocalEditOutcome::Committed } GuestLocalEditResolution::Rejected(_) => { self.set_notice( host, CollabNoticeKind::Reject(CollabRejectUiCode::Unsupported), ); + LocalEditOutcome::Rejected } - } + }); } for effect in output.effects { match effect { diff --git a/crates/op-collab-host/src/runtime/local_edit.rs b/crates/op-collab-host/src/runtime/local_edit.rs index 06a428423..b5c1e1222 100644 --- a/crates/op-collab-host/src/runtime/local_edit.rs +++ b/crates/op-collab-host/src/runtime/local_edit.rs @@ -2,6 +2,25 @@ //! conflict-discarded edit. Split off the runtime spine at the //! 800-line cap; pure code motion. +/// What a session did with a completed local-edit capture. +/// +/// `finish_local_edit` used to return `bool`, which could not distinguish +/// "the session sequenced this" from "the session rejected it and rolled it +/// back" — so a rejected whole-document push was answered 200 with a version +/// bump and the browser never learned to refetch. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LocalEditOutcome { + /// Sequenced and committed. + Committed, + /// The capture closed with nothing to send — the document did not move. + NoChange, + /// The session refused the edit and rolled it back. The caller's copy is + /// now behind and must be refetched. + Rejected, + /// The capture could not be opened or closed cleanly. + Failed, +} + use op_editor_core::{CollabNoticeKind, CollabRejectUiCode}; use super::actor::EditorActor; @@ -42,13 +61,16 @@ impl CollabRuntime { } } - pub fn finish_local_edit(&mut self, host: &mut impl CollabHost) -> bool { + pub fn finish_local_edit(&mut self, host: &mut impl CollabHost) -> LocalEditOutcome { if !std::mem::take(&mut self.transaction_active) { - return false; + return LocalEditOutcome::Failed; } let Some(mut actor) = self.actor.take() else { - return false; + return LocalEditOutcome::Failed; }; + // Cleared here so a stale resolution from an earlier edit cannot be + // read as this one's answer. + self.last_local_edit = None; let result = match &mut actor { EditorActor::Owner(owner) => match owner.session.finish_local_edit(host) { Ok(output) => self.route_owner_output(owner, output, host), @@ -61,13 +83,20 @@ impl CollabRuntime { }; self.actor = Some(actor); match result { - Ok(()) => true, + // The routing layer records what the session actually decided; a + // bare `Ok` used to be reported as success even when the session + // had REJECTED and rolled the edit back, so the caller answered + // 200 for a write that never landed. + Ok(()) => self + .last_local_edit + .take() + .unwrap_or(LocalEditOutcome::Committed), Err(error) => { // A local editor/core failure can occur after the document // changed or the sequencer prepared a commit. Continuing to // advertise Active would silently fork owner and guests. self.fail_network(host, error.failure); - false + LocalEditOutcome::Failed } } } diff --git a/crates/op-collab-host/src/runtime/tests.rs b/crates/op-collab-host/src/runtime/tests.rs index 717ef40c5..7ac258506 100644 --- a/crates/op-collab-host/src/runtime/tests.rs +++ b/crates/op-collab-host/src/runtime/tests.rs @@ -170,7 +170,10 @@ fn reliable_owner_delivery_failure_falls_back_to_standalone() { assert!(runtime.begin_local_edit(&mut host)); host.editor_state_mut().doc = document_named("Changed"); - assert!(!runtime.finish_local_edit(&mut host)); + assert_eq!( + runtime.finish_local_edit(&mut host), + crate::runtime::local_edit::LocalEditOutcome::Failed + ); assert!(runtime.actor.is_none()); assert!(runtime.network.is_none()); assert_eq!( @@ -199,7 +202,10 @@ fn commit_broadcast_reuses_one_encoded_allocation_across_peer_commands() { assert!(runtime.begin_local_edit(&mut host)); host.editor_state_mut().doc = document_named("Shared encoded commit"); - assert!(runtime.finish_local_edit(&mut host)); + assert_ne!( + runtime.finish_local_edit(&mut host), + crate::runtime::local_edit::LocalEditOutcome::Failed + ); let mut queued = Vec::new(); for _ in 0..2 { @@ -558,7 +564,10 @@ fn retry_against_new_epoch_ends_without_replaying_pending_edit() { let (mut runtime, mut host, commands, original_connection, welcome) = guest_runtime(8); assert!(runtime.begin_local_edit(&mut host)); host.editor_state_mut().doc = document_named("Changed"); - assert!(runtime.finish_local_edit(&mut host)); + assert_ne!( + runtime.finish_local_edit(&mut host), + crate::runtime::local_edit::LocalEditOutcome::Failed + ); assert!(matches!( commands.recv_timeout(Duration::from_secs(1)).unwrap(), GuestNetworkCommand::Send { diff --git a/crates/op-host-services/src/web_canvas_server.rs b/crates/op-host-services/src/web_canvas_server.rs index 516f1b617..3be757f97 100644 --- a/crates/op-host-services/src/web_canvas_server.rs +++ b/crates/op-host-services/src/web_canvas_server.rs @@ -278,8 +278,12 @@ impl WebCanvasState { }); } } - let editor_meta = push.editor_meta; - let Some(prepared) = push.prepared else { + // Taken rather than moved: `PendingDocumentPush` owns a `Drop` that + // releases the thumbnail seed of a push that never installs, and + // taking `prepared` here is what tells it this one DID. + let mut push = push; + let editor_meta = push.take_editor_meta(); + let Some(prepared) = push.prepared.take() else { // Active-page changes do not mutate canonical document content. // Apply the scalar pair without replacing the identical document, // bumping its generation, or publishing a version that another @@ -318,16 +322,21 @@ impl WebCanvasState { // is what the old `bool` did for `Failed`) left the browser // believing a write landed that had been discarded, with no // signal to resync. + // Both mean "your push did not land, refetch". The browser's + // recovery only fires on `version-conflict` WITH a `version` + // (see `WebSyncClient::parse_push_conflict`), so both carry + // the authoritative version — a bare refusal would leave the + // tab latched with no way to resync. collab_state::IngestOutcome::Rejected => { - return Err(WebCanvasError::Collab( - collab_state::DaemonMutationRefusal::Collab( - op_editor_core::CollabGateReason::PendingEdit, - ), + return Err(WebCanvasError::IngestRejected( + collab_state::IngestOutcome::Rejected, + self.version, )); } collab_state::IngestOutcome::Failed => { return Err(WebCanvasError::IngestRejected( collab_state::IngestOutcome::Failed, + self.version, )); } } @@ -360,24 +369,6 @@ impl WebCanvasState { /// A collaboration refusal is not a malformed request, so a client needs the /// code to decide between retrying, refetching, and telling the user to leave /// the session. -fn collab_aware_error_reply(error: &WebCanvasError) -> WebReply { - match error.error_code() { - Some(code) => WebReply { - status: error.http_status(), - body: serde_json::json!({ - "ok": false, - "error": code, - "message": error.to_string(), - }) - .to_string(), - }, - None => WebReply { - status: error.http_status(), - body: crate::mcp_serve::rest_error_body(&error.to_string()), - }, - } -} - /// Outcome of [`WebCanvasState::reset_document_guarded`]. pub(crate) struct ResetOutcome { pub skipped: bool, @@ -705,34 +696,6 @@ pub fn handle_web_canvas_request( } } -/// Render the reply for an already-parsed document push. -/// -/// Shares its verdict shape with the in-handler route below, so the two paths -/// cannot answer the same outcome differently. -pub(crate) fn document_push_reply( - parsed: Result, - state: &mut WebCanvasState, -) -> WebReply { - match parsed.and_then(|push| state.apply_prepared_document_push(push, None)) { - Ok(outcome) if outcome.applied => WebReply { - status: "200 OK", - body: crate::mcp_serve::document_sync_ok(outcome.current_version), - }, - Ok(outcome) => WebReply { - // Stale baseVersion: reject without writing, plus the current - // version so the caller can decide whether to refetch and retry. - status: "409 Conflict", - body: serde_json::json!({ - "ok": false, - "error": "version-conflict", - "version": outcome.current_version, - }) - .to_string(), - }, - Err(error) => collab_aware_error_reply(&error), - } -} - /// The daemon's canonical unknown-route reply. fn not_found_reply() -> WebReply { WebReply { @@ -749,7 +712,11 @@ fn is_device_login_route(path: &str) -> bool { mod collab_driver; mod document_push; -pub(crate) use document_push::PendingDocumentPush; +#[cfg(test)] +pub(crate) use document_push::collab_aware_error_reply_for_test; +pub(crate) use document_push::{document_push_reply, PendingDocumentPush}; +// Spine-local: the shared coded-error renderer for the document routes. +use document_push::collab_aware_error_reply; mod sse_hub; pub use sse_hub::SseHub; pub(crate) use sse_hub::SseSlot; diff --git a/crates/op-host-services/src/web_canvas_server/collab_state.rs b/crates/op-host-services/src/web_canvas_server/collab_state.rs index b6a399852..3dd4752cf 100644 --- a/crates/op-host-services/src/web_canvas_server/collab_state.rs +++ b/crates/op-host-services/src/web_canvas_server/collab_state.rs @@ -171,10 +171,10 @@ impl IngestOutcome { pub const fn error_code(self) -> Option<&'static str> { match self { Self::Committed | Self::NoChange => None, - Self::Rejected => Some("collab-busy"), - // Shares the existing conflict code so the browser's established - // refetch path handles it without a new client branch. - Self::Failed => Some("version-conflict"), + // Both share the existing conflict code so the browser's + // established refetch path handles them without a new client + // branch — `parse_push_conflict` only recognises this one. + Self::Rejected | Self::Failed => Some("version-conflict"), } } } @@ -193,10 +193,10 @@ impl LocalEditCapture<'_> { self.state.as_mut().expect("capture is open") } - /// Close the capture, reporting whether the session committed it. - fn finish(mut self) -> bool { + /// Close the capture, reporting what the session decided. + fn finish(mut self) -> op_collab_host::LocalEditOutcome { let Some(state) = self.state.take() else { - return false; + return op_collab_host::LocalEditOutcome::Failed; }; let (runtime, mut host) = state.collab_runtime_and_host(); runtime.finish_local_edit(&mut host) @@ -341,7 +341,6 @@ impl WebCanvasState { &mut self, prepared: PreparedDocument, ) -> IngestOutcome { - let revision_before = self.editor.document_revision(); let (runtime, mut host) = self.collab_runtime_and_host(); if !runtime.begin_local_edit(&mut host) { return IngestOutcome::Rejected; @@ -355,16 +354,16 @@ impl WebCanvasState { .state_mut() .editor .install_prepared_document(prepared, EditOrigin::Local); - let committed = capture.finish(); - if !committed { - return IngestOutcome::Failed; + // Straight from the session's own resolution. The previous revision + // heuristic could not see a REJECTION: the session rolls the edit back, + // so the revision is unchanged and the push read as `NoChange` — a + // 200 for a write that had been discarded. + match capture.finish() { + op_collab_host::LocalEditOutcome::Committed => IngestOutcome::Committed, + op_collab_host::LocalEditOutcome::NoChange => IngestOutcome::NoChange, + op_collab_host::LocalEditOutcome::Rejected => IngestOutcome::Rejected, + op_collab_host::LocalEditOutcome::Failed => IngestOutcome::Failed, } - if self.editor.document_revision() == revision_before { - // The session diffed the document and found nothing to send. The - // push succeeded; there is simply no new version to publish. - return IngestOutcome::NoChange; - } - IngestOutcome::Committed } } diff --git a/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs index 046567dc2..9acbec16f 100644 --- a/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/collab_state_tests.rs @@ -314,8 +314,33 @@ fn a_rejected_ingest_is_distinguishable_from_an_accepted_one() { use crate::web_canvas_server::IngestOutcome; assert_eq!(IngestOutcome::Committed.error_code(), None); assert_eq!(IngestOutcome::NoChange.error_code(), None); - assert_eq!(IngestOutcome::Rejected.error_code(), Some("collab-busy")); - // Shares the browser's existing conflict code so its established refetch - // path handles it with no new client branch. + // Both rejections use the code the browser recovers from: its + // `parse_push_conflict` only recognises `version-conflict`, so any other + // code would leave the tab latched with no way to resync. + assert_eq!( + IngestOutcome::Rejected.error_code(), + Some("version-conflict") + ); assert_eq!(IngestOutcome::Failed.error_code(), Some("version-conflict")); } + +#[test] +fn a_rejected_ingest_answers_409_with_the_authoritative_version() { + // The browser's recovery only fires on `version-conflict` carrying a + // `version` (`WebSyncClient::parse_push_conflict`). A 409 without one + // leaves the tab latched with nothing to refetch from. + use crate::web_canvas_server::IngestOutcome; + for outcome in [IngestOutcome::Rejected, IngestOutcome::Failed] { + let error = crate::web_canvas_server_error::WebCanvasError::IngestRejected(outcome, 42); + assert_eq!(error.http_status(), "409 Conflict", "{outcome:?}"); + let reply = crate::web_canvas_server::collab_aware_error_reply_for_test(&error); + let body: serde_json::Value = serde_json::from_str(&reply.body).expect("json"); + assert_eq!(body["ok"], false, "{outcome:?}"); + assert_eq!(body["error"], "version-conflict", "{outcome:?}"); + assert_eq!(body["version"], 42, "{outcome:?}"); + assert!( + op_editor_core::web_sync::WebSyncClient::parse_push_conflict(&reply.body).is_some(), + "{outcome:?}: the browser must be able to parse its recovery version" + ); + } +} diff --git a/crates/op-host-services/src/web_canvas_server/connection.rs b/crates/op-host-services/src/web_canvas_server/connection.rs index 087f77918..d735686ee 100644 --- a/crates/op-host-services/src/web_canvas_server/connection.rs +++ b/crates/op-host-services/src/web_canvas_server/connection.rs @@ -29,10 +29,7 @@ pub(super) struct ConnCtx<'a> { /// /// `None` for the local and managed daemons, which have no per-request /// identity and are unrestricted — the REST scope gate is skipped whole. - pub(super) rest_identity: Option<( - super::tenant_auth::IdentityVia, - crate::mcp_serve::tool_profile::McpScopes, - )>, + pub(super) rest_identity: Option, } /// Handle one connection against the single-user document authority. @@ -198,6 +195,47 @@ pub(super) fn dispatch( )?; return Ok(false); } + // Scopes apply to every credentialed route, not just the REST tier and + // `/mcp`. They used to be checked inside the `/api/*` branch, which sits + // BELOW the specially dispatched routes (AI streams, SSE, figma) — so a + // read-only token could drive all of those. Checked here, ahead of every + // branch, there is no route left to slip past it. + if let Some(identity) = ctx.rest_identity.as_ref() { + if let Some(refusal) = super::tool_scopes::check_rest_scope( + identity.via, + identity.scopes, + &req.method, + &req.path, + ) { + crate::mcp_serve::write_mcp_http_response_with_origin( + stream, + refusal.http_status(), + &serde_json::json!({ + "ok": false, + "error": refusal.code(), + "message": refusal.to_string(), + }) + .to_string(), + cors_origin, + )?; + return Ok(false); + } + } + // Online account projection. The device-login proxy stays 404 (it drives + // a process-wide device session), but the shell must be able to learn + // which account it is showing — without this the identity epoch never + // fires and an account switch leaks the previous account's document. + if req.method == "GET" && req.path == op_editor_core::auth_routes::STATUS { + if let Some(identity) = ctx.rest_identity.as_ref() { + crate::mcp_serve::write_mcp_http_response_with_origin( + stream, + "200 OK", + &identity.auth_status_json(), + cors_origin, + )?; + return Ok(false); + } + } // Current-account avatar proxy: performs bounded public HTTPS I/O on this // connection thread, never while holding the editor-state mutex. Part of // the device-login proxy, so it is off wherever that is. @@ -285,27 +323,6 @@ pub(super) fn dispatch( // daemon doesn't implement yet, which it answers with 404 rather than // mis-routing them into the JSON-RPC dispatch below. if req.path.starts_with("/api/") { - // Scopes apply to REST exactly as they apply to `/mcp`. Without this - // a read-only token could replace the whole document here — strictly - // more damage than any tool call it is refused. - if let Some((via, scopes)) = ctx.rest_identity { - if let Some(refusal) = - super::tool_scopes::check_rest_scope(via, scopes, &req.method, &req.path) - { - crate::mcp_serve::write_mcp_http_response_with_origin( - stream, - refusal.http_status(), - &serde_json::json!({ - "ok": false, - "error": refusal.code(), - "message": refusal.to_string(), - }) - .to_string(), - cors_origin, - )?; - return Ok(false); - } - } // Parse the whole-document push BEFORE taking the state lock. A push // can carry megabytes of embedded images, and parsing it under the // lock stalled every other request to this tenant — including the diff --git a/crates/op-host-services/src/web_canvas_server/document_push.rs b/crates/op-host-services/src/web_canvas_server/document_push.rs index 2d54a1140..c0d6e67c2 100644 --- a/crates/op-host-services/src/web_canvas_server/document_push.rs +++ b/crates/op-host-services/src/web_canvas_server/document_push.rs @@ -3,7 +3,7 @@ //! Split out of the `web_canvas_server` spine at the 800-line cap. The point //! of the type is WHERE it is built — see [`PendingDocumentPush`]. -use super::{Result, ServeMode, WebCanvasError}; +use super::{Result, ServeMode, WebCanvasError, WebCanvasState, WebReply}; /// A whole-document push, parsed and validated but not yet installed. /// @@ -21,6 +21,33 @@ pub(crate) struct PendingDocumentPush { pub(super) prepared: Option, } +impl PendingDocumentPush { + /// Take the editor metadata, leaving an empty one behind. + /// + /// A method because the type owns a `Drop`, which forbids moving fields + /// out of it directly. + pub(super) fn take_editor_meta(&mut self) -> op_pen_loader::EditorMeta { + std::mem::take(&mut self.editor_meta) + } +} + +impl Drop for PendingDocumentPush { + /// Release the pending thumbnail seed of a push that was parsed and then + /// dropped — a stale `baseVersion`, a collaboration refusal, or a + /// connection that went away between parse and install. + /// + /// `load_canonical` registers the seed in a process-global side table + /// keyed by the document, and only an activation consumes it. A push that + /// never installs would leave it there until bounded eviction pushed it + /// out, and in the meantime a *different* document that happened to reuse + /// the key could activate someone else's thumbnails. + fn drop(&mut self) { + if let Some(prepared) = self.prepared.as_ref() { + jian_ops_schema::image_thumbs::discard_for_document(prepared.document()); + } + } +} + impl PendingDocumentPush { /// Parse and validate a push body. **Call this OUTSIDE the state lock.** /// @@ -67,3 +94,68 @@ impl PendingDocumentPush { }) } } + +/// Render the reply for an already-parsed document push. +/// +/// Shares its verdict shape with the in-handler route below, so the two paths +/// cannot answer the same outcome differently. +pub(crate) fn document_push_reply( + parsed: Result, + state: &mut WebCanvasState, +) -> WebReply { + match parsed.and_then(|push| state.apply_prepared_document_push(push, None)) { + Ok(outcome) if outcome.applied => WebReply { + status: "200 OK", + body: crate::mcp_serve::document_sync_ok(outcome.current_version), + }, + Ok(outcome) => WebReply { + // Stale baseVersion: reject without writing, plus the current + // version so the caller can decide whether to refetch and retry. + status: "409 Conflict", + body: serde_json::json!({ + "ok": false, + "error": "version-conflict", + "version": outcome.current_version, + }) + .to_string(), + }, + Err(error) => collab_aware_error_reply(&error), + } +} + +#[cfg(test)] +pub(crate) fn collab_aware_error_reply_for_test(error: &WebCanvasError) -> WebReply { + collab_aware_error_reply(error) +} + +pub(super) fn collab_aware_error_reply(error: &WebCanvasError) -> WebReply { + // An ingest rejection must carry the authoritative version: it is what the + // browser refetches from, and without it the conflict recovery never runs. + if let WebCanvasError::IngestRejected(_, version) = error { + return WebReply { + status: error.http_status(), + body: serde_json::json!({ + "ok": false, + "error": error.error_code().unwrap_or("version-conflict"), + "version": version, + "message": error.to_string(), + }) + .to_string(), + }; + } + match error.error_code() { + Some(code) => WebReply { + status: error.http_status(), + body: serde_json::json!({ + "ok": false, + "error": code, + "message": error.to_string(), + }) + .to_string(), + }, + None => WebReply { + status: error.http_status(), + body: crate::mcp_serve::rest_error_body(&error.to_string()), + }, + } +} diff --git a/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs b/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs index 57127e2c4..65cfe8eab 100644 --- a/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs @@ -307,3 +307,74 @@ fn a_browser_session_is_not_scope_limited_over_rest() { ); assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); } + +#[test] +fn a_read_scope_token_cannot_reach_the_specially_dispatched_write_routes() { + // These are dispatched ahead of the `/api/*` branch the gate used to live + // in, so a read-only token could drive all of them. + let registry = registry(); + let verifier = scoped_verifier(); + for (method, path, body) in [ + ( + "POST", + op_editor_core::share_routes::GRANT, + r#"{"userId":"userB"}"#, + ), + ( + "POST", + op_editor_core::share_routes::REVOKE, + r#"{"userId":"userB"}"#, + ), + ("POST", "/api/ai/standard", "{}"), + ("POST", "/api/ai/stream", "{}"), + ("POST", "/api/figma/convert", "{}"), + ( + "POST", + op_editor_core::collab_routes::ACTION, + r#"{"type":"openCreate"}"#, + ), + ] { + let response = serve( + ®istry, + &verifier, + Request::json(method, path, body).with_bearer("tokR"), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 403 Forbidden", + "{method} {path}: {response}" + ); + assert_eq!(body_of(&response)["error"], "scope-insufficient", "{path}"); + } +} + +#[test] +fn a_scopeless_token_cannot_even_subscribe_to_the_event_stream() { + // SSE is a GET, so it needs `mcp:read` — and a scopeless token has none. + let response = serve( + ®istry(), + &StaticVerifier::parse("tokN=userN:none"), + Request::new("GET", "/api/mcp/events").with_bearer("tokN"), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 403 Forbidden", + "{response}" + ); +} + +#[test] +fn a_read_scope_token_may_still_subscribe_to_the_event_stream() { + // A read token reading is the whole point; only writes are refused. + let response = serve( + ®istry(), + &scoped_verifier(), + Request::new("GET", op_editor_core::collab_routes::STATE).with_bearer("tokR"), + ); + assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); +} + +/// The share/AI routes answer with a plain body; reuse the shared decoder. +fn body_of(response: &str) -> serde_json::Value { + body(response) +} diff --git a/crates/op-host-services/src/web_canvas_server/online_policy.rs b/crates/op-host-services/src/web_canvas_server/online_policy.rs index 667021103..b7ba97839 100644 --- a/crates/op-host-services/src/web_canvas_server/online_policy.rs +++ b/crates/op-host-services/src/web_canvas_server/online_policy.rs @@ -199,11 +199,16 @@ pub(super) fn cookie_write_origin_allowed(allow: &[String], origin: Option<&str> /// paints nothing. pub(super) const EMPTY_INDICATOR_RELAY: &str = r#"{"epoch":0,"active":false,"cursorAgent":null,"nodes":[],"frames":[],"previews":[],"reveals":[]}"#; -/// REST paths an API token may reach without any scope. +/// Paths the method-based REST scope rule does not decide. /// -/// Only the deployment health probe: a client has to be able to discover the -/// daemon exists before it can be told its scopes are insufficient. -const SCOPE_EXEMPT_READS: &[&str] = &["/api/mcp/server"]; +/// - `/api/mcp/server` is the deployment health probe: a client has to be able +/// to discover the daemon exists before it can be told its scopes are +/// insufficient. +/// - `/mcp` and its `/` alias are JSON-RPC, where the MCP dispatch runs a +/// strictly better check — it knows which TOOL is being called, so a +/// read-only token can still call read tools. Applying the coarse +/// "POST means write" rule here would refuse those outright. +const SCOPE_EXEMPT_PATHS: &[&str] = &["/api/mcp/server", "/mcp", "/"]; /// Which scope a REST request needs, if any. /// @@ -214,7 +219,7 @@ pub(super) fn rest_scope_required( method: &str, path: &str, ) -> Option { - if SCOPE_EXEMPT_READS.contains(&path) { + if SCOPE_EXEMPT_PATHS.contains(&path) { return None; } match method { diff --git a/crates/op-host-services/src/web_canvas_server/online_run_loop.rs b/crates/op-host-services/src/web_canvas_server/online_run_loop.rs index 9d75587ab..773c56f13 100644 --- a/crates/op-host-services/src/web_canvas_server/online_run_loop.rs +++ b/crates/op-host-services/src/web_canvas_server/online_run_loop.rs @@ -39,6 +39,27 @@ use super::tenant::{now_unix, TenantLimits, TenantRegistry}; use super::tenant_auth::{IdentityVerifier, PresentedCredentials, StaticVerifier}; use super::*; +/// How long a controlled shutdown waits for in-flight requests before it +/// writes. Long enough for a large document push to finish installing, short +/// enough that a deploy is not held up by a long-lived SSE stream. +const SHUTDOWN_DRAIN_SECS: u64 = 10; + +/// Wait for the active-connection count to reach zero, up to the bound. +/// +/// Returns whether it actually drained. SSE streams routinely outlive this, +/// which is fine: they hold no document lock, so flushing past them is safe — +/// the wait exists for the writers. +fn drain_connections(conn_count: &Arc) -> bool { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(SHUTDOWN_DRAIN_SECS); + while std::time::Instant::now() < deadline { + if conn_count.load(Ordering::Acquire) == 0 { + return true; + } + std::thread::sleep(std::time::Duration::from_millis(50)); + } + conn_count.load(Ordering::Acquire) == 0 +} + /// Longest gap between idle sweeps, however long the idle deadline is. const MAX_SWEEP_INTERVAL_SECS: u64 = 300; @@ -67,6 +88,11 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> { // gone. That is defensible for a demo and indefensible for a deployment, // and the two are indistinguishable from inside the process — so the // operator has to say which one this is. + // Fail closed on an unwritable data directory: the alternative is an + // eviction failing silently half an hour after start. + store.check_writable().map_err(|error| { + WebCanvasError::Config(format!("--online cannot use its data directory: {error}")) + })?; check_persistence_configured( store.is_enabled(), ephemeral_opt_in(), @@ -127,8 +153,8 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> { // process where it stands — losing every resident tenant that had not // happened to be evicted. The handler only raises the flag the accept // loop already observes, so the existing exit path (which flushes) runs. - install_shutdown_signals(&shutdown, local_addr); - spawn_sweeper(®istry, &shutdown); + install_shutdown_signals(&shutdown, local_addr)?; + spawn_sweeper(®istry, &shutdown)?; for stream in listener.incoming() { if shutdown.load(Ordering::Acquire) { @@ -176,7 +202,21 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> { } // The sweeper observes the same flag and retires on its next wake. shutdown.store(true, Ordering::Release); + // Let in-flight requests finish before writing. A push that is mid-install + // when the flush runs would otherwise be written in its pre-push state and + // the client's work lost — it acked 200 and then vanished. Bounded, because + // an SSE stream can hold a connection open indefinitely and a deploy + // cannot wait for it. + let drained = drain_connections(&conn_count); let flushed = registry.flush_all(); + if !drained { + eprintln!( + "openpencil --serve-web --online: {} connection(s) still active after {}s; \ + flushing anyway", + conn_count.load(Ordering::Acquire), + SHUTDOWN_DRAIN_SECS + ); + } eprintln!( "openpencil --serve-web --online: shutdown requested; flushed {flushed} account(s); \ exiting" @@ -221,7 +261,10 @@ static SIGNAL_RECEIVED: AtomicBool = AtomicBool::new(false); /// lifecycle they have always had — a token-authed shutdown request, or /// stdin EOF under a supervisor — and adding a handler there would change /// what Ctrl-C means for an interactive operator. -fn install_shutdown_signals(shutdown: &Arc, local_addr: std::net::SocketAddr) { +fn install_shutdown_signals( + shutdown: &Arc, + local_addr: std::net::SocketAddr, +) -> Result<()> { #[cfg(unix)] { // SAFETY: `handle_shutdown_signal` is async-signal-safe — it performs @@ -250,16 +293,18 @@ fn install_shutdown_signals(shutdown: &Arc, local_addr: std::net::So } std::thread::sleep(std::time::Duration::from_millis(200)); }); - if spawned.is_err() { - eprintln!( - "openpencil --serve-web --online: could not start the signal watcher; a \ - container stop will not flush tenants" - ); - } + // A container stop that does not flush is data loss on every deploy, + // so a watcher that cannot start is a start-up failure. + spawned.map(|_| ()).map_err(|error| { + WebCanvasError::Config(format!( + "could not start the shutdown signal watcher: {error}" + )) + }) } #[cfg(not(unix))] { let _ = (shutdown, local_addr); + Ok(()) } } @@ -284,7 +329,7 @@ fn ephemeral_opt_in() -> bool { /// it never ran in: idle accounts stayed resident indefinitely and were never /// written to disk. This thread observes `shutdown` on every wake, so it /// retires within one interval of the daemon being asked to stop. -fn spawn_sweeper(registry: &Arc, shutdown: &Arc) { +fn spawn_sweeper(registry: &Arc, shutdown: &Arc) -> Result<()> { let registry = Arc::clone(registry); let shutdown = Arc::clone(shutdown); let interval = sweep_interval_secs(registry.limits().idle_evict_secs); @@ -304,12 +349,11 @@ fn spawn_sweeper(registry: &Arc, shutdown: &Arc) { } } }); - if spawned.is_err() { - eprintln!( - "openpencil --serve-web --online: could not start the idle sweeper; accounts will \ - stay resident until a connection triggers a sweep" - ); - } + spawned.map(|_| ()).map_err(|error| { + // Without the sweeper nothing is ever evicted OR persisted, so a + // "degraded" daemon here quietly stops saving anyone's work. + WebCanvasError::Config(format!("could not start the idle sweeper: {error}")) + }) } /// How often to sweep, given the idle deadline. @@ -472,6 +516,27 @@ pub(super) fn serve_one_online( // here rather than in the document tier — a visitor holding a `?tenant=` // lease on someone else's document must not be able to re-share it. if super::share_routes::is_share_route(&req.path) { + // Same gate the connection tier applies, because this route is + // dispatched before it ever runs: a grant is a write. + if let Some(refusal) = super::tool_scopes::check_rest_scope( + identity.via, + identity.scopes, + &req.method, + &req.path, + ) { + crate::mcp_serve::write_mcp_http_response_with_origin( + stream, + refusal.http_status(), + &serde_json::json!({ + "ok": false, + "error": refusal.code(), + "message": refusal.to_string(), + }) + .to_string(), + cors_origin.as_deref(), + )?; + return Ok(false); + } let own = match registry.lease_for(&identity) { Ok(own) => own, Err(error) => { @@ -517,7 +582,7 @@ pub(super) fn serve_one_online( // The public tool profile, narrowed further by whatever scopes // this particular credential carries. mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::online(identity.scopes), - rest_identity: Some((identity.via, identity.scopes)), + rest_identity: Some(identity.clone()), }, ) } diff --git a/crates/op-host-services/src/web_canvas_server/online_run_loop_tests.rs b/crates/op-host-services/src/web_canvas_server/online_run_loop_tests.rs index 2758f7045..f43d0a139 100644 --- a/crates/op-host-services/src/web_canvas_server/online_run_loop_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/online_run_loop_tests.rs @@ -449,12 +449,17 @@ fn the_agent_indicator_relay_is_empty() { #[test] fn the_device_login_proxy_is_not_routed() { + // `/api/auth/status` is deliberately EXCLUDED: it is the read-only account + // projection the shell needs to detect an account switch, and it answers + // from the connection's verified identity rather than from the daemon's + // process-wide device session. Everything that drives that session stays + // unreachable. See `the_sign_in_and_sign_out_routes_stay_unreachable_online`. let registry = registry(); let verifier = verifier(); for request in [ - Request::new("GET", op_editor_core::auth_routes::STATUS).with_bearer("tokA"), Request::json("POST", op_editor_core::auth_routes::LOGOUT, "{}").with_bearer("tokA"), Request::json("POST", op_editor_core::auth_routes::LOGIN_BEGIN, "{}").with_bearer("tokA"), + Request::new("GET", op_editor_core::auth_routes::LOGIN_STATUS).with_bearer("tokA"), ] { let path = request.path; let response = serve(®istry, &verifier, request); diff --git a/crates/op-host-services/src/web_canvas_server/online_share_tests.rs b/crates/op-host-services/src/web_canvas_server/online_share_tests.rs index e613caf0b..80e4703e1 100644 --- a/crates/op-host-services/src/web_canvas_server/online_share_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/online_share_tests.rs @@ -661,3 +661,114 @@ fn a_share_that_cannot_be_persisted_is_reported_and_rolled_back() { let _ = std::fs::remove_file(&temp.root); } + +// --------------------------------------------------------------------------- +// B2: the shell must be able to learn which account it is showing. +// --------------------------------------------------------------------------- + +#[test] +fn the_online_status_route_projects_the_verified_identity() { + // Without this the shell's identity epoch never fires, and an account + // switch in one tab leaks the previous account's document. + let response = serve( + ®istry(), + &verifier(), + Request::new("GET", op_editor_core::auth_routes::STATUS).with_bearer("tokA"), + ); + assert_eq!(status_line(&response), "HTTP/1.1 200 OK", "{response}"); + let payload = body(&response); + assert_eq!(payload["signed_in"], true); + assert_eq!(payload["available"], true); + assert_eq!(payload["username"], "userA"); +} + +#[test] +fn two_accounts_see_their_own_identity_on_the_status_route() { + let registry = registry(); + let verifier = verifier(); + for (token, account) in [("tokA", "userA"), ("tokB", "userB")] { + let response = serve( + ®istry, + &verifier, + Request::new("GET", op_editor_core::auth_routes::STATUS).with_bearer(token), + ); + assert_eq!(body(&response)["username"], account, "{token}"); + } +} + +#[test] +fn the_sign_in_and_sign_out_routes_stay_unreachable_online() { + // Only the read-only projection is exposed: the rest drive the + // process-wide device session an online deployment must never share. + let registry = registry(); + let verifier = verifier(); + for request in [ + Request::json("POST", op_editor_core::auth_routes::LOGIN_BEGIN, "{}").with_bearer("tokA"), + Request::json("POST", op_editor_core::auth_routes::LOGOUT, "{}").with_bearer("tokA"), + Request::new("GET", op_editor_core::auth_routes::LOGIN_STATUS).with_bearer("tokA"), + ] { + let path = request.path; + let response = serve(®istry, &verifier, request); + assert_eq!( + status_line(&response), + "HTTP/1.1 404 Not Found", + "{path}: {response}" + ); + } +} + +#[test] +fn an_unauthenticated_caller_cannot_reach_the_status_projection() { + let response = serve( + ®istry(), + &verifier(), + Request::new("GET", op_editor_core::auth_routes::STATUS), + ); + assert_eq!( + status_line(&response), + "HTTP/1.1 401 Unauthorized", + "{response}" + ); +} + +// --------------------------------------------------------------------------- +// B3 residue: start-up probes and the shutdown drain. +// --------------------------------------------------------------------------- + +#[test] +fn an_unwritable_data_directory_is_refused_at_start_up() { + // The alternative is an eviction failing silently half an hour after + // start, one account at a time. + let root = std::env::temp_dir().join(format!("op-unwritable-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&root); + std::fs::write(&root, b"not a directory").expect("block the root"); + + let store = TenantStore::new(Some(root.clone())); + let error = store.check_writable().expect_err("must refuse"); + let message = error.to_string(); + assert!( + message.contains("10001"), + "must name the likely cause: {message}" + ); + + let _ = std::fs::remove_file(&root); +} + +#[test] +fn a_writable_data_directory_passes_the_probe() { + let temp = PersistentRegistry::new("probe-ok"); + temp.registry.store().check_writable().expect("writable"); + // The probe cleans up after itself. + assert!( + !temp.root.join(".probe").join("write.ok").exists(), + "the probe must not leave files behind" + ); +} + +#[test] +fn a_disabled_store_passes_the_probe_trivially() { + // Nothing to write, so nothing to prove. + TenantStore::new(None) + .check_writable() + .expect("no store, no probe"); +} diff --git a/crates/op-host-services/src/web_canvas_server/share_routes.rs b/crates/op-host-services/src/web_canvas_server/share_routes.rs index b42ebfa02..d4a1b1a8c 100644 --- a/crates/op-host-services/src/web_canvas_server/share_routes.rs +++ b/crates/op-host-services/src/web_canvas_server/share_routes.rs @@ -126,6 +126,19 @@ fn mutate( }) .to_string(), }, + // A full access list is the caller's problem, not the server's: the + // store writes a bounded list, so accepting the grant would report a + // success that vanishes on the next save. + Err(super::tenant_store::TenantStoreError::ShareLimitReached(limit)) => WebReply { + status: "400 Bad Request", + body: serde_json::json!({ + "ok": false, + "error": "share-limit-reached", + "limit": limit, + "message": format!("this document is already shared with {limit} accounts"), + }) + .to_string(), + }, // The change has been rolled back, so memory and disk agree and a // retry starts from a known state. Reporting 200 here — as the // previous code did — told the user a share had succeeded that would diff --git a/crates/op-host-services/src/web_canvas_server/share_routes_tests.rs b/crates/op-host-services/src/web_canvas_server/share_routes_tests.rs index 840ffcf77..71e76e3b4 100644 --- a/crates/op-host-services/src/web_canvas_server/share_routes_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/share_routes_tests.rs @@ -245,3 +245,63 @@ fn a_forbidden_share_and_an_unknown_one_answer_identically() { let forbidden = registry.lease_for_shared("userA", &stranger).unwrap_err(); assert_eq!(unknown, forbidden); } + +#[test] +fn the_grant_past_the_ceiling_is_refused_rather_than_silently_dropped() { + // The store writes a bounded list, so accepting the 257th grant would + // report a success that vanishes on the next save. + let registry = registry(); + let owner = identity("userA"); + let lease = registry.lease_for(&owner).expect("lease"); + for index in 0..super::super::tenant_store::MAX_SHARED_ACCOUNTS { + let reply = handle( + "POST", + share_routes::GRANT, + &serde_json::json!({ "userId": format!("guest-{index}") }).to_string(), + &owner, + &lease, + ®istry, + ); + assert_eq!(reply.status, "200 OK", "grant {index}"); + } + let overflow = handle( + "POST", + share_routes::GRANT, + r#"{"userId":"one-too-many"}"#, + &owner, + &lease, + ®istry, + ); + assert_eq!(overflow.status, "400 Bad Request", "{}", overflow.body); + assert_eq!(body_of(&overflow)["error"], "share-limit-reached"); + // And the refused account really is absent, not quietly present in memory. + assert!(!lease.tenant().shared_with().contains("one-too-many")); +} + +#[test] +fn a_repeat_grant_at_the_ceiling_still_succeeds() { + // The ceiling bounds NEW accounts; re-granting one already on the list + // changes nothing and must not be refused. + let registry = registry(); + let owner = identity("userA"); + let lease = registry.lease_for(&owner).expect("lease"); + for index in 0..super::super::tenant_store::MAX_SHARED_ACCOUNTS { + handle( + "POST", + share_routes::GRANT, + &serde_json::json!({ "userId": format!("guest-{index}") }).to_string(), + &owner, + &lease, + ®istry, + ); + } + let repeat = handle( + "POST", + share_routes::GRANT, + r#"{"userId":"guest-0"}"#, + &owner, + &lease, + ®istry, + ); + assert_eq!(repeat.status, "200 OK", "{}", repeat.body); +} diff --git a/crates/op-host-services/src/web_canvas_server/sse_hub.rs b/crates/op-host-services/src/web_canvas_server/sse_hub.rs index ec63500d5..61ec205ef 100644 --- a/crates/op-host-services/src/web_canvas_server/sse_hub.rs +++ b/crates/op-host-services/src/web_canvas_server/sse_hub.rs @@ -78,10 +78,12 @@ impl SseHub { /// slot for ticks. Dropping it unregisters. pub(crate) fn subscribe(&self) -> Arc { let slot = Arc::new(SseSlot::new()); - self.subscribers - .lock() - .unwrap_or_else(|p| p.into_inner()) - .push(Arc::downgrade(&slot)); + let mut subscribers = self.subscribers.lock().unwrap_or_else(|p| p.into_inner()); + // Prune here too, not only on broadcast: a tenant whose clients all + // disconnected and which then never publishes again would otherwise + // accumulate one dead `Weak` per reconnect, forever. + subscribers.retain(|slot| slot.strong_count() > 0); + subscribers.push(Arc::downgrade(&slot)); slot } diff --git a/crates/op-host-services/src/web_canvas_server/tenant.rs b/crates/op-host-services/src/web_canvas_server/tenant.rs index 9566ee745..089d53ff2 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant.rs @@ -534,7 +534,11 @@ impl TenantRegistry { let mut written = 0; for (id, tenant) in tenants.iter() { let guard = tenant.state.lock().unwrap_or_else(|p| p.into_inner()); - match self.store.save(id, &guard.editor, &tenant.shared_with()) { + // Held across the write, exactly as `update_acl` does: otherwise a + // grant landing mid-flush is written by one path and overwritten + // by the other, and the user's share silently disappears. + let shared = tenant.shared_with_guard(); + match self.store.save(id, &guard.editor, &shared) { Ok(()) => written += 1, Err(error) => eprintln!( "openpencil --serve-web --online: could not flush a tenant on shutdown \ @@ -566,6 +570,18 @@ impl TenantRegistry { change: AclChange, ) -> Result { let mut list = tenant.shared_with_guard(); + if let AclChange::Grant(account) = &change { + // The store writes at most `MAX_SHARED_ACCOUNTS`, so accepting a + // grant past the ceiling would report success for a share that + // silently vanishes on the next save. Refuse it instead. + if !list.contains(account.as_str()) + && list.len() >= super::tenant_store::MAX_SHARED_ACCOUNTS + { + return Err(TenantStoreError::ShareLimitReached( + super::tenant_store::MAX_SHARED_ACCOUNTS, + )); + } + } let changed = match &change { AclChange::Grant(account) => list.insert(account.clone()), AclChange::Revoke(account) => list.remove(account.as_str()), @@ -616,7 +632,8 @@ impl TenantRegistry { // registry lock is held, so it waits) or, afterwards, the file. if self.store.is_enabled() { let guard = victim.state.lock().unwrap_or_else(|p| p.into_inner()); - if let Err(error) = self.store.save(&id, &guard.editor, &victim.shared_with()) { + let shared = victim.shared_with_guard(); + if let Err(error) = self.store.save(&id, &guard.editor, &shared) { // A tenant that cannot be written is kept resident. Evicting // it anyway would discard the document to reclaim memory, // which is the wrong trade for the user whose work it is. diff --git a/crates/op-host-services/src/web_canvas_server/tenant_auth.rs b/crates/op-host-services/src/web_canvas_server/tenant_auth.rs index a597c979d..dbc518530 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_auth.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_auth.rs @@ -137,6 +137,36 @@ impl PresentedCredentials { } } +impl ResolvedIdentity { + /// The `/api/auth/status` projection for an online deployment. + /// + /// The shell's account layer polls this route to learn WHO it is showing; + /// online used to 404 it wholesale, which left the identity epoch — and + /// therefore the account-switch reset — permanently dormant. So the route + /// answers here instead of through the device-login proxy. + /// + /// Only a verified connection reaches this: the online loop resolves the + /// credential before dispatch, so there is no anonymous caller to leak to. + /// It is a strictly READ-ONLY projection of the caller's own identity — + /// the sign-in and sign-out routes stay 404, because they drive the + /// process-wide device session that an online deployment must not expose. + pub fn auth_status_json(&self) -> String { + serde_json::json!({ + // The account UI is meaningful: this deployment has accounts. + "available": true, + "signed_in": true, + // `username` is what `identity_epoch` keys the tab's partition on. + "username": self.username, + "display_name": self.display_name, + // Not projected: the hub owns the address, and the shell only uses + // it for display. Absent is honest rather than fabricated. + "primary_email": serde_json::Value::Null, + "avatar_revision": serde_json::Value::Null, + }) + .to_string() + } +} + /// Turns a presented credential into a verified account. /// /// Implementations must be safe to call from many connection threads at diff --git a/crates/op-host-services/src/web_canvas_server/tenant_store.rs b/crates/op-host-services/src/web_canvas_server/tenant_store.rs index 7fe4f893f..8274bd39b 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_store.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_store.rs @@ -48,7 +48,7 @@ const CORRUPT_SUFFIX: &str = "corrupt"; /// /// A share list is a handful of accounts; a larger one is a bug or an attempt /// to make the daemon allocate on every eviction. -const MAX_SHARED_ACCOUNTS: usize = 256; +pub(super) const MAX_SHARED_ACCOUNTS: usize = 256; /// Why a tenant could not be persisted or restored. /// @@ -66,6 +66,10 @@ pub enum TenantStoreError { /// The stored document exists but could not be loaded. It has been moved /// aside; the caller should start fresh. Unreadable(String), + /// The access list is already at its ceiling. A client fault, not an IO + /// one — reported so the grant is refused rather than accepted and then + /// silently dropped by the bounded write. + ShareLimitReached(usize), } impl std::fmt::Display for TenantStoreError { @@ -77,6 +81,9 @@ impl std::fmt::Display for TenantStoreError { Self::Unreadable(detail) => { write!(f, "stored document could not be loaded: {detail}") } + Self::ShareLimitReached(limit) => { + write!(f, "this document is already shared with {limit} accounts") + } } } } @@ -109,6 +116,38 @@ impl TenantStore { self.root.is_some() } + /// Prove at start-up that this process can actually write the data + /// directory. + /// + /// A read-only mount, or a volume owned by root while the container runs + /// as UID 10001, fails every eviction — silently, half an hour after + /// start, one account at a time. Exercising the full create → write → + /// rename → delete cycle here turns that into a start-up failure that + /// names the likely cause. + pub fn check_writable(&self) -> Result<(), TenantStoreError> { + let Some(root) = self.root.as_ref() else { + return Ok(()); + }; + let probe_dir = root.join(".probe"); + let describe = |error: std::io::Error| { + TenantStoreError::Io(format!( + "{error} (is {} writable by the user this process runs as? \ + the container image runs as UID 10001)", + root.display() + )) + }; + std::fs::create_dir_all(&probe_dir).map_err(describe)?; + let staged = probe_dir.join("write.tmp"); + let landed = probe_dir.join("write.ok"); + std::fs::write(&staged, b"probe").map_err(describe)?; + // Rename specifically: the atomic publish every save depends on can + // fail where a plain write succeeds (some network filesystems). + std::fs::rename(&staged, &landed).map_err(describe)?; + std::fs::remove_file(&landed).map_err(describe)?; + let _ = std::fs::remove_dir(&probe_dir); + Ok(()) + } + /// The directory holding `user_id`'s tenant, if persistence is on. /// /// See the module docs for why this is a hash and not the id. diff --git a/crates/op-host-services/src/web_canvas_server/tenant_store_tests.rs b/crates/op-host-services/src/web_canvas_server/tenant_store_tests.rs index 7356012b1..a4027fed1 100644 --- a/crates/op-host-services/src/web_canvas_server/tenant_store_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/tenant_store_tests.rs @@ -313,3 +313,50 @@ fn a_persisted_tenant_carries_no_thumbnail_data() { "another tenant's image data must not ride along" ); } + +#[test] +fn a_push_that_never_installs_releases_its_thumbnail_seed() { + // `load_canonical` registers the seed in a process-global side table keyed + // by the document; only an activation consumes it. A push that is parsed + // and then dropped — stale baseVersion, collaboration refusal, client gone + // — used to leave it there, where a later document reusing the key could + // activate someone else's thumbnails. + use crate::web_canvas_server::{PendingDocumentPush, ServeMode}; + + let body = serde_json::json!({ + "document": { + "version": "1.0.0", + "children": [{ + "id": "n1", "type": "rectangle", "name": "seeded", + "x": 0, "y": 0, "width": 4, "height": 4, + }], + "imageThumbs": { "5150": "AQID" }, + }, + "sourceClientId": "s", + }) + .to_string(); + + let push = PendingDocumentPush::parse(&body, ServeMode::Local).expect("parses"); + let document_version = push + .prepared + .as_ref() + .expect("a document push") + .document() + .version + .clone(); + // Dropped without installing — the failure paths this exists for. + drop(push); + + // The seed is gone: a fresh document carrying the same key activates + // nothing rather than inheriting the abandoned push's thumbnails. + let mut probe = op_pen_loader::load_canonical( + &serde_json::json!({ "version": document_version, "children": [] }).to_string(), + ) + .expect("probe document") + .value; + probe.version.clone_from(&document_version); + assert!( + !jian_ops_schema::image_thumbs::activate_for_document(&probe), + "an abandoned push must not leave an activatable seed behind" + ); +} diff --git a/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs b/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs index 24b371d1c..e1e302612 100644 --- a/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs +++ b/crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs @@ -47,6 +47,10 @@ fn a_token_with_no_scopes_is_refused_everything_but_the_health_probe() { assert!(refused(NONE, "POST", "/api/mcp/document")); // …except the probe, so a client can discover the daemon and be told why. assert!(!refused(NONE, "GET", "/api/mcp/server")); + // …and except JSON-RPC, where the MCP dispatch enforces per-tool scopes + // and would otherwise be refused wholesale by the coarse method rule. + assert!(!refused(NONE, "POST", "/mcp")); + assert!(!refused(READ_ONLY, "POST", "/mcp")); } #[test] diff --git a/crates/op-host-services/src/web_canvas_server_conn_tests.rs b/crates/op-host-services/src/web_canvas_server_conn_tests.rs index 891132981..556bd5c54 100644 --- a/crates/op-host-services/src/web_canvas_server_conn_tests.rs +++ b/crates/op-host-services/src/web_canvas_server_conn_tests.rs @@ -82,117 +82,6 @@ fn tick(version: u64, collab_seq: u64) -> SseTick { } } -#[test] -fn sse_hub_broadcasts_version_to_all_subscribers() { - let hub = SseHub::default(); - let a = hub.subscribe(); - let b = hub.subscribe(); - hub.broadcast(tick(5, 0)); - assert_eq!(a.pending().expect("published"), tick(5, 0)); - assert_eq!(b.pending().expect("published"), tick(5, 0)); -} - -#[test] -fn sse_hub_prunes_disconnected_subscribers() { - let hub = SseHub::default(); - let live = hub.subscribe(); - drop(hub.subscribe()); // a disconnected client (receiver dropped) - assert_eq!(hub.subscriber_count(), 2); - hub.broadcast(tick(1, 0)); // prunes the dropped one - assert_eq!(hub.subscriber_count(), 1); - assert_eq!(live.pending().expect("published"), tick(1, 0)); -} - -#[test] -fn write_sse_event_emits_data_frame() { - let mut stream = mock_stream(""); - write_sse_event(&mut stream, tick(42, 7)).expect("write"); - assert_eq!( - String::from_utf8_lossy(&stream.output), - "data: {\"version\":42,\"collabSeq\":7}\n\n" - ); -} - -#[test] -fn sse_payload_stays_a_superset_of_the_original_version_frame() { - let mut stream = mock_stream(""); - write_sse_event(&mut stream, tick(3, 0)).expect("write"); - let out = String::from_utf8_lossy(&stream.output).into_owned(); - // A client written against the original `{"version":N}` frame parses this - // one unchanged: `version` keeps its spelling and stays the first field. - assert!(out.starts_with("data: {\"version\":3,"), "{out}"); - let payload: serde_json::Value = - serde_json::from_str(out.trim_start_matches("data: ").trim()).expect("valid JSON"); - assert_eq!(payload["version"], 3); - assert_eq!(payload["collabSeq"], 0); -} - -#[test] -fn serve_sse_emits_the_initial_tick_then_each_published_one() { - // The stream ends when a socket write fails, which is how a disconnected - // client is detected — so this writes to a stream that fails on the - // second event. - let hub = SseHub::default(); - let slot = hub.subscribe(); - hub.broadcast(tick(9, 0)); - let mut stream = FailingStream { - written: Vec::new(), - writes_before_failure: 3, - }; - let _ = serve_sse(&mut stream, &slot, tick(7, 0), Some("*")); - let out = String::from_utf8_lossy(&stream.written); - assert!(out.contains("text/event-stream"), "{out}"); - assert!(out.contains(r#"data: {"version":7,"#), "{out}"); // initial sync - assert!(out.contains(r#"data: {"version":9,"#), "{out}"); // published bump -} - -/// A stream that stops accepting writes, standing in for a client that went -/// away — which is the only thing that ends an SSE loop. -struct FailingStream { - written: Vec, - writes_before_failure: usize, -} - -impl std::io::Write for FailingStream { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - if self.writes_before_failure == 0 { - return Err(std::io::Error::other("client went away")); - } - self.writes_before_failure -= 1; - self.written.extend_from_slice(buf); - Ok(buf.len()) - } - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - -#[test] -fn a_subscriber_that_never_reads_keeps_exactly_one_pending_tick() { - // The unbounded-queue bug: a paused tab accumulated one entry per - // mutation, in a process shared with every other account. - let hub = SseHub::default(); - let slot = hub.subscribe(); - for version in 0..1000 { - hub.broadcast(tick(version, 0)); - } - // Only the newest survives — an older tick is not information the client - // lost, it is information the newest one already contains. - assert_eq!(slot.pending(), Some(tick(999, 0))); - assert_eq!(slot.pending(), None, "taking it leaves the slot empty"); -} - -#[test] -fn a_dropped_subscriber_is_pruned_without_signalling_anything() { - let hub = SseHub::default(); - let live = hub.subscribe(); - drop(hub.subscribe()); - assert_eq!(hub.subscriber_count(), 2); - hub.broadcast(tick(1, 0)); - assert_eq!(hub.subscriber_count(), 1); - assert_eq!(live.pending(), Some(tick(1, 0))); -} - #[test] fn serve_one_post_document_broadcasts_new_version_to_sse() { let state = Mutex::new(fresh_state()); @@ -776,3 +665,6 @@ fn cors_echoes_only_allowlisted_origin() { assert_eq!(cors_origin_for(&allow, Some("http://evil.local")), None); assert_eq!(cors_origin_for(&allow, None), None); } + +#[path = "web_canvas_server_sse_tests.rs"] +mod sse_tests; diff --git a/crates/op-host-services/src/web_canvas_server_error.rs b/crates/op-host-services/src/web_canvas_server_error.rs index 74c2389c3..19c4b8b0c 100644 --- a/crates/op-host-services/src/web_canvas_server_error.rs +++ b/crates/op-host-services/src/web_canvas_server_error.rs @@ -63,7 +63,7 @@ pub enum WebCanvasError { /// declined to commit it. Distinct from `Collab`: the write was not /// refused up front, it was discarded, so the browser's copy is now /// definitively behind and must be refetched rather than retried. - IngestRejected(crate::web_canvas_server::IngestOutcome), + IngestRejected(crate::web_canvas_server::IngestOutcome, u64), } impl WebCanvasError { @@ -83,7 +83,7 @@ impl WebCanvasError { | WebCanvasError::Io(_) => "400 Bad Request", WebCanvasError::Config(_) | WebCanvasError::Transport(_) => "500 Internal Server Error", WebCanvasError::Collab(refusal) => refusal.http_status(), - WebCanvasError::IngestRejected(_) => "409 Conflict", + WebCanvasError::IngestRejected(..) => "409 Conflict", } } @@ -95,7 +95,7 @@ impl WebCanvasError { pub fn error_code(&self) -> Option<&'static str> { match self { WebCanvasError::Collab(refusal) => Some(refusal.code()), - WebCanvasError::IngestRejected(outcome) => outcome.error_code(), + WebCanvasError::IngestRejected(outcome, _) => outcome.error_code(), _ => None, } } @@ -111,7 +111,7 @@ impl fmt::Display for WebCanvasError { | WebCanvasError::Config(m) | WebCanvasError::Transport(m) => f.write_str(m), WebCanvasError::Collab(refusal) => write!(f, "{refusal}"), - WebCanvasError::IngestRejected(_) => { + WebCanvasError::IngestRejected(..) => { f.write_str("the live session discarded this document push; refetch and reapply") } } diff --git a/crates/op-host-services/src/web_canvas_server_sse_tests.rs b/crates/op-host-services/src/web_canvas_server_sse_tests.rs new file mode 100644 index 000000000..eed0bc060 --- /dev/null +++ b/crates/op-host-services/src/web_canvas_server_sse_tests.rs @@ -0,0 +1,141 @@ +//! SSE fan-out tests — the hub's latest-value slots and the stream writer. +//! +//! Split out of `web_canvas_server_conn_tests.rs` at the 800-line cap; nested +//! under it so `use super::*` still reaches the mock stream and helpers. + +use super::*; + +#[test] +fn sse_hub_broadcasts_version_to_all_subscribers() { + let hub = SseHub::default(); + let a = hub.subscribe(); + let b = hub.subscribe(); + hub.broadcast(tick(5, 0)); + assert_eq!(a.pending().expect("published"), tick(5, 0)); + assert_eq!(b.pending().expect("published"), tick(5, 0)); +} + +#[test] +fn sse_hub_prunes_disconnected_subscribers() { + let hub = SseHub::default(); + let live = hub.subscribe(); + drop(hub.subscribe()); // a disconnected client (receiver dropped) + assert_eq!(hub.subscriber_count(), 2); + hub.broadcast(tick(1, 0)); // prunes the dropped one + assert_eq!(hub.subscriber_count(), 1); + assert_eq!(live.pending().expect("published"), tick(1, 0)); +} + +#[test] +fn write_sse_event_emits_data_frame() { + let mut stream = mock_stream(""); + write_sse_event(&mut stream, tick(42, 7)).expect("write"); + assert_eq!( + String::from_utf8_lossy(&stream.output), + "data: {\"version\":42,\"collabSeq\":7}\n\n" + ); +} + +#[test] +fn sse_payload_stays_a_superset_of_the_original_version_frame() { + let mut stream = mock_stream(""); + write_sse_event(&mut stream, tick(3, 0)).expect("write"); + let out = String::from_utf8_lossy(&stream.output).into_owned(); + // A client written against the original `{"version":N}` frame parses this + // one unchanged: `version` keeps its spelling and stays the first field. + assert!(out.starts_with("data: {\"version\":3,"), "{out}"); + let payload: serde_json::Value = + serde_json::from_str(out.trim_start_matches("data: ").trim()).expect("valid JSON"); + assert_eq!(payload["version"], 3); + assert_eq!(payload["collabSeq"], 0); +} + +#[test] +fn serve_sse_emits_the_initial_tick_then_each_published_one() { + // The stream ends when a socket write fails, which is how a disconnected + // client is detected — so this writes to a stream that fails on the + // second event. + let hub = SseHub::default(); + let slot = hub.subscribe(); + hub.broadcast(tick(9, 0)); + let mut stream = FailingStream { + written: Vec::new(), + writes_before_failure: 3, + }; + let _ = serve_sse(&mut stream, &slot, tick(7, 0), Some("*")); + let out = String::from_utf8_lossy(&stream.written); + assert!(out.contains("text/event-stream"), "{out}"); + assert!(out.contains(r#"data: {"version":7,"#), "{out}"); // initial sync + assert!(out.contains(r#"data: {"version":9,"#), "{out}"); // published bump +} + +/// A stream that stops accepting writes, standing in for a client that went +/// away — which is the only thing that ends an SSE loop. +struct FailingStream { + written: Vec, + writes_before_failure: usize, +} + +impl std::io::Write for FailingStream { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + if self.writes_before_failure == 0 { + return Err(std::io::Error::other("client went away")); + } + self.writes_before_failure -= 1; + self.written.extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +#[test] +fn a_subscriber_that_never_reads_keeps_exactly_one_pending_tick() { + // The unbounded-queue bug: a paused tab accumulated one entry per + // mutation, in a process shared with every other account. + let hub = SseHub::default(); + let slot = hub.subscribe(); + for version in 0..1000 { + hub.broadcast(tick(version, 0)); + } + // Only the newest survives — an older tick is not information the client + // lost, it is information the newest one already contains. + assert_eq!(slot.pending(), Some(tick(999, 0))); + assert_eq!(slot.pending(), None, "taking it leaves the slot empty"); +} + +#[test] +fn a_dropped_subscriber_is_pruned_without_signalling_anything() { + let hub = SseHub::default(); + let live = hub.subscribe(); + drop(hub.subscribe()); + assert_eq!(hub.subscriber_count(), 2); + hub.broadcast(tick(1, 0)); + assert_eq!(hub.subscriber_count(), 1); + assert_eq!(live.pending(), Some(tick(1, 0))); +} + +#[test] +fn subscribing_prunes_dead_slots_even_when_nothing_is_ever_broadcast() { + // A tenant whose clients all disconnected and which never publishes again + // would otherwise accumulate one dead `Weak` per reconnect, forever. + let hub = SseHub::default(); + for _ in 0..100 { + drop(hub.subscribe()); + } + // Each subscribe prunes the previous corpse, so at most the live one plus + // the one just added remain. + assert!( + hub.subscriber_count() <= 2, + "dead subscribers accumulated: {}", + hub.subscriber_count() + ); +} + +#[test] +fn broadcasting_to_no_subscribers_is_a_no_op() { + let hub = SseHub::default(); + hub.broadcast(tick(1, 0)); + assert_eq!(hub.subscriber_count(), 0); +} diff --git a/crates/op-host-web/src/collab_sync.rs b/crates/op-host-web/src/collab_sync.rs index 2ce2fb5d7..d5e53c1d9 100644 --- a/crates/op-host-web/src/collab_sync.rs +++ b/crates/op-host-web/src/collab_sync.rs @@ -348,6 +348,24 @@ fn sync_id_allocation(host: &mut crate::widget_host::WidgetHost, wire: &CollabSt .then(|| wire.session.as_ref().and_then(|s| s.peer_namespace.clone())) .flatten(); match namespace { + // The installed allocator disagrees with the namespace the daemon now + // publishes — a different session (or a different account) owns this + // document. Tear the old allocator down so the arm below installs the + // new one; leaving it would keep minting ids in a namespace this peer + // no longer holds. + Some(namespace) + if host.collaboration_ids_enabled() + && !session_namespace_matches(namespace.as_str()) => + { + host.disable_collaboration_ids(); + set_session_namespace(None); + if let Ok(parsed) = op_editor_core::PeerNamespace::parse(namespace.clone()) { + let installed = parsed.clone(); + if host.enable_collaboration_ids(parsed).is_ok() { + set_session_namespace(Some(installed)); + } + } + } Some(namespace) if !host.collaboration_ids_enabled() => { match op_editor_core::PeerNamespace::parse(namespace) { Ok(namespace) => { @@ -391,6 +409,30 @@ pub(crate) fn reset_for_new_identity() { SESSION_NAMESPACE.with(|slot| *slot.borrow_mut() = None); APPLIED_SEQ.set(None); SESSION_LIVE.set(false); + ACTION_RETRY.with(|slot| *slot.borrow_mut() = None); + ACTION_BUSY.set(false); + UNDO_REQUESTED.set(false); +} + +/// Drop the id allocator the previous account's session installed. +/// +/// Separate from [`reset_for_new_identity`] because it needs the host, which +/// the caller holds. Without it a new account keeps minting ids inside the +/// previous account's namespace — ids the new session never granted it. +pub(crate) fn reset_id_allocation(host: &mut crate::widget_host::WidgetHost) { + if host.collaboration_ids_enabled() { + host.disable_collaboration_ids(); + } + set_session_namespace(None); +} + +/// Whether the installed allocator is minting under `namespace`. +fn session_namespace_matches(namespace: &str) -> bool { + SESSION_NAMESPACE.with(|slot| { + slot.borrow() + .as_ref() + .is_some_and(|installed| installed.as_str() == namespace) + }) } /// Record (or clear) the namespace this peer mints under. diff --git a/crates/op-host-web/src/identity_epoch.rs b/crates/op-host-web/src/identity_epoch.rs index 98a1b582a..d284e798c 100644 --- a/crates/op-host-web/src/identity_epoch.rs +++ b/crates/op-host-web/src/identity_epoch.rs @@ -56,12 +56,34 @@ pub fn epoch() -> u64 { EPOCH.with(std::cell::Cell::get) } +/// What an observation means for the tab. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum IdentityObservation { + /// Same account as before. Nothing to do. + Unchanged, + /// The tab's FIRST answer, and it named an account. Nothing account-scoped + /// has been shown, so the document stands — but the storage partition just + /// moved off `anon`, and anything already loaded under `anon` has to be + /// reloaded from the account's own partition. + FirstIdentified, + /// A different account. Everything keyed to the previous one must go. + Changed, +} + +impl IdentityObservation { + /// Whether account-scoped storage must be re-read. + pub const fn requires_storage_reload(self) -> bool { + matches!(self, Self::FirstIdentified | Self::Changed) + } + + /// Whether the tab's document and sync state must be dropped. + pub const fn requires_reset(self) -> bool { + matches!(self, Self::Changed) + } +} + /// Record the subject an `/api/auth/status` answer reported. -/// -/// Returns `true` when the identity CHANGED and the caller must therefore -/// drop everything keyed to the previous account. The first observation is -/// not a change: the tab has shown nothing yet. -pub fn observe_subject(subject: Option<&str>) -> bool { +pub fn observe_subject(subject: Option<&str>) -> IdentityObservation { let next = subject .map(str::trim) .filter(|value| !value.is_empty()) @@ -71,20 +93,26 @@ pub fn observe_subject(subject: Option<&str>) -> bool { let previous = slot.replace(next.clone()); match previous { // The tab's first status answer. Whatever it says is what this - // tab has always been showing, so there is nothing to discard — - // but the storage partition is now known, so the epoch moves. + // tab has always been showing, so the document stands — but the + // shell already loaded settings and credentials under `anon` at + // mount, and those belong to a different partition than the one + // now in force. They have to be re-read. None => { EPOCH.with(|epoch| epoch.set(epoch.get().saturating_add(1))); - false + if next.is_some() { + IdentityObservation::FirstIdentified + } else { + IdentityObservation::Unchanged + } } // A repeat of the same answer: the common case, and it must not // churn state or every poll would reset the tab. - Some(before) if before == next => false, + Some(before) if before == next => IdentityObservation::Unchanged, // A genuine change — sign-in, sign-out, or a switch between two // accounts. All three must drop the previous account's state. Some(_) => { EPOCH.with(|epoch| epoch.set(epoch.get().saturating_add(1))); - true + IdentityObservation::Changed } } }) @@ -120,15 +148,19 @@ mod tests { #[test] fn the_first_anonymous_observation_is_not_a_change() { reset_for_test(); - assert!(!observe_subject(None)); + assert_eq!(observe_subject(None), IdentityObservation::Unchanged); assert_eq!(current_subject(), ANONYMOUS_SUBJECT); } #[test] fn the_first_answer_of_a_tab_is_never_a_reset() { reset_for_test(); - // Whatever the first answer says is what this tab has always shown. - assert!(!observe_subject(Some("alice"))); + // Whatever the first answer says is what this tab has always shown, + // so the document stands — but the storage partition moved off `anon`. + let outcome = observe_subject(Some("alice")); + assert_eq!(outcome, IdentityObservation::FirstIdentified); + assert!(outcome.requires_storage_reload()); + assert!(!outcome.requires_reset()); assert_eq!(current_subject(), "alice"); assert!(epoch() > 0); } @@ -138,8 +170,8 @@ mod tests { // Distinct from the case above: the tab HAS shown the anonymous // state, so signing in replaces what was on screen. reset_for_test(); - assert!(!observe_subject(None)); - assert!(observe_subject(Some("alice"))); + assert_eq!(observe_subject(None), IdentityObservation::Unchanged); + assert_eq!(observe_subject(Some("alice")), IdentityObservation::Changed); assert_eq!(current_subject(), "alice"); } @@ -148,8 +180,9 @@ mod tests { reset_for_test(); observe_subject(Some("alice")); let before = epoch(); - assert!( + assert_eq!( observe_subject(Some("bob")), + IdentityObservation::Changed, "a different account must reset the tab" ); assert_eq!(current_subject(), "bob"); @@ -160,7 +193,7 @@ mod tests { fn signing_out_reports_a_change() { reset_for_test(); observe_subject(Some("alice")); - assert!(observe_subject(None)); + assert_eq!(observe_subject(None), IdentityObservation::Changed); assert_eq!(current_subject(), ANONYMOUS_SUBJECT); } @@ -169,8 +202,8 @@ mod tests { // The leak this exists for: A → anonymous → B in one tab. reset_for_test(); observe_subject(Some("alice")); - assert!(observe_subject(None)); - assert!(observe_subject(Some("bob"))); + assert_eq!(observe_subject(None), IdentityObservation::Changed); + assert_eq!(observe_subject(Some("bob")), IdentityObservation::Changed); assert_eq!(current_subject(), "bob"); } @@ -180,7 +213,10 @@ mod tests { observe_subject(Some("alice")); let epoch_after_sign_in = epoch(); for _ in 0..5 { - assert!(!observe_subject(Some("alice"))); + assert_eq!( + observe_subject(Some("alice")), + IdentityObservation::Unchanged + ); } assert_eq!(epoch(), epoch_after_sign_in, "a poll must not churn state"); } @@ -210,7 +246,10 @@ mod tests { fn whitespace_around_a_subject_does_not_create_a_second_partition() { reset_for_test(); observe_subject(Some("alice")); - assert!(!observe_subject(Some(" alice "))); + assert_eq!( + observe_subject(Some(" alice ")), + IdentityObservation::Unchanged + ); } } @@ -263,3 +302,49 @@ mod partition_tests { } } } + +#[cfg(test)] +mod observation_tests { + use super::*; + + #[test] + fn the_first_identified_answer_reloads_storage_without_resetting() { + // The mount-timing bug: the shell loads settings under `anon` before + // any status answer, so the first real subject has to re-read them — + // but must NOT throw away the document, which is not account-scoped + // until an account has actually been shown. + reset_for_test(); + let outcome = observe_subject(Some("alice")); + assert_eq!(outcome, IdentityObservation::FirstIdentified); + assert!(outcome.requires_storage_reload()); + assert!(!outcome.requires_reset()); + } + + #[test] + fn a_switch_both_resets_and_reloads() { + reset_for_test(); + observe_subject(Some("alice")); + let outcome = observe_subject(Some("bob")); + assert_eq!(outcome, IdentityObservation::Changed); + assert!(outcome.requires_reset()); + assert!(outcome.requires_storage_reload()); + } + + #[test] + fn a_repeat_does_neither() { + reset_for_test(); + observe_subject(Some("alice")); + let outcome = observe_subject(Some("alice")); + assert!(!outcome.requires_reset()); + assert!(!outcome.requires_storage_reload()); + } + + #[test] + fn an_anonymous_first_answer_does_neither() { + // Nothing moved: the tab was already on the `anon` partition. + reset_for_test(); + let outcome = observe_subject(None); + assert_eq!(outcome, IdentityObservation::Unchanged); + assert!(!outcome.requires_storage_reload()); + } +} diff --git a/crates/op-host-web/src/live_sync_conflict.rs b/crates/op-host-web/src/live_sync_conflict.rs index 97e1707e5..d9166ea59 100644 --- a/crates/op-host-web/src/live_sync_conflict.rs +++ b/crates/op-host-web/src/live_sync_conflict.rs @@ -18,7 +18,7 @@ use crate::repaint_ctx::RepaintContext; /// Best effort on the borrow: if the shell is mid-render the accept still has /// to proceed (the alternative is the latch that never lifts), and the next /// conflict will stash again. -fn preserve_local_document(inner: &Rc>) { +pub(super) fn preserve_local_document(inner: &Rc>) { let Ok(mut context) = inner.try_borrow_mut() else { return; }; diff --git a/crates/op-host-web/src/live_sync_glue.rs b/crates/op-host-web/src/live_sync_glue.rs index eb847fcf8..add581732 100644 --- a/crates/op-host-web/src/live_sync_glue.rs +++ b/crates/op-host-web/src/live_sync_glue.rs @@ -309,6 +309,12 @@ fn maybe_auto_resolve_conflict_in_session( if !auto_resolve_is_safe(has_conflict, phase, server_is_authoritative()) { return; } + // Accepting the remote overwrites whatever this tab had not yet pushed. + // Undo is the primary way back (the apply is undoable); this stash is the + // belt-and-braces copy plus the notice that tells the user any of it + // happened. It must run BEFORE the resolve, while the local document is + // still the one on screen. + live_sync_conflict::preserve_local_document(inner); // Re-opens the pull for THIS pair only; the resolving pull's apply calls // `note_synced`, which is what finally clears the baseline and reopens the // push side too. diff --git a/crates/op-host-web/src/live_sync_identity.rs b/crates/op-host-web/src/live_sync_identity.rs index 7bed0e68b..ab6cf37cc 100644 --- a/crates/op-host-web/src/live_sync_identity.rs +++ b/crates/op-host-web/src/live_sync_identity.rs @@ -63,5 +63,11 @@ pub(crate) fn reset_for_new_identity(inner: &Rc(inner: &Rc>, base: &str) }; // Identity first: everything below paints for an account, so the // account has to be settled before any of it runs. - if crate::identity_epoch::observe_subject( + let observation = crate::identity_epoch::observe_subject( crate::identity_epoch::subject_from_status(&body).as_deref(), - ) { + ); + if observation.requires_reset() { crate::live_sync_glue::reset_for_new_identity(&inner); } + if observation.requires_storage_reload() { + // The shell loaded settings and credentials under `anon` at + // mount; the account's own partition is a different key, so + // what is in memory belongs to the wrong one until re-read. + crate::web_settings::reload_for_active_partition(&inner); + } sync_account_avatar(&inner, parsed["avatar_revision"].as_str()); let mut b = inner.borrow_mut(); let ui = &mut b.host_mut().editor_state_mut().editor_ui; diff --git a/crates/op-host-web/src/web_settings.rs b/crates/op-host-web/src/web_settings.rs index fa40269c6..5e32dc9d0 100644 --- a/crates/op-host-web/src/web_settings.rs +++ b/crates/op-host-web/src/web_settings.rs @@ -30,6 +30,24 @@ const CREDENTIAL_PAYLOAD_VERSION: u32 = 2; const STORAGE_KEY: &str = "openpencil-rust-web-settings"; const CREDENTIAL_STORAGE_KEY: &str = "openpencil-rust-web-credentials"; +/// Re-read account-scoped storage after the tab's partition changed. +/// +/// The shell loads settings and credentials at mount, before any +/// `/api/auth/status` answer has arrived — so it loads them from the `anon` +/// partition. Once the account is known the partition key changes, and what is +/// in memory belongs to the wrong one: without this the tab keeps showing (and +/// re-saving) anonymous settings under the account's key. +pub(crate) fn reload_for_active_partition( + inner: &std::rc::Rc>, +) { + let Ok(mut context) = inner.try_borrow_mut() else { + return; + }; + let _ = storage::load_into(context.host_mut().editor_state_mut()); + context.host_mut().mark_editor_state_dirty(); + let _ = context.repaint(); +} + /// Per-account storage keys. /// /// The base keys above are same-origin and carry no account dimension, so two