SemianiakaVY/openpencil
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

fix(web): wire the review remedies end to end

Browse source 
The re-review found several remedies that existed but were not
reachable; this closes them:

- online answers /api/auth/status with the verified identity
  projection (login/logout stay 404), so the account-switch epoch
  actually fires; first identification reloads the account's settings
  partition, identity reset also tears down the id allocator, and an
  Active projection whose namespace disagrees rebuilds it
- finish_local_edit reports Committed/NoChange/Rejected/Failed and a
  real runtime rejection answers 409 version-conflict with the
  authoritative version — the one shape the browser's recovery parses
- the REST scope gate runs ahead of every dispatch path including
  share, AI, figma, and SSE (/mcp keeps its stronger per-tool check)
- shutdown drains connections (bounded) before the flush, the tenant
  store probes create/write/rename/delete at startup and fails closed
  naming the container uid, and worker-thread spawn failures abort
  startup instead of silently never persisting
- ACL flushes hold the same guard as updates, and the 257th grant is
  refused rather than silently truncated on the next write
- the conflict stash lost its call site in an earlier split and is
  invoked again before accept; SSE subscribe prunes dead slots; an
  uninstalled document push drops its pending thumbnail seed
This commit is contained in:
Kayshen-X 2026-08-08 19:57:08 +08:00
parent 6e036e1528
commit 1507911199
36 changed files with 1138 additions and 284 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Dockerfile.web-rust
View file

@ -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}

1
crates/op-collab-host/src/lib.rs
View file

@ -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;

11
crates/op-collab-host/src/runtime.rs
View file

@ -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<crate::runtime::local_edit::LocalEditOutcome>,
events: Receiver<TaggedNetworkEvent>,
event_sender: SyncSender<TaggedNetworkEvent>,
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,

5
crates/op-collab-host/src/runtime/conflict_tests.rs
View file

@ -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.

23
crates/op-collab-host/src/runtime/effects.rs
View file

@ -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 {

39
crates/op-collab-host/src/runtime/local_edit.rs
View file

@ -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
}
}
}

15
crates/op-collab-host/src/runtime/tests.rs
View file

@ -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 {

73
crates/op-host-services/src/web_canvas_server.rs
View file

@ -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<PendingDocumentPush>,
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;

33
crates/op-host-services/src/web_canvas_server/collab_state.rs
View file

@ -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
}
}

31
crates/op-host-services/src/web_canvas_server/collab_state_tests.rs
View file

@ -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"
);
}
}

67
crates/op-host-services/src/web_canvas_server/connection.rs
View file

@ -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<super::tenant_auth::ResolvedIdentity>,
}
/// Handle one connection against the single-user document authority.
@ -198,6 +195,47 @@ pub(super) fn dispatch<S: Read + Write>(
)?;
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<S: Read + Write>(
// 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

94
crates/op-host-services/src/web_canvas_server/document_push.rs
View file

@ -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<op_editor_core::PreparedDocument>,
}
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<PendingDocumentPush>,
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()),
},
}
}

71
crates/op-host-services/src/web_canvas_server/online_mcp_tests.rs
View file

@ -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(
&registry,
&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(
&registry(),
&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(
&registry(),
&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)
}

15
crates/op-host-services/src/web_canvas_server/online_policy.rs
View file

@ -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<super::tool_scopes::RestScope> {
if SCOPE_EXEMPT_READS.contains(&path) {
if SCOPE_EXEMPT_PATHS.contains(&path) {
return None;
}
match method {

99
crates/op-host-services/src/web_canvas_server/online_run_loop.rs
View file

@ -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<AtomicUsize>) -> 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(&registry, &shutdown);
install_shutdown_signals(&shutdown, local_addr)?;
spawn_sweeper(&registry, &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<AtomicBool>, local_addr: std::net::SocketAddr) {
fn install_shutdown_signals(
shutdown: &Arc<AtomicBool>,
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<AtomicBool>, 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<TenantRegistry>, shutdown: &Arc<AtomicBool>) {
fn spawn_sweeper(registry: &Arc<TenantRegistry>, shutdown: &Arc<AtomicBool>) -> 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<TenantRegistry>, shutdown: &Arc<AtomicBool>) {
}
}
});
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<S: Read + Write>(
// 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<S: Read + Write>(
// 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()),
},
)
}

7
crates/op-host-services/src/web_canvas_server/online_run_loop_tests.rs
View file

@ -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(&registry, &verifier, request);

111
crates/op-host-services/src/web_canvas_server/online_share_tests.rs
View file

@ -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(
&registry(),
&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(
&registry,
&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(&registry, &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(
&registry(),
&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");
}

13
crates/op-host-services/src/web_canvas_server/share_routes.rs
View file

@ -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

60
crates/op-host-services/src/web_canvas_server/share_routes_tests.rs
View file

@ -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,
&registry,
);
assert_eq!(reply.status, "200 OK", "grant {index}");
}
let overflow = handle(
"POST",
share_routes::GRANT,
r#"{"userId":"one-too-many"}"#,
&owner,
&lease,
&registry,
);
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,
&registry,
);
}
let repeat = handle(
"POST",
share_routes::GRANT,
r#"{"userId":"guest-0"}"#,
&owner,
&lease,
&registry,
);
assert_eq!(repeat.status, "200 OK", "{}", repeat.body);
}

10
crates/op-host-services/src/web_canvas_server/sse_hub.rs
View file

@ -78,10 +78,12 @@ impl SseHub {
/// slot for ticks. Dropping it unregisters.
pub(crate) fn subscribe(&self) -> Arc<SseSlot> {
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
}

21
crates/op-host-services/src/web_canvas_server/tenant.rs
View file

@ -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<AclUpdate, TenantStoreError> {
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.

30
crates/op-host-services/src/web_canvas_server/tenant_auth.rs
View file

@ -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

41
crates/op-host-services/src/web_canvas_server/tenant_store.rs
View file

@ -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.

47
crates/op-host-services/src/web_canvas_server/tenant_store_tests.rs
View file

@ -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"
);
}

4
crates/op-host-services/src/web_canvas_server/tool_scopes_tests.rs
View file

@ -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]

114
crates/op-host-services/src/web_canvas_server_conn_tests.rs
View file

@ -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<u8>,
writes_before_failure: usize,
}
impl std::io::Write for FailingStream {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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;

8
crates/op-host-services/src/web_canvas_server_error.rs
View file

@ -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")
}
}

141
crates/op-host-services/src/web_canvas_server_sse_tests.rs Normal file
View file

@ -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<u8>,
writes_before_failure: usize,
}
impl std::io::Write for FailingStream {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
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);
}

42
crates/op-host-web/src/collab_sync.rs
View file

@ -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.

127
crates/op-host-web/src/identity_epoch.rs
View file

@ -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());
}
}

2
crates/op-host-web/src/live_sync_conflict.rs
View file

@ -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<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
pub(super) fn preserve_local_document<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
let Ok(mut context) = inner.try_borrow_mut() else {
return;
};

6
crates/op-host-web/src/live_sync_glue.rs
View file

@ -309,6 +309,12 @@ fn maybe_auto_resolve_conflict_in_session<C: RepaintContext + 'static>(
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.

6
crates/op-host-web/src/live_sync_identity.rs
View file

@ -63,5 +63,11 @@ pub(crate) fn reset_for_new_identity<C: RepaintContext + 'static>(inner: &Rc<Ref
let _ = context.repaint();
}
crate::collab_sync::reset_for_new_identity();
// The id allocator lives on the host, so it is torn down here where the
// borrow is already held. B1's namespace latch alone was not enough: the
// allocator itself would keep minting in the previous account's namespace.
if let Ok(mut context) = inner.try_borrow_mut() {
crate::collab_sync::reset_id_allocation(context.host_mut());
}
clear_auth_invalid();
}

21
crates/op-host-web/src/live_sync_recovery.rs
View file

@ -104,3 +104,24 @@ mod tests {
assert!(take().is_none());
}
}
#[cfg(test)]
mod call_site_tests {
/// The stash was previously defined but never called — a dead safety net.
/// This pins that the auto-accept path actually references it, so a future
/// refactor that drops the call fails here rather than silently.
#[test]
fn the_auto_accept_path_stashes_before_resolving() {
let glue = include_str!("live_sync_glue.rs");
let resolve_at = glue
.find("resolve_accept_remote")
.expect("the auto-accept path exists");
let stash_at = glue
.find("preserve_local_document")
.expect("the auto-accept path must stash the local document first");
assert!(
stash_at < resolve_at,
"the stash must run while the local document is still on screen"
);
}
}

11
crates/op-host-web/src/web_auth_sync.rs
View file

@ -177,11 +177,18 @@ fn fetch_status<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>, 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;

18
crates/op-host-web/src/web_settings.rs
View file

@ -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<C: crate::repaint_ctx::RepaintContext>(
inner: &std::rc::Rc<std::cell::RefCell<C>>,
) {
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