fix(web): credential partition hygiene, shutdown write barrier, seed lifecycle
- an empty settings partition now clears the previous account's in-memory credentials instead of keeping them; identity reset also drains the credential upload queue and rebuilds both persistence fingerprints after the partition load (before it would re-measure the old account's state as a pending write); partitions key on the stable subject the daemon now advertises, so a handle rename cannot move or merge accounts - online shutdown closes a write barrier before draining: late writes get 503 shutting-down instead of an ack that will not persist, and the flush logs its duration so stop_grace_period can be sized - a conflict auto-accept that cannot stash skips the accept and retries rather than overwriting unpushed work without a backup - the collab gate runs before the push surrenders its document, so refused pushes release their thumbnail seed through Drop; the runtime-reject path discards explicitly, and the seed tests now use the same allocation the parse produced instead of a clone that could never match
This commit is contained in:
parent
1507911199
commit
62d944710b
|
|
@ -278,10 +278,25 @@ impl WebCanvasState {
|
|||
});
|
||||
}
|
||||
}
|
||||
// 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.
|
||||
// The gateway runs BEFORE the document is taken out of `push`, so a
|
||||
// refusal leaves `push` still owning it — and its `Drop` releases the
|
||||
// pending thumbnail seed. Taking first meant a gated push orphaned its
|
||||
// seed in the process-global side table.
|
||||
//
|
||||
// A metadata-only push carries no document and is not a document
|
||||
// mutation, so it is not gated; it is handled below.
|
||||
let mut push = push;
|
||||
if push.prepared.is_some() {
|
||||
self.gate_daemon_mutation(
|
||||
op_editor_core::CollabGateAction::Document(
|
||||
op_editor_core::CollabDocumentMutation::NodePropertyBatch,
|
||||
),
|
||||
op_editor_core::CollabEditSource::User,
|
||||
)
|
||||
.map_err(WebCanvasError::Collab)?;
|
||||
}
|
||||
// Taken rather than moved: taking `prepared` is what tells the `Drop`
|
||||
// this push reached the install.
|
||||
let editor_meta = push.take_editor_meta();
|
||||
let Some(prepared) = push.prepared.take() else {
|
||||
// Active-page changes do not mutate canonical document content.
|
||||
|
|
@ -294,15 +309,6 @@ impl WebCanvasState {
|
|||
current_version: self.version,
|
||||
});
|
||||
};
|
||||
// The gateway. Returns Ok untouched when there is no session, so a
|
||||
// standalone daemon behaves exactly as it did before collaboration.
|
||||
self.gate_daemon_mutation(
|
||||
op_editor_core::CollabGateAction::Document(
|
||||
op_editor_core::CollabDocumentMutation::NodePropertyBatch,
|
||||
),
|
||||
op_editor_core::CollabEditSource::User,
|
||||
)
|
||||
.map_err(WebCanvasError::Collab)?;
|
||||
let version = if self.collab_session_is_active() {
|
||||
// Inside the session's local-edit capture: the session core diffs
|
||||
// the document and emits precise node operations, so pushing an
|
||||
|
|
|
|||
|
|
@ -343,6 +343,10 @@ impl WebCanvasState {
|
|||
) -> IngestOutcome {
|
||||
let (runtime, mut host) = self.collab_runtime_and_host();
|
||||
if !runtime.begin_local_edit(&mut host) {
|
||||
// The document is about to be dropped without ever activating, so
|
||||
// release its pending thumbnail seed — otherwise it lingers in the
|
||||
// process-global side table until bounded eviction pushes it out.
|
||||
jian_ops_schema::image_thumbs::discard_for_document(prepared.document());
|
||||
return IngestOutcome::Rejected;
|
||||
}
|
||||
// The capture is now open and MUST be closed on every path. The guard
|
||||
|
|
|
|||
|
|
@ -344,3 +344,37 @@ fn a_rejected_ingest_answers_409_with_the_authoritative_version() {
|
|||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_gated_push_still_owns_its_document_so_the_seed_is_released() {
|
||||
// The gate used to run AFTER `prepared.take()`, so a refusal left the
|
||||
// push without the document — and its `Drop` had nothing to release. This
|
||||
// pins the ordering by exercising it: a session that refuses the push must
|
||||
// not leave a seed behind.
|
||||
use crate::web_canvas_server::{PendingDocumentPush, ServeMode};
|
||||
|
||||
let mut state = daemon();
|
||||
in_session(
|
||||
&mut state,
|
||||
CollabConnectionPhase::Active,
|
||||
CollabUiRole::Viewer,
|
||||
);
|
||||
|
||||
let body = serde_json::json!({
|
||||
"document": {
|
||||
"version": "1.0.0",
|
||||
"children": [{
|
||||
"id": "n1", "type": "rectangle", "name": "gated",
|
||||
"x": 0, "y": 0, "width": 4, "height": 4,
|
||||
}],
|
||||
"imageThumbs": { "6161": "AQID" },
|
||||
},
|
||||
"sourceClientId": "s",
|
||||
})
|
||||
.to_string();
|
||||
|
||||
let push = PendingDocumentPush::parse(&body, ServeMode::Local).expect("parses");
|
||||
// A viewer may not write, so the gate refuses before the document moves.
|
||||
let refused = state.apply_prepared_document_push(push, None);
|
||||
assert!(refused.is_err(), "a viewer's push must be refused");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,6 +30,9 @@ 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::ResolvedIdentity>,
|
||||
/// Admits document writes until shutdown closes it. `None` for the local
|
||||
/// and managed daemons, whose lifecycle has no flush to protect.
|
||||
pub(super) write_barrier: Option<&'a super::tenant::WriteBarrier>,
|
||||
}
|
||||
|
||||
/// Handle one connection against the single-user document authority.
|
||||
|
|
@ -82,6 +85,7 @@ pub(super) fn serve_one_in_mode<S: Read + Write>(
|
|||
mode,
|
||||
mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::UNRESTRICTED,
|
||||
rest_identity: None,
|
||||
write_barrier: None,
|
||||
},
|
||||
)
|
||||
}
|
||||
|
|
@ -331,6 +335,24 @@ pub(super) fn dispatch<S: Read + Write>(
|
|||
// the lock, where it is not a race.
|
||||
let pending_push = (req.method == "POST" && req.path == "/api/mcp/document")
|
||||
.then(|| PendingDocumentPush::parse(&req.body, ctx.mode));
|
||||
// A document write must hold a pass for its whole locked segment, so
|
||||
// shutdown cannot snapshot the document between the pass being taken
|
||||
// and the commit landing.
|
||||
let write_pass = match (ctx.write_barrier, pending_push.is_some()) {
|
||||
(Some(barrier), true) => match barrier.enter() {
|
||||
Some(pass) => Some(pass),
|
||||
None => {
|
||||
crate::mcp_serve::write_mcp_http_response_with_origin(
|
||||
stream,
|
||||
"503 Service Unavailable",
|
||||
r#"{"ok":false,"error":"shutting-down","message":"this daemon is stopping and cannot accept writes"}"#,
|
||||
cors_origin,
|
||||
)?;
|
||||
return Ok(false);
|
||||
}
|
||||
},
|
||||
_ => None,
|
||||
};
|
||||
let reply = {
|
||||
let mut guard = state.lock().unwrap_or_else(|p| p.into_inner());
|
||||
let before = guard.version;
|
||||
|
|
@ -363,6 +385,8 @@ pub(super) fn dispatch<S: Read + Write>(
|
|||
}
|
||||
reply
|
||||
};
|
||||
// Released only after the commit is visible in the shared state.
|
||||
drop(write_pass);
|
||||
crate::mcp_serve::write_mcp_http_response_with_origin(
|
||||
stream,
|
||||
reply.status,
|
||||
|
|
|
|||
|
|
@ -60,6 +60,21 @@ fn drain_connections(conn_count: &Arc<AtomicUsize>) -> bool {
|
|||
conn_count.load(Ordering::Acquire) == 0
|
||||
}
|
||||
|
||||
/// Wait for in-flight document writes to finish, up to the bound.
|
||||
///
|
||||
/// Separate from the connection drain: an SSE stream holds a connection for
|
||||
/// minutes and is safe to flush past, while a write in progress is not.
|
||||
fn drain_write_barrier(barrier: &Arc<super::tenant::WriteBarrier>) -> bool {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(SHUTDOWN_DRAIN_SECS);
|
||||
while std::time::Instant::now() < deadline {
|
||||
if barrier.active() == 0 {
|
||||
return true;
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(20));
|
||||
}
|
||||
barrier.active() == 0
|
||||
}
|
||||
|
||||
/// Longest gap between idle sweeps, however long the idle deadline is.
|
||||
const MAX_SWEEP_INTERVAL_SECS: u64 = 300;
|
||||
|
||||
|
|
@ -148,6 +163,7 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> {
|
|||
store,
|
||||
));
|
||||
let conn_count = Arc::new(AtomicUsize::new(0));
|
||||
let write_barrier = Arc::new(super::tenant::WriteBarrier::default());
|
||||
let shutdown = Arc::new(AtomicBool::new(false));
|
||||
// A container stop is a SIGTERM, and without a handler it kills the
|
||||
// process where it stands — losing every resident tenant that had not
|
||||
|
|
@ -180,6 +196,7 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> {
|
|||
let registry = Arc::clone(®istry);
|
||||
let verifier = Arc::clone(&verifier);
|
||||
let conns = Arc::clone(&conn_count);
|
||||
let write_barrier = Arc::clone(&write_barrier);
|
||||
let shutdown_flag = Arc::clone(&shutdown);
|
||||
let spawned = thread::Builder::new()
|
||||
.name("op-serve-web-online-conn".into())
|
||||
|
|
@ -187,7 +204,12 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> {
|
|||
let _conn_guard = ConnGuard(conns);
|
||||
let _ = s.set_read_timeout(Some(IO_TIMEOUT));
|
||||
let _ = s.set_write_timeout(Some(IO_TIMEOUT));
|
||||
match serve_one_online(&mut s, registry.as_ref(), verifier.as_ref()) {
|
||||
match serve_one_online(
|
||||
&mut s,
|
||||
registry.as_ref(),
|
||||
verifier.as_ref(),
|
||||
write_barrier.as_ref(),
|
||||
) {
|
||||
Ok(true) => {
|
||||
shutdown_flag.store(true, Ordering::Release);
|
||||
let _ = std::net::TcpStream::connect(("127.0.0.1", bound));
|
||||
|
|
@ -207,19 +229,35 @@ pub fn run_online_web_canvas(options: ServeWebOptions) -> Result<()> {
|
|||
// 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.
|
||||
// Stop admitting writes FIRST: a worker past the drain check but not yet
|
||||
// holding a pass would otherwise commit after the flush snapshotted the
|
||||
// document, having already answered 200.
|
||||
write_barrier.close();
|
||||
let writes_settled = drain_write_barrier(&write_barrier);
|
||||
if !writes_settled {
|
||||
eprintln!(
|
||||
"openpencil --serve-web --online: {} write(s) still in flight after {}s; \
|
||||
flushing anyway — those clients were acked but may not be persisted",
|
||||
write_barrier.active(),
|
||||
SHUTDOWN_DRAIN_SECS
|
||||
);
|
||||
}
|
||||
let drained = drain_connections(&conn_count);
|
||||
let flush_started = std::time::Instant::now();
|
||||
let flushed = registry.flush_all();
|
||||
if !drained {
|
||||
eprintln!(
|
||||
"openpencil --serve-web --online: {} connection(s) still active after {}s; \
|
||||
flushing anyway",
|
||||
"openpencil --serve-web --online: {} connection(s) still active after {}s",
|
||||
conn_count.load(Ordering::Acquire),
|
||||
SHUTDOWN_DRAIN_SECS
|
||||
);
|
||||
}
|
||||
// The duration is logged because a slow volume is what makes
|
||||
// `stop_grace_period` too short, and there is no other way to size it.
|
||||
eprintln!(
|
||||
"openpencil --serve-web --online: shutdown requested; flushed {flushed} account(s); \
|
||||
exiting"
|
||||
"openpencil --serve-web --online: shutdown requested; flushed {flushed} account(s) in \
|
||||
{} ms; exiting",
|
||||
flush_started.elapsed().as_millis()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -438,6 +476,7 @@ pub(super) fn serve_one_online<S: Read + Write>(
|
|||
stream: &mut S,
|
||||
registry: &TenantRegistry,
|
||||
verifier: &dyn IdentityVerifier,
|
||||
write_barrier: &super::tenant::WriteBarrier,
|
||||
) -> Result<bool> {
|
||||
let req = crate::mcp_serve::read_http_request(stream)?;
|
||||
let allow_origins = registry.allow_origins();
|
||||
|
|
@ -583,6 +622,7 @@ pub(super) fn serve_one_online<S: Read + Write>(
|
|||
// this particular credential carries.
|
||||
mcp_profile: crate::mcp_serve::tool_profile::McpAccessProfile::online(identity.scopes),
|
||||
rest_identity: Some(identity.clone()),
|
||||
write_barrier: Some(write_barrier),
|
||||
},
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -133,7 +133,9 @@ fn serve(registry: &TenantRegistry, verifier: &StaticVerifier, request: Request)
|
|||
input: std::io::Cursor::new(request.wire().into_bytes()),
|
||||
output: Vec::new(),
|
||||
};
|
||||
serve_one_online(&mut stream, registry, verifier).expect("serve_one_online");
|
||||
// An open barrier: the shutdown path is exercised by its own tests.
|
||||
let barrier = crate::web_canvas_server::tenant::WriteBarrier::default();
|
||||
serve_one_online(&mut stream, registry, verifier, &barrier).expect("serve_one_online");
|
||||
String::from_utf8_lossy(&stream.output).into_owned()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -772,3 +772,21 @@ fn a_disabled_store_passes_the_probe_trivially() {
|
|||
.check_writable()
|
||||
.expect("no store, no probe");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_status_projection_carries_a_stable_subject_distinct_from_the_handle() {
|
||||
// The shell partitions its storage by this, so it must be the same key the
|
||||
// tenant registry uses — not a renameable display handle.
|
||||
let response = serve(
|
||||
®istry(),
|
||||
&verifier(),
|
||||
Request::new("GET", op_editor_core::auth_routes::STATUS).with_bearer("tokA"),
|
||||
);
|
||||
let payload = body(&response);
|
||||
assert_eq!(payload["subject"], "userA");
|
||||
assert!(payload["username"].is_string());
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "online_shutdown_tests.rs"]
|
||||
mod shutdown;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,83 @@
|
|||
//! The shutdown write barrier and its start-up probes.
|
||||
//!
|
||||
//! Split out of `online_share_tests.rs` at the 800-line cap; nested under it
|
||||
//! so `use super::*` still reaches the request builder and helpers.
|
||||
|
||||
use super::*;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// B3: the shutdown write barrier.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[test]
|
||||
fn the_write_barrier_admits_writes_until_shutdown_closes_it() {
|
||||
use crate::web_canvas_server::tenant::WriteBarrier;
|
||||
let barrier = WriteBarrier::default();
|
||||
assert!(!barrier.is_closed());
|
||||
|
||||
let pass = barrier.enter().expect("open");
|
||||
assert_eq!(barrier.active(), 1);
|
||||
drop(pass);
|
||||
assert_eq!(barrier.active(), 0);
|
||||
|
||||
barrier.close();
|
||||
assert!(barrier.is_closed());
|
||||
assert!(
|
||||
barrier.enter().is_none(),
|
||||
"a stopping daemon must refuse writes rather than ack one it will not persist"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_held_pass_keeps_the_barrier_busy_so_the_flush_waits() {
|
||||
// The window this closes: a worker past the connection drain, about to
|
||||
// take the state lock, commits after the flush snapshotted the document —
|
||||
// having already answered 200.
|
||||
use crate::web_canvas_server::tenant::WriteBarrier;
|
||||
let barrier = WriteBarrier::default();
|
||||
let held = barrier.enter().expect("open");
|
||||
barrier.close();
|
||||
assert_eq!(
|
||||
barrier.active(),
|
||||
1,
|
||||
"closing must not abandon a write already inside"
|
||||
);
|
||||
drop(held);
|
||||
assert_eq!(barrier.active(), 0, "the drain can now proceed");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn closing_between_the_check_and_the_increment_does_not_admit_a_writer() {
|
||||
// `enter` re-checks after incrementing precisely so a close landing in
|
||||
// that window cannot admit a writer the drain has stopped waiting for.
|
||||
use crate::web_canvas_server::tenant::WriteBarrier;
|
||||
let barrier = std::sync::Arc::new(WriteBarrier::default());
|
||||
barrier.close();
|
||||
for _ in 0..100 {
|
||||
assert!(barrier.enter().is_none());
|
||||
}
|
||||
assert_eq!(barrier.active(), 0, "a refused entry must not leak a count");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_write_during_shutdown_is_refused_rather_than_acked() {
|
||||
use crate::web_canvas_server::tenant::WriteBarrier;
|
||||
let registry = registry();
|
||||
let verifier = verifier();
|
||||
let barrier = WriteBarrier::default();
|
||||
barrier.close();
|
||||
|
||||
let request = Request::json("POST", "/api/mcp/document", SYNC_BODY).with_bearer("tokA");
|
||||
let mut stream = MockStream {
|
||||
input: std::io::Cursor::new(request.wire().into_bytes()),
|
||||
output: Vec::new(),
|
||||
};
|
||||
serve_one_online(&mut stream, ®istry, &verifier, &barrier).expect("serve");
|
||||
let response = String::from_utf8_lossy(&stream.output).into_owned();
|
||||
assert_eq!(
|
||||
status_line(&response),
|
||||
"HTTP/1.1 503 Service Unavailable",
|
||||
"{response}"
|
||||
);
|
||||
assert_eq!(body(&response)["error"], "shutting-down");
|
||||
}
|
||||
|
|
@ -28,7 +28,7 @@
|
|||
//! still the exact `Arc` it decided to evict.
|
||||
|
||||
use std::collections::{BTreeSet, HashMap};
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use op_editor_core::EditorState;
|
||||
|
|
@ -323,6 +323,65 @@ impl Drop for TenantLease {
|
|||
}
|
||||
}
|
||||
|
||||
/// Counts requests that are inside a document write, and refuses new ones
|
||||
/// once shutdown has begun.
|
||||
///
|
||||
/// Draining connections alone is not enough: a worker can be past the drain
|
||||
/// check and about to take the state lock when the flush snapshots the
|
||||
/// document. It then commits and answers 200 for work the flush never saw.
|
||||
/// The barrier closes that window — shutdown stops admitting writes and then
|
||||
/// waits for the ones already inside to finish.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct WriteBarrier {
|
||||
active: AtomicUsize,
|
||||
closed: AtomicBool,
|
||||
}
|
||||
|
||||
impl WriteBarrier {
|
||||
/// Enter the write path, unless shutdown has closed it.
|
||||
///
|
||||
/// `None` means the caller must refuse the request (503): the daemon is
|
||||
/// stopping and cannot durably accept a write.
|
||||
pub fn enter(&self) -> Option<WritePass<'_>> {
|
||||
if self.closed.load(Ordering::Acquire) {
|
||||
return None;
|
||||
}
|
||||
self.active.fetch_add(1, Ordering::AcqRel);
|
||||
// Re-checked after the increment: a close landing between the two
|
||||
// would otherwise admit a writer the drain has already stopped
|
||||
// waiting for.
|
||||
if self.closed.load(Ordering::Acquire) {
|
||||
self.active.fetch_sub(1, Ordering::AcqRel);
|
||||
return None;
|
||||
}
|
||||
Some(WritePass { barrier: self })
|
||||
}
|
||||
|
||||
/// Stop admitting writes. Idempotent.
|
||||
pub fn close(&self) {
|
||||
self.closed.store(true, Ordering::Release);
|
||||
}
|
||||
|
||||
pub fn is_closed(&self) -> bool {
|
||||
self.closed.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
pub fn active(&self) -> usize {
|
||||
self.active.load(Ordering::Acquire)
|
||||
}
|
||||
}
|
||||
|
||||
/// Proof that a write is in flight. Decrements on drop.
|
||||
pub struct WritePass<'a> {
|
||||
barrier: &'a WriteBarrier,
|
||||
}
|
||||
|
||||
impl Drop for WritePass<'_> {
|
||||
fn drop(&mut self) {
|
||||
self.barrier.active.fetch_sub(1, Ordering::AcqRel);
|
||||
}
|
||||
}
|
||||
|
||||
/// Every live tenant, keyed by verified account id.
|
||||
pub struct TenantRegistry {
|
||||
tenants: Mutex<HashMap<String, Arc<Tenant>>>,
|
||||
|
|
|
|||
|
|
@ -155,7 +155,14 @@ impl ResolvedIdentity {
|
|||
// 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.
|
||||
// The STABLE account key — the same value the tenant registry
|
||||
// keys on. The shell partitions its storage by this, not by
|
||||
// `username`: a username is a display handle a hub may let a user
|
||||
// change, and a rename would silently move the tab to a new
|
||||
// partition (losing its settings) or, worse, collide with another
|
||||
// account that later takes the old handle.
|
||||
"subject": self.user_id,
|
||||
// Display only.
|
||||
"username": self.username,
|
||||
"display_name": self.display_name,
|
||||
// Not projected: the hub owns the address, and the shell only uses
|
||||
|
|
|
|||
|
|
@ -315,15 +315,50 @@ fn a_persisted_tenant_carries_no_thumbnail_data() {
|
|||
}
|
||||
|
||||
#[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.
|
||||
fn a_parsed_push_really_registers_its_thumbnail_seed() {
|
||||
// Grounded in the true key semantics: `DocumentSeedKey` is the version
|
||||
// String's (ptr, len, capacity), so the seed can only be observed through
|
||||
// the SAME document allocation the parse produced. Asserting on a cloned
|
||||
// document — as an earlier version of this test did — is vacuous: a clone
|
||||
// has a different pointer and always misses.
|
||||
use crate::web_canvas_server::{PendingDocumentPush, ServeMode};
|
||||
|
||||
let body = serde_json::json!({
|
||||
let body = seeded_push_body();
|
||||
let mut push = PendingDocumentPush::parse(&body, ServeMode::Local).expect("parses");
|
||||
let prepared = push.prepared.take().expect("a document push");
|
||||
|
||||
// Consuming the seed through the real document proves it was registered:
|
||||
// `discard_for_document` returns true only when a pending seed exists for
|
||||
// exactly this allocation.
|
||||
assert!(
|
||||
jian_ops_schema::image_thumbs::discard_for_document(prepared.document()),
|
||||
"the parse must register a pending seed under this document"
|
||||
);
|
||||
// …and it is gone, so a second release is a no-op — which is what makes
|
||||
// the `Drop`, the gate path and the runtime-reject path idempotent.
|
||||
assert!(!jian_ops_schema::image_thumbs::discard_for_document(
|
||||
prepared.document()
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_online_push_never_registers_a_seed_at_all() {
|
||||
// Online policy discards at parse time, so there is nothing for any later
|
||||
// path to leak.
|
||||
use crate::web_canvas_server::{PendingDocumentPush, ServeMode};
|
||||
|
||||
let body = seeded_push_body();
|
||||
let mut push = PendingDocumentPush::parse(&body, ServeMode::Online).expect("parses");
|
||||
let prepared = push.prepared.take().expect("a document push");
|
||||
assert!(
|
||||
!jian_ops_schema::image_thumbs::discard_for_document(prepared.document()),
|
||||
"online must not register a seed the shared registry could activate"
|
||||
);
|
||||
}
|
||||
|
||||
/// A push body carrying an embedded thumbnail.
|
||||
fn seeded_push_body() -> String {
|
||||
serde_json::json!({
|
||||
"document": {
|
||||
"version": "1.0.0",
|
||||
"children": [{
|
||||
|
|
@ -334,29 +369,5 @@ fn a_push_that_never_installs_releases_its_thumbnail_seed() {
|
|||
},
|
||||
"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"
|
||||
);
|
||||
.to_string()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -177,6 +177,17 @@ impl crate::repaint_ctx::RepaintContext for CkInner {
|
|||
fn host_mut(&mut self) -> &mut crate::widget_host::WidgetHost {
|
||||
&mut self.host
|
||||
}
|
||||
|
||||
fn reset_persistence_baselines(&mut self) {
|
||||
// Recomputed from the account's own freshly loaded state on the next
|
||||
// comparison, rather than carried over from the previous account's.
|
||||
self.settings_fingerprint = None;
|
||||
// Rebuilt from the state the account's own partition just loaded, so
|
||||
// the next comparison measures against THIS account rather than
|
||||
// reporting the whole partition as a change.
|
||||
self.credential_fingerprint =
|
||||
crate::web_settings::credential_fingerprint(self.host.editor_state());
|
||||
}
|
||||
fn viewport_size(&self) -> (f32, f32) {
|
||||
self.backend.logical_size()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -127,8 +127,11 @@ pub fn subject_from_status(body: &str) -> Option<String> {
|
|||
if !parsed["signed_in"].as_bool().unwrap_or(false) {
|
||||
return None;
|
||||
}
|
||||
parsed["username"]
|
||||
// `subject` is the stable account key; `username` is a display handle and
|
||||
// is only used when talking to a daemon that predates the field.
|
||||
parsed["subject"]
|
||||
.as_str()
|
||||
.or_else(|| parsed["username"].as_str())
|
||||
.map(str::trim)
|
||||
.filter(|value| !value.is_empty())
|
||||
.map(str::to_string)
|
||||
|
|
@ -348,3 +351,58 @@ mod observation_tests {
|
|||
assert!(!outcome.requires_storage_reload());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod subject_stability_tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn the_stable_subject_wins_over_the_display_handle() {
|
||||
// The partition key must be the account id, not the username: a hub
|
||||
// may let a user rename, and a rename would silently move the tab to a
|
||||
// fresh partition — or collide with whoever later takes the old handle.
|
||||
assert_eq!(
|
||||
subject_from_status(r#"{"signed_in":true,"subject":"user-uuid-1","username":"alice"}"#)
|
||||
.as_deref(),
|
||||
Some("user-uuid-1")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rename_does_not_move_the_partition() {
|
||||
reset_for_test();
|
||||
observe_subject(
|
||||
subject_from_status(r#"{"signed_in":true,"subject":"u1","username":"alice"}"#)
|
||||
.as_deref(),
|
||||
);
|
||||
let before = current_subject();
|
||||
let outcome = observe_subject(
|
||||
subject_from_status(r#"{"signed_in":true,"subject":"u1","username":"renamed"}"#)
|
||||
.as_deref(),
|
||||
);
|
||||
assert_eq!(outcome, IdentityObservation::Unchanged);
|
||||
assert_eq!(current_subject(), before);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_older_daemon_without_a_subject_falls_back_to_the_username() {
|
||||
assert_eq!(
|
||||
subject_from_status(r#"{"signed_in":true,"username":"alice"}"#).as_deref(),
|
||||
Some("alice")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn two_accounts_that_share_a_display_handle_stay_separate() {
|
||||
reset_for_test();
|
||||
observe_subject(
|
||||
subject_from_status(r#"{"signed_in":true,"subject":"u1","username":"same"}"#)
|
||||
.as_deref(),
|
||||
);
|
||||
let outcome = observe_subject(
|
||||
subject_from_status(r#"{"signed_in":true,"subject":"u2","username":"same"}"#)
|
||||
.as_deref(),
|
||||
);
|
||||
assert_eq!(outcome, IdentityObservation::Changed);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,12 +15,13 @@ use crate::repaint_ctx::RepaintContext;
|
|||
/// Keep the about-to-be-overwritten local document, and raise the notice that
|
||||
/// tells the user it is recoverable.
|
||||
///
|
||||
/// 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.
|
||||
pub(super) fn preserve_local_document<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) {
|
||||
/// Returns whether the local document was preserved. A `false` means the
|
||||
/// caller must NOT accept the remote this tick: overwriting without a backup
|
||||
/// is the data loss this exists to prevent, and the latch simply stays up for
|
||||
/// one more tick, which converges a few hundred milliseconds later.
|
||||
pub(super) fn preserve_local_document<C: RepaintContext + 'static>(inner: &Rc<RefCell<C>>) -> bool {
|
||||
let Ok(mut context) = inner.try_borrow_mut() else {
|
||||
return;
|
||||
return false;
|
||||
};
|
||||
let now_ms = crate::collab_sync::now_ms();
|
||||
let state = context.host_mut().editor_state_mut();
|
||||
|
|
@ -30,6 +31,7 @@ pub(super) fn preserve_local_document<C: RepaintContext + 'static>(inner: &Rc<Re
|
|||
.collab
|
||||
.set_notice(op_editor_core::CollabNoticeKind::LocalEditPreserved, now_ms);
|
||||
context.host_mut().mark_editor_state_dirty();
|
||||
true
|
||||
}
|
||||
|
||||
/// The safety decision behind [`maybe_auto_resolve_conflict_in_session`],
|
||||
|
|
|
|||
|
|
@ -314,7 +314,11 @@ fn maybe_auto_resolve_conflict_in_session<C: RepaintContext + 'static>(
|
|||
// 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);
|
||||
if !live_sync_conflict::preserve_local_document(inner) {
|
||||
// No backup taken, so nothing is accepted. Better to converge one tick
|
||||
// later than to overwrite unpushed work with no way back.
|
||||
return;
|
||||
}
|
||||
// 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.
|
||||
|
|
|
|||
|
|
@ -125,3 +125,21 @@ mod call_site_tests {
|
|||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod skip_without_backup_tests {
|
||||
/// A failed stash must abort the accept for this tick, not proceed
|
||||
/// unprotected: overwriting unpushed work with no way back is the exact
|
||||
/// loss this path exists to prevent, and the latch simply retries.
|
||||
#[test]
|
||||
fn the_accept_is_skipped_when_the_document_cannot_be_stashed() {
|
||||
let glue = include_str!("live_sync_glue.rs");
|
||||
let guarded = glue
|
||||
.find("if !live_sync_conflict::preserve_local_document(inner)")
|
||||
.expect("the accept must be guarded on a successful stash");
|
||||
let resolve = glue
|
||||
.find("resolve_accept_remote")
|
||||
.expect("the accept exists");
|
||||
assert!(guarded < resolve, "the guard must precede the accept");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,6 +16,16 @@ use wasm_bindgen::JsValue;
|
|||
pub(crate) trait RepaintContext {
|
||||
fn host(&self) -> &WidgetHost;
|
||||
fn host_mut(&mut self) -> &mut WidgetHost;
|
||||
/// Forget the settings/credential change-detection baselines.
|
||||
///
|
||||
/// Called when the tab switches accounts. The baselines were computed
|
||||
/// against the PREVIOUS account's state, so keeping them means the first
|
||||
/// comparison after the switch reports a spurious change and writes — or
|
||||
/// uploads — the new account's partition against the old one's shape.
|
||||
///
|
||||
/// Defaulted to a no-op: a backend with no persistence baselines (the
|
||||
/// smoke harness) has nothing to forget.
|
||||
fn reset_persistence_baselines(&mut self) {}
|
||||
/// Logical viewport size in CSS pixels. File-open/import flows fit the
|
||||
/// loaded content to this size after replacing the document.
|
||||
fn viewport_size(&self) -> (f32, f32);
|
||||
|
|
|
|||
|
|
@ -44,6 +44,11 @@ pub(crate) fn reload_for_active_partition<C: crate::repaint_ctx::RepaintContext>
|
|||
return;
|
||||
};
|
||||
let _ = storage::load_into(context.host_mut().editor_state_mut());
|
||||
// AFTER the load, not before: the baselines must measure against the state
|
||||
// this partition just produced. Rebuilding them from the previous
|
||||
// account's state would make the very next comparison report the whole
|
||||
// partition as a change and write it back under the wrong key.
|
||||
context.reset_persistence_baselines();
|
||||
context.host_mut().mark_editor_state_dirty();
|
||||
let _ = context.repaint();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -163,6 +163,12 @@ where
|
|||
}
|
||||
}
|
||||
StoredCredentialSource::None => {
|
||||
// No stored credentials for THIS partition. Clearing is the point:
|
||||
// after an account switch the in-memory state still holds the
|
||||
// previous account's API keys, and an empty partition must mean
|
||||
// "no keys", not "keep whatever was there". At mount the state is
|
||||
// fresh, so this is a no-op for the local daemon.
|
||||
clear_local_credentials(state);
|
||||
let sanitized = stored.sanitized_settings_json;
|
||||
let settings_saved = sanitized
|
||||
.as_deref()
|
||||
|
|
|
|||
|
|
@ -737,3 +737,32 @@ fn pending_legacy_migration_blocks_an_ordinary_settings_write() {
|
|||
assert!(!general_write_attempted);
|
||||
assert_ne!(settings_baseline, fingerprint(&state));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_empty_partition_clears_the_previous_accounts_credentials() {
|
||||
// The account-switch leak: after switching, the in-memory state still held
|
||||
// account A's API keys, and an empty partition for B meant "keep whatever
|
||||
// was there" instead of "no keys".
|
||||
let mut state = EditorState::new();
|
||||
let mut writes = Vec::new();
|
||||
// A signs in and stores a key.
|
||||
let credential_json = r#"{"version":2,"builtin_agents":[{"id":"a1","preset":"custom",
|
||||
"display_name":"A's model","kind":"openai-compat","api_key":"sk-account-a",
|
||||
"model":"m","base_url":"https://api.example.com/v1","enabled":true}],
|
||||
"image_gen_profiles":[],"active_image_gen_profile_id":null,"openverse_oauth":null}"#;
|
||||
super::storage::load_into_with(&mut state, None, Some(credential_json), |k, v| {
|
||||
writes.push((k.to_string(), v.to_string()));
|
||||
true
|
||||
});
|
||||
assert!(
|
||||
!state.editor_ui.agent_settings.builtin_agents.is_empty(),
|
||||
"A's credential must load in the first place"
|
||||
);
|
||||
|
||||
// B signs in: their partition is empty.
|
||||
super::storage::load_into_with(&mut state, None, None, |_, _| true);
|
||||
assert!(
|
||||
state.editor_ui.agent_settings.builtin_agents.is_empty(),
|
||||
"B must not inherit A's API keys from an empty partition"
|
||||
);
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue