From 4f2c713af645f840f920def32514241f83c5c841 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 11:42:46 +0200 Subject: [PATCH 1/5] feat: cancel an upload and collect orphaned staged blocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds `Command::CancelUpload { op_id }`, content-only and guaranteed until publish entry. The drain yields at every block boundary so a cancel can actually land mid-transfer, and an interlock decides the race in one borrow: either the facade claims the op and the drain abandons its upload, or the drain claims it for publish and the cancel is refused with `TooLateToCancel`. A cancelled create takes every later queued op on its node; a cancelled version takes nothing. Gives orphan GC its first production caller — cold start and after each drain pass — with open write handles excluded by CID, and reconciles the four-way staged-byte lifetime in one place: preserved on a terminally unrebasable dead letter, released on cancel, on a proven-unopenable blob, and on an unexpandable staged root. Preservation now survives the cold start that drops the op record, via a durable preserved-roots record that GC treats as a second reference source. Closes #853 Part of #655 --- crates/engine/src/content/write.rs | 6 - crates/engine/src/facade.rs | 243 ++++++++-- crates/engine/src/sync/cancel.rs | 94 ++++ crates/engine/src/sync/drain.rs | 93 +++- crates/engine/src/sync/mod.rs | 6 +- crates/engine/src/sync/staging.rs | 326 +++++++++++++- crates/engine/tests/write_plane.rs | 420 +++++++++++++++++- crates/fuse/src/error.rs | 2 + crates/wasm/src/host.rs | 2 + crates/wasm/src/lib.rs | 9 + packages/client/src/facade.ts | 8 + .../client/src/worker/commandCodec.test.ts | 17 + packages/client/src/worker/commandCodec.ts | 2 + packages/client/src/worker/engineWasm.ts | 1 + packages/client/src/worker/protocol.ts | 1 + 15 files changed, 1155 insertions(+), 75 deletions(-) create mode 100644 crates/engine/src/sync/cancel.rs diff --git a/crates/engine/src/content/write.rs b/crates/engine/src/content/write.rs index 86be2bc45..6860f2e6c 100644 --- a/crates/engine/src/content/write.rs +++ b/crates/engine/src/content/write.rs @@ -99,12 +99,6 @@ impl ContentWriter { self.observed } - /// The leaves framed so far, in file order — the staging keys an abandoned - /// write must release. - pub fn staged_leaf_cids(&self) -> &[Vec] { - &self.leaf_cids - } - /// Seal the tail and assemble the root. An empty version frames to exactly /// one empty leaf, so every version has at least one addressable block. pub fn finish(mut self, entropy: &mut impl Entropy) -> Result { diff --git a/crates/engine/src/facade.rs b/crates/engine/src/facade.rs index fae30a448..a0113fe1b 100644 --- a/crates/engine/src/facade.rs +++ b/crates/engine/src/facade.rs @@ -37,6 +37,7 @@ use crate::content::{ use crate::entropy::Entropy; use crate::gate::{GateError, floor}; use crate::hex::hex_lower; +use crate::net::retire::retire; use crate::net::{ Adopter, ChildAdopter, ChildResolveError, EolRenewResult, FolderRefresh, HeldMaterial, HeldRecord, HeldRecords, LivenessControl, PublishError, PublishOutcome, RE_PUT_INTERVAL, @@ -49,9 +50,10 @@ use crate::seams::{OpId, Scheduler, SeamError, SeamSet, SeamTypes, StagingStore, use crate::session::SessionIdentity; use crate::storage_policy::StoragePolicy; use crate::sync::boot::{ColdStartError, ColdStartOutcome, ColdStartParams, cold_start}; -use crate::sync::drain::{Drain, DrainReport, DrainScope}; +use crate::sync::cancel::UploadCancels; +use crate::sync::drain::{Drain, DrainReport, DrainScope, registry_cids}; use crate::sync::model::{NodeMeta, Snapshot, collation_key}; -use crate::sync::op::{NewNode, Op, Replaced, StagedContent}; +use crate::sync::op::{NewNode, Op, OpKind, Replaced, StagedContent}; use crate::sync::overlay::apply_overlay; use crate::sync::pointer::PointerFetch; use crate::sync::project::project_child_version; @@ -59,8 +61,10 @@ use crate::sync::rebase::{QueueScan, QueueScanMemo, decode_queue}; pub use crate::sync::drain::BlockedOp; pub use crate::sync::rebase::DeadLetterReason; -use crate::sync::record::{RecordReader, RecordSeal}; -use crate::sync::staging::stage_op; +use crate::sync::record::{RecordReader, RecordSeal, record_content_root_cid}; +use crate::sync::staging::{ + LiveBlocks, collect_orphans, release_version_blocks, stage_op, version_leaf_cids, +}; use crate::sync::staleness::{Connectivity, classify}; use crate::sync::tick::{FocusWindow, focus_folders, focus_folders_due}; @@ -364,6 +368,20 @@ pub enum Command { /// The node the destination name currently holds, if any. replacing: Option, }, + /// Cancel a queued upload, releasing its staged blocks and retiring + /// whatever of it already reached the network. + /// + /// Content-only: for a metadata op a compensating mutation is already + /// equivalent, while for an upload it is not — a compensating delete still + /// pushes the whole file through the network and never returns the staging + /// budget. Guaranteed until the version's last block confirms and refused + /// after with [`EngineError::TooLateToCancel`], so a cancel never mutates + /// published state (#824). + CancelUpload { + /// The queue id [`Engine::commit_write`] returned. + op_id: OpId, + }, + // --- focus and refresh --- /// Set the open folder driving the focus window; `None` when no folder /// is open. @@ -448,6 +466,7 @@ impl Command { Command::Rename { .. } => "rename", Command::Relink { .. } => "relink", Command::Move { .. } => "move", + Command::CancelUpload { .. } => "cancelUpload", Command::SetFocus { .. } => "setFocus", Command::ManualRefresh => "manualRefresh", Command::ImportContact { .. } => "importContact", @@ -632,6 +651,20 @@ pub enum EngineError { /// A write-handle call named a handle this engine does not hold — never /// minted, or already committed, failed, or aborted. UnknownWriteHandle, + /// [`Command::CancelUpload`] named an op that is no longer cancellable: its + /// version's last block confirmed and its record is publishing, or it has + /// already left the durable queue. Never converted into a compensating + /// delete, which would substitute an irreversible published mutation (#824). + TooLateToCancel { + /// The op the cancel named. + op_id: OpId, + }, + /// [`Command::CancelUpload`] named a queued op that carries no upload. A + /// metadata op is undone by a compensating mutation, not a cancel. + NotAnUpload { + /// The op the cancel named. + op_id: OpId, + }, /// The file is past the flat-DAG ceiling: its root would inline more leaf /// links than a readable block can hold, so no device could ever serve it /// back. A format limit, not a budget verdict @@ -783,6 +816,16 @@ impl fmt::Display for EngineError { "the file changed while it was being read: {declared} bytes were declared and {observed} arrived" ), EngineError::UnknownWriteHandle => f.write_str("unknown write handle"), + EngineError::TooLateToCancel { op_id } => write!( + f, + "upload {} is already publishing and can no longer be cancelled", + op_id.0 + ), + EngineError::NotAnUpload { op_id } => write!( + f, + "queued op {} carries no upload to cancel; undo it with a compensating change", + op_id.0 + ), EngineError::ContentTooLarge { check } => write!( f, "this file is too large to store as a single version: [{check}]" @@ -1061,7 +1104,6 @@ struct LiveWrite { struct LiveWrites { ledger: StagingLedger, open: BTreeMap, - next: u64, } /// The re-point pointer-payload wire version the owner's own vault seals under @@ -1091,6 +1133,11 @@ pub struct Engine { /// only: a restart drops every reservation, and the blocks a dropped handle /// staged are unreferenced and collectible as orphans. writes: RefCell, + /// The staging keys those handles hold — orphan GC's live set, shared with + /// the tick loop that sweeps after each drain pass. + live_blocks: Rc>, + /// The upload-cancel interlock, shared with the drain the tick loop runs. + cancels: Rc>, /// The API base URL the liveness loop's [`ApiClient`] registers renewals /// against. Empty until the auth/config slice supplies it; the register-first /// renewal is a no-op against an empty base until then. @@ -1187,6 +1234,8 @@ impl Engine { storage_policy, content_profile, writes: RefCell::new(LiveWrites::default()), + live_blocks: Rc::new(RefCell::new(LiveBlocks::default())), + cancels: Rc::new(RefCell::new(UploadCancels::default())), api_base_url, gateway: gateway.into_gateway(), events, @@ -1331,6 +1380,11 @@ impl Engine { // ladder starts Fresh rather than Reconciling. self.sync_status.borrow_mut().last_success = Some(self.seams.scheduler.now()); + // A crash between staging a version's blocks and journaling its op + // leaves them referenced by nothing, so cold start is the first place + // that residue can be reclaimed (#828). + self.collect_staging_orphans().await; + self.spawn_liveness_loop(api.clone()); self.spawn_resolve_tick_loop(root_name, api.clone()); self.api = Some(api); @@ -1415,6 +1469,17 @@ impl Engine { .remove_op(*op_id) .await .map_err(ColdStartError::Seam)?; + // Its intent body never opened, so the version's only content + // key is gone and its blocks are ciphertext nothing can ever + // read: released under the proven-unopenable rule rather than + // preserved (#818, #853). + if let Some(root_cid) = raw + .iter() + .find(|(id, _)| id == op_id) + .and_then(|(_, record)| record_content_root_cid(record).ok().flatten()) + { + release_version_blocks(&self.seams.staging_store, &root_cid).await; + } } } @@ -1516,6 +1581,8 @@ impl Engine { let dead_letters = self.dead_letters.clone(); let blocked = self.blocked.clone(); let orphan_heads = self.orphan_heads.clone(); + let cancels = self.cancels.clone(); + let live_blocks = self.live_blocks.clone(); let transport = self.seams.record_transport.clone(); let snapshot_cache = self.seams.snapshot_cache.clone(); let floors = self.seams.floor_store.clone(); @@ -1637,6 +1704,7 @@ impl Engine { held: &held, blocked: &blocked, orphan_heads: &orphan_heads, + cancels: &cancels, events: &events, } .run(&DrainScope { @@ -1650,6 +1718,9 @@ impl Engine { .await; surface_drain_report(&events, &dead_letters, &report); } + // After the drain, so the pass's own removals are swept in the + // same tick rather than a cadence later (#828). + collect_orphans(&staging, &live_blocks).await; // `Adopted`/`Current` are the reconciled outcomes: both prove the // record plane answered with gate-passing state, so both stamp // the ladder's `last_success` (#33 D4). @@ -1767,6 +1838,7 @@ impl Engine { ); self.stage_and_notify(&op).await } + Command::CancelUpload { op_id } => self.cancel_upload(op_id).await.map(|()| None), Command::SetFocus { node } => { self.focus.borrow_mut().open_folder = node; // Navigation is the tick model's second trigger source (#33 D2): @@ -1887,8 +1959,7 @@ impl Engine { } }; - writes.next += 1; - let handle = WriteHandle(writes.next); + let handle = self.live_blocks.borrow_mut().open(); writes.open.insert( handle, LiveWrite { @@ -1951,11 +2022,8 @@ impl Engine { leaf }; if let Some(leaf) = leaf { - self.seams - .staging_store - .put_staged_bytes(&leaf.cid, &leaf.sealed) - .await - .map_err(EngineError::from_seam)?; + self.stage_handle_block(handle, &leaf.cid, &leaf.sealed) + .await?; } if rest.is_empty() { return Ok(()); @@ -1976,14 +2044,16 @@ impl Engine { // Taken out of the ledger up front: from here the handle is spent // whatever happens, and its reservation must not outlive it. let write = self.take_write(handle)?; - let mut staged = write.writer.staged_leaf_cids().to_vec(); - match self.commit_write_inner(write, &mut staged).await { + match self.commit_write_inner(handle, write).await { Ok(op_id) => { + // The journaled op now references the blocks, so GC no longer + // needs the handle to vouch for them. + self.live_blocks.borrow_mut().close(handle); let _ = self.events.unbounded_send(Event::SnapshotUpdated); Ok(op_id) } Err(error) => { - self.release_blocks(&staged).await; + self.release_handle_blocks(handle).await; Err(error) } } @@ -1992,10 +2062,10 @@ impl Engine { /// Abandon a write handle: release its reservation and the blocks it staged. /// Idempotent — an unknown handle is already gone. pub async fn abort_write(&mut self, handle: WriteHandle) { - let Ok(write) = self.take_write(handle) else { + if self.take_write(handle).is_err() { return; - }; - self.release_blocks(write.writer.staged_leaf_cids()).await; + } + self.release_handle_blocks(handle).await; } /// Remove a handle from the ledger, releasing its budget reservation. @@ -2009,20 +2079,35 @@ impl Engine { Ok(write) } - /// Drop staged blocks no op will ever reference. Best-effort: a failed - /// removal is orphan residue a later GC pass collects. - async fn release_blocks(&self, keys: &[Vec]) { + /// Stage one of a handle's blocks, recording its key as live **before** the + /// bytes land so an orphan-GC pass in the same turn cannot collect it. + async fn stage_handle_block( + &self, + handle: WriteHandle, + cid: &[u8], + sealed: &[u8], + ) -> Result<(), EngineError> { + self.live_blocks.borrow_mut().record(handle, cid); + self.seams + .staging_store + .put_staged_bytes(cid, sealed) + .await + .map_err(EngineError::from_seam) + } + + /// Drop every block a handle staged — no op will ever reference them. + /// Best-effort: a failed removal is orphan residue a later GC pass collects. + async fn release_handle_blocks(&self, handle: WriteHandle) { + let keys = self.live_blocks.borrow_mut().close(handle); for key in keys { - let _ = self.seams.staging_store.remove_staged_bytes(key).await; + let _ = self.seams.staging_store.remove_staged_bytes(&key).await; } } - /// `staged` accumulates every block this commit puts into the store, so a - /// failure after the tail or the root landed still releases them. async fn commit_write_inner( &self, + handle: WriteHandle, write: LiveWrite, - staged: &mut Vec>, ) -> Result { let LiveWrite { node, @@ -2043,20 +2128,12 @@ impl Engine { .map_err(seal_error)?; if let Some(tail) = &finished.tail { - staged.push(tail.cid.clone()); - self.seams - .staging_store - .put_staged_bytes(&tail.cid, &tail.sealed) - .await - .map_err(EngineError::from_seam)?; + self.stage_handle_block(handle, &tail.cid, &tail.sealed) + .await?; } let root_cid = finished.content.content_cid().to_vec(); - staged.push(root_cid.clone()); - self.seams - .staging_store - .put_staged_bytes(&root_cid, &finished.root_block) - .await - .map_err(EngineError::from_seam)?; + self.stage_handle_block(handle, &root_cid, &finished.root_block) + .await?; // The `{scope, epoch}` the key blob's AAD binds — see `seal_content_key` // for why they are values and not key inputs. @@ -2122,6 +2199,98 @@ impl Engine { Ok(nonce.nonce) } + /// Cancel one queued upload ([`Command::CancelUpload`]). + /// + /// Staged bytes are **released**, not preserved: the rule that splits the + /// two is whether the engine gave up on the op or the user did (#824). + async fn cancel_upload(&self, op_id: OpId) -> Result<(), EngineError> { + let queued = self.scan_queue().await?.mine; + let Some((_, op)) = queued.iter().find(|(id, _)| *id == op_id) else { + return Err(EngineError::TooLateToCancel { op_id }); + }; + let Some(root_cid) = op.content_root_cid().map(<[u8]>::to_vec) else { + return Err(EngineError::NotAnUpload { op_id }); + }; + // Claimed before anything is undone, and refused once the drain holds + // the op for publish — cancel never mutates published state. + if !self.cancels.borrow_mut().request(op_id) { + return Err(EngineError::TooLateToCancel { op_id }); + } + let node = op.target; + // A cancelled create takes every later queued op on the node it will + // never bring into being; a cancelled version takes nothing, since + // versions are independent full writes. + let cascade: Vec<(OpId, Op)> = match op.kind { + OpKind::Create { .. } => queued + .iter() + .filter(|(id, later)| *id > op_id && later.target == node) + .cloned() + .collect(), + _ => Vec::new(), + }; + + if let Err(error) = self.discard_upload(op_id, node, &root_cid).await { + self.cancels.borrow_mut().withdraw(op_id); + return Err(error); + } + for (later_id, later) in cascade { + match later.content_root_cid() { + Some(root_cid) => { + self.discard_upload(later_id, later.target, root_cid) + .await? + } + None => self.dequeue_op(later_id).await?, + } + } + let _ = self.events.unbounded_send(Event::SnapshotUpdated); + Ok(()) + } + + /// Undo one queued upload: retire what of it reached the network, drop the + /// op, release its blocks, and tell the host. + /// + /// Retiring first is what makes the leaf CIDs recoverable at all — the + /// staged root is the only record of them, and releasing the blocks destroys + /// it (#916). The retire is best-effort: a refused batch leaves pin rows + /// charged, which is a leak, where leaving the op queued after the user + /// cancelled it would be a broken guarantee. + async fn discard_upload( + &self, + op_id: OpId, + node: NodeId, + root_cid: &[u8], + ) -> Result<(), EngineError> { + if let Some(api) = &self.api { + let leaf_cids = version_leaf_cids(&self.seams.staging_store, root_cid).await; + let _ = retire(api, ®istry_cids(root_cid, &leaf_cids)).await; + } + // Dequeued before the blocks go, so no window leaves a durable op + // referencing content the store no longer holds. + self.dequeue_op(op_id).await?; + release_version_blocks(&self.seams.staging_store, root_cid).await; + let _ = self.events.unbounded_send(Event::OpProgress { + op_id: Some(op_id), + node, + phase: OpPhase::UploadCancelled, + progress: None, + error: None, + }); + Ok(()) + } + + async fn dequeue_op(&self, op_id: OpId) -> Result<(), EngineError> { + self.seams + .staging_store + .remove_op(op_id) + .await + .map_err(EngineError::from_seam) + } + + /// One orphan-GC pass over the staged-byte plane (#828). + async fn collect_staging_orphans(&self) { + collect_orphans(&self.seams.staging_store, &self.live_blocks).await; + } + /// A rendered read of the current state — the gate-passing base snapshot ⊕ /// the pending-op overlay — for FUSE-shaped reads (children/lookup/attrs/ /// statfs). Fails `NotStarted` before [`start`](Self::start). diff --git a/crates/engine/src/sync/cancel.rs b/crates/engine/src/sync/cancel.rs new file mode 100644 index 000000000..f00ba9207 --- /dev/null +++ b/crates/engine/src/sync/cancel.rs @@ -0,0 +1,94 @@ +//! The upload-cancel interlock shared by the facade and the drain (#824). +//! +//! Cancel is **guaranteed until publish entry and refused after**, so it can +//! never mutate published state. Both halves of that guarantee are decided here, +//! each in one borrow with no await inside it: either the facade claims the op +//! first and the drain abandons its upload, or the drain claims it first and the +//! facade refuses. There is no third outcome. + +use std::collections::BTreeSet; + +use crate::seams::OpId; + +/// Which uploads the user cancelled, and which one has passed publish entry. +#[derive(Default)] +pub(crate) struct UploadCancels { + /// Cancelled op ids. Held for the session: an op leaves the durable queue + /// with its cancel and never returns, so nothing here is ever reused. + cancelled: BTreeSet, + /// The op whose version has finished uploading and is now being published. + publishing: Option, +} + +impl UploadCancels { + /// Claim `op_id` for cancellation. `false` once its record is publishing — + /// the caller must refuse the cancel rather than compensate a published + /// mutation. + pub(crate) fn request(&mut self, op_id: OpId) -> bool { + if self.publishing == Some(op_id) { + return false; + } + self.cancelled.insert(op_id); + true + } + + /// Give the claim back, for a cancel that could not carry out its removals. + /// The op stays queued, so leaving it claimed would halt every pass behind + /// it forever. + pub(crate) fn withdraw(&mut self, op_id: OpId) { + self.cancelled.remove(&op_id); + } + + /// Claim `op_id` for publishing, now that every block of its version is on + /// the network. `false` if the user already cancelled it. + pub(crate) fn enter_publish(&mut self, op_id: OpId) -> bool { + if self.cancelled.contains(&op_id) { + return false; + } + self.publishing = Some(op_id); + true + } + + /// The op has left the drain's hands, published or not. + pub(crate) fn leave_publish(&mut self) { + self.publishing = None; + } + + pub(crate) fn is_cancelled(&self, op_id: OpId) -> bool { + self.cancelled.contains(&op_id) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_cancel_and_a_publish_entry_cannot_both_win_one_op() { + let mut cancels = UploadCancels::default(); + assert!(cancels.enter_publish(OpId(1))); + assert!( + !cancels.request(OpId(1)), + "the record is publishing; cancel must be refused" + ); + + cancels.leave_publish(); + let mut cancels = UploadCancels::default(); + assert!(cancels.request(OpId(1))); + assert!( + !cancels.enter_publish(OpId(1)), + "the user cancelled first; the publish must abandon" + ); + } + + /// The hold is per op: a cancel of a queued upload must not be refused + /// because a different op happens to be publishing. + #[test] + fn publishing_one_op_does_not_refuse_a_cancel_of_another() { + let mut cancels = UploadCancels::default(); + cancels.enter_publish(OpId(1)); + assert!(cancels.request(OpId(2))); + assert!(cancels.is_cancelled(OpId(2))); + assert!(!cancels.is_cancelled(OpId(1))); + } +} diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 9e93c2caa..79c8b96de 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -58,12 +58,14 @@ use crate::seams::{ StagingStore, }; use crate::session::SessionIdentity; +use crate::sync::cancel::UploadCancels; use crate::sync::model::{Snapshot, collation_key}; use crate::sync::op::{NewNode, Op, OpKind, StagedContent}; use crate::sync::overlay::apply_overlay; use crate::sync::project::project_folder; use crate::sync::rebase::{AppliedOp, DeadLetterReason, decode_queue, replay}; use crate::sync::record::RecordReader; +use crate::sync::staging::{preserve_staged_root, release_version_blocks}; /// The staging key holding the drained-op high-water mark: every op id at or /// below the stored value has left this device's queue (#860). @@ -189,6 +191,10 @@ enum Halt { /// resume probe must find room for. needed_bytes: u64, }, + /// The user cancelled the upload. The facade already retired what it had + /// uploaded, released its blocks and dequeued it, so the valve does nothing + /// but stop the pass (#824). + Cancelled, } /// The one verdict every unrecoverable-content path returns: the version's key, @@ -296,6 +302,8 @@ pub(crate) struct Drain<'a, T, H: Http, C: CredentialStore, F, S, St, Sch> { /// Head blocks this session's publishes orphaned, pending retirement /// ([`Drain::retire_orphan_heads`]). pub(crate) orphan_heads: &'a RefCell>, + /// The upload-cancel interlock, shared with the facade's cancel command. + pub(crate) cancels: &'a RefCell, /// The facade's outbound event stream, for upload progress. pub(crate) events: &'a mpsc::UnboundedSender, } @@ -465,6 +473,13 @@ where let Some((_, op)) = queued.iter().find(|(id, _)| id == op_id) else { continue; }; + // A terminally unrebasable op keeps its staged bytes, and this is + // what keeps them reachable once its op record is gone (#853). + if let Some(root_cid) = op.content_root_cid() { + preserve_staged_root(self.staging, root_cid) + .await + .map_err(seam)?; + } self.abandon(scope, *op_id, op).await?; report.dead_letters.push((*op_id, op.target, *reason)); } @@ -477,10 +492,11 @@ where } for applied in &rebased.applied { - if let Err(halt) = self + let outcome = self .publish_applied(scope, &mut pass, applied, &rebased.rebased) - .await - { + .await; + self.cancels.borrow_mut().leave_publish(); + if let Err(halt) = outcome { self.apply_valve(scope, applied.op_id, &applied.op, halt, attempts, report) .await; return Err(halt); @@ -502,7 +518,7 @@ where report: &mut DrainReport, ) { match halt { - Halt::Unclassified => {} + Halt::Unclassified | Halt::Cancelled => {} Halt::Attempt | Halt::UploadAttempt => { if attempts.charge(op_id) < ATTEMPT_BUDGET { return; @@ -1291,19 +1307,35 @@ where /// One version's blocks, uploaded and pinned, with the transfer's progress /// reported on the event stream throughout. + /// + /// Publish entry — the point past which a cancel is refused — is the moment + /// the last block confirms: everything after it authors and publishes the + /// version's record with no further block boundary to stop at (#824). async fn upload_version( &self, scope: &DrainScope<'_>, applied: &AppliedOp, staged: &StagedContent, ) -> Result { - let uploaded = self.upload_blocks(scope, applied, staged).await; - if let Err(halt) = &uploaded - && let Some(error) = upload_failure(*halt) - { - self.emit_upload(applied, OpPhase::UploadFailed, None, Some(error)); + let uploaded = match self.upload_blocks(scope, applied, staged).await { + Ok(uploaded) => uploaded, + Err(halt) => { + // A cancel that landed inside one of the loop's awaits released + // this version's blocks, so the halt it reported is that + // cancel's shadow, not a failure of the upload. + if self.cancels.borrow().is_cancelled(applied.op_id) { + return Err(Halt::Cancelled); + } + if let Some(error) = upload_failure(halt) { + self.emit_upload(applied, OpPhase::UploadFailed, None, Some(error)); + } + return Err(halt); + } + }; + if !self.cancels.borrow_mut().enter_publish(applied.op_id) { + return Err(Halt::Cancelled); } - uploaded + Ok(uploaded) } /// One version's blocks, uploaded and pinned: the `Version` its record @@ -1360,6 +1392,13 @@ where }; emit(OpPhase::UploadStarted, blocks(uploaded)); for (index, leaf_cid) in content.leaf_cids().iter().enumerate() { + // The block boundary a cancel gets to run at. Without it a whole + // version uploads inside one turn of the host's executor and the + // cancel guarantee collapses to "only before the op starts" (#824). + yield_now().await; + if self.cancels.borrow().is_cancelled(applied.op_id) { + return Err(Halt::Cancelled); + } match self.staged_block(leaf_cid).await? { Some(block) => { self.upload_block(leaf_cid, &block).await?; @@ -1382,6 +1421,10 @@ where None => {} } } + yield_now().await; + if self.cancels.borrow().is_cancelled(applied.op_id) { + return Err(Halt::Cancelled); + } // The root goes up last and stays staged until the publish confirms: it // is the manifest every retry re-derives the plan from, so releasing it // before the record lands would strand a fully-uploaded version. @@ -1495,15 +1538,10 @@ where /// Best-effort: a failed removal is orphan residue a later GC pass collects, /// never a reason to fail a landed publish. async fn release_staged_blocks(&self, op: &Op) { - let Some(staged) = op.staged_content() else { + let Some(root_cid) = op.content_root_cid() else { return; }; - if let Some(content) = self.staged_manifest(&staged.root_cid).await { - for leaf_cid in content.leaf_cids() { - let _ = self.staging.remove_staged_bytes(leaf_cid).await; - } - } - let _ = self.staging.remove_staged_bytes(&staged.root_cid).await; + release_version_blocks(self.staging, root_cid).await; } /// The parent a node is published under, from the base the pass repaints as @@ -1931,6 +1969,22 @@ fn seam(_: crate::seams::SeamError) -> Halt { Halt::Unclassified } +/// Hand control back to the host's executor once, so a facade command queued +/// behind this task gets a turn. The engine runs pinned to one execution +/// context, so a long await-free stretch is one the host cannot interrupt. +async fn yield_now() { + let mut yielded = false; + core::future::poll_fn(move |cx| { + if yielded { + return core::task::Poll::Ready(()); + } + yielded = true; + cx.waker().wake_by_ref(); + core::task::Poll::Pending + }) + .await; +} + /// Classify a publish failure for the valve. Only the head-block upload and the /// register-first call carry a server verdict this pass can act on; everything /// else is availability. @@ -2023,7 +2077,8 @@ fn blocks(count: usize) -> u32 { /// its reservation, and the host reads it from `SnapshotView::blocked` (#841). fn upload_failure(halt: Halt) -> Option<&'static str> { match halt { - Halt::Blocked { .. } => None, + // A cancel reports `UploadCancelled` from the facade that ordered it. + Halt::Blocked { .. } | Halt::Cancelled => None, Halt::Unclassified => Some("the upload did not complete"), // Both charge the attempt budget; which one it is decides only what // exhausting that budget retires, not what the host is told. @@ -2042,7 +2097,7 @@ fn upload_failure(halt: Halt) -> Option<&'static str> { /// batch names exactly what a register batch claimed — and every block is its /// own accountable pin row (blueprint/api.md "Pin/name registry"), so leaving /// a leaf out of a retirement spends account quota forever. -fn registry_cids(root_cid: &[u8], leaf_cids: &[Vec]) -> Vec { +pub(crate) fn registry_cids(root_cid: &[u8], leaf_cids: &[Vec]) -> Vec { core::iter::once(root_cid) .chain(leaf_cids.iter().map(Vec::as_slice)) .map(encode_content_cid_str) diff --git a/crates/engine/src/sync/mod.rs b/crates/engine/src/sync/mod.rs index a29fbcf56..854a065ee 100644 --- a/crates/engine/src/sync/mod.rs +++ b/crates/engine/src/sync/mod.rs @@ -17,6 +17,7 @@ //! primitives themselves land with the rotation slice. pub mod boot; +pub(crate) mod cancel; pub(crate) mod drain; pub mod model; pub mod op; @@ -47,7 +48,10 @@ pub use record::{ OpRecordError, RecordClass, RecordReader, RecordSeal, RetainedReason, encode_op_record, record_content_root_cid, }; -pub use staging::{orphan_staging_keys, stage_op}; +pub use staging::{ + PRESERVED_ROOTS_KEY, orphan_staging_keys, preserve_staged_root, release_version_blocks, + stage_op, +}; pub use staleness::{Connectivity, classify, withheld_escalation}; pub use tick::{ FocusTarget, FocusWindow, ResolveMode, TickCause, TickControl, focus_folders, diff --git a/crates/engine/src/sync/staging.rs b/crates/engine/src/sync/staging.rs index d3d109f21..3b3ca6c48 100644 --- a/crates/engine/src/sync/staging.rs +++ b/crates/engine/src/sync/staging.rs @@ -10,9 +10,12 @@ //! ([`crate::content::StagingLedger`]); this module owns the journal entry and //! the staged-byte hygiene that outlives it. +use std::collections::{BTreeMap, HashSet}; + use cipherbox_core::content::verify_cid; use crate::content::decode_root; +use crate::facade::WriteHandle; use crate::seams::{OpId, SeamError, SeamResult, StagingStore}; use crate::sync::drain::{DRAINED_OP_MARK_KEY, OP_ATTEMPTS_KEY, UPLOAD_MARK_KEY}; use crate::sync::op::Op; @@ -47,38 +50,247 @@ pub async fn stage_op( store.enqueue_op(&record).await } +/// The staging key holding the roots of dead-lettered versions whose bytes are +/// **preserved**: the op record that referenced them is gone from the queue, and +/// this record is what keeps orphan GC off them (#853). +/// +/// It lives in the staging store for the same reason the drain's other +/// bookkeeping does — the roots and the blocks they name share one durability +/// domain — and [`orphan_staging_keys`] treats it as referenced. +pub const PRESERVED_ROOTS_KEY: &[u8] = b"cipherbox/preserved-roots"; + +/// The preserved-roots record's format tag. The staging store is shared with +/// whatever build wrote it, so bytes that merely happen to be well-shaped must +/// not parse as roots. +const PRESERVED_FORMAT_V1: u8 = 1; + +/// Drop every staged block of one version: the leaves its root manifest lists, +/// in file order, then the root itself. +/// +/// Staged bytes are released on exactly three verdicts — the user cancelled the +/// upload, the version is proven unopenable (#818), or its staged root cannot be +/// expanded — and preserved on the fourth, a terminally unrebasable op +/// (blueprint/engine.md; #33 D6). File order keeps the blocks that remain a +/// suffix at every step, which is the invariant the drain's resume reads. +/// +/// Best-effort: a failed removal is orphan residue a later GC pass collects. +pub async fn release_version_blocks(store: &S, root_cid: &[u8]) { + for leaf_cid in version_leaf_cids(store, root_cid).await { + let _ = store.remove_staged_bytes(&leaf_cid).await; + } + let _ = store.remove_staged_bytes(root_cid).await; +} + +/// The leaves a staged root manifest lists, in file order. Empty when the root +/// is gone, fails its own CID, or does not decode — every caller is a +/// reconciliation path that must still make progress on a store that has lost +/// bytes. +pub async fn version_leaf_cids(store: &S, root_cid: &[u8]) -> Vec> { + let Ok(Some(block)) = store.staged_bytes(root_cid).await else { + return Vec::new(); + }; + if verify_cid(root_cid, &block).is_err() { + return Vec::new(); + } + decode_root(&block).map(|m| m.leaf_cids).unwrap_or_default() +} + +/// Record `root_cid` as a preserved dead letter, so orphan GC keeps its blocks +/// once the op that referenced them leaves the durable queue. +/// +/// Fails closed on a preserved record this build cannot read: overwriting it +/// would drop the roots it already names, and those blocks are exactly what the +/// dead-letter contract promises to keep. +pub async fn preserve_staged_root(store: &S, root_cid: &[u8]) -> SeamResult<()> { + let mut roots = read_preserved_roots(store) + .await? + .ok_or_else(|| SeamError::new("preserve_staged_root: unreadable preserved-roots record"))?; + if roots.iter().any(|root| root == root_cid) { + return Ok(()); + } + roots.push(root_cid.to_vec()); + write_preserved_roots(store, &roots).await +} + +/// The preserved roots the store holds. `None` when the record is present but +/// not one this build wrote — the fail-safe direction is to preserve, so a +/// caller must freeze rather than treat it as empty. +async fn read_preserved_roots(store: &S) -> SeamResult>>> { + let Some(stored) = store.staged_bytes(PRESERVED_ROOTS_KEY).await? else { + return Ok(Some(Vec::new())); + }; + let Some(mut rest) = stored.strip_prefix(&[PRESERVED_FORMAT_V1]) else { + return Ok(None); + }; + let mut roots = Vec::new(); + while let Some((&len, tail)) = rest.split_first() { + let Some((cid, next)) = tail.split_at_checked(len as usize) else { + return Ok(None); + }; + roots.push(cid.to_vec()); + rest = next; + } + Ok(Some(roots)) +} + +/// Fails closed on a root too long to length-prefix: writing a truncated record +/// would silently unpin every root behind it (AGENTS.md rule 8). +/// +/// An empty list removes the key rather than storing a tag-only record, so a +/// device that has never dead-lettered spends no staging budget on this. +async fn write_preserved_roots(store: &S, roots: &[Vec]) -> SeamResult<()> { + if roots.is_empty() { + return store.remove_staged_bytes(PRESERVED_ROOTS_KEY).await; + } + let mut bytes = vec![PRESERVED_FORMAT_V1]; + for root in roots { + let len = u8::try_from(root.len()) + .map_err(|_| SeamError::new("preserved root is not length-prefixable"))?; + bytes.push(len); + bytes.extend_from_slice(root); + } + store.put_staged_bytes(PRESERVED_ROOTS_KEY, &bytes).await +} + +/// The staging keys write handles hold outside the durable queue. +/// +/// A handle stages every block under its own content address before any op +/// references it, so nothing in the queue can vouch for those blocks and orphan +/// GC would otherwise collect a version mid-write. The handle records each key +/// here **before** the bytes are staged, and keeps them until its op is +/// journaled or its blocks are released. +#[derive(Default)] +pub(crate) struct LiveBlocks { + by_handle: BTreeMap>>, + /// Handles opened so far — the handle-id source, and the evidence a GC sweep + /// reads to tell that no handle opened while it was running. + opened: u64, +} + +impl LiveBlocks { + /// Mint the next handle and start holding its staging keys. + pub(crate) fn open(&mut self) -> WriteHandle { + self.opened += 1; + let handle = WriteHandle(self.opened); + self.by_handle.insert(handle, Vec::new()); + handle + } + + /// Hold one more staging key for `handle`, before its bytes are staged. + pub(crate) fn record(&mut self, handle: WriteHandle, key: &[u8]) { + if let Some(keys) = self.by_handle.get_mut(&handle) { + keys.push(key.to_vec()); + } + } + + /// Stop holding `handle`'s keys, returning them so a caller that is + /// abandoning the write can release the blocks. + pub(crate) fn close(&mut self, handle: WriteHandle) -> Vec> { + self.by_handle.remove(&handle).unwrap_or_default() + } + + /// Every key currently held, across all open handles. + pub(crate) fn keys(&self) -> Vec> { + self.by_handle.values().flatten().cloned().collect() + } + + /// How many handles have ever been opened. + pub(crate) fn opened(&self) -> u64 { + self.opened + } +} + +/// One orphan-GC pass: expand every staged root into its leaf set and remove the +/// blocks nothing references (#828). Runs at cold start and after each drain +/// pass. +/// +/// Best-effort throughout — a store that cannot enumerate or remove leaves its +/// residue for the next pass — and it also prunes [`PRESERVED_ROOTS_KEY`] of +/// roots whose bytes are already gone, so that record cannot grow without bound. +pub(crate) async fn collect_orphans( + store: &S, + live: &core::cell::RefCell, +) { + let opened_before = live.borrow().opened(); + let live_keys = live.borrow().keys(); + let Ok(orphans) = orphan_staging_keys(store, &live_keys).await else { + return; + }; + for key in orphans { + // A handle opened after the scan may be staging exactly these bytes. + if live.borrow().opened() != opened_before { + return; + } + let _ = store.remove_staged_bytes(&key).await; + } + prune_preserved_roots(store).await; +} + +/// Drop preserved roots whose blocks the store no longer holds. +async fn prune_preserved_roots(store: &S) { + let Ok(Some(roots)) = read_preserved_roots(store).await else { + return; + }; + let mut live = Vec::with_capacity(roots.len()); + for root in &roots { + if matches!(store.staged_bytes(root).await, Ok(Some(_))) { + live.push(root.clone()); + } + } + // An unchanged record is not rewritten, so an idle sweep makes no store + // write. + if live.len() != roots.len() { + let _ = write_preserved_roots(store, &live).await; + } +} + /// Staging keys held by the store that nothing references — orphan residue from /// a superseded or abandoned upload, safe to GC (#33 D6 staged-bytes hygiene). /// -/// Two things reference a block. A **queued record**'s content root rides its +/// Three things reference a block. A **queued record**'s content root rides its /// clear header, and since a version stages one block per key, a root is /// expanded into the leaf keys its own manifest lists — so a foreign account's -/// or a forward-version record's whole block set is retained. An **open write -/// handle**'s leaves are staged before any op is journaled, so they are -/// unreferenced by construction and must be passed in as `live`; collecting them -/// mid-write would publish a version whose manifest names blocks nothing holds. +/// or a forward-version record's whole block set is retained. A **preserved +/// dead letter**'s root is expanded the same way, which is what keeps the bytes +/// the dead-letter contract promises once its op record is gone (#853). An +/// **open write handle**'s blocks are staged before any op is journaled, so they +/// are unreferenced by construction and must be passed in as `live`; collecting +/// them mid-write would publish a version whose manifest names blocks nothing +/// holds. /// -/// Fail-closed: an unreadable queue entry, or a referenced root the store cannot -/// produce or this build cannot decode, classes **nothing** an orphan. +/// Fail-closed: an unreadable queue entry, an unreadable preserved-roots record, +/// or a referenced root the store cannot produce or this build cannot decode, +/// classes **nothing** an orphan. A root that cannot be expanded therefore +/// freezes the whole pass — self-clearing, because the drain classifies that +/// same root permanent and releases the version. pub async fn orphan_staging_keys( store: &S, live: &[Vec], ) -> SeamResult>> { let queued = store.queued_ops().await?; // The drain's own queue bookkeeping is not upload residue. - let mut referenced = std::collections::HashSet::from([ + let mut referenced = HashSet::from([ DRAINED_OP_MARK_KEY.to_vec(), OP_ATTEMPTS_KEY.to_vec(), UPLOAD_MARK_KEY.to_vec(), + PRESERVED_ROOTS_KEY.to_vec(), ]); referenced.extend(live.iter().cloned()); + let Some(preserved) = read_preserved_roots(store).await? else { + return Ok(Vec::new()); + }; + let mut roots = preserved; for (_, record) in &queued { let Ok(root) = record_content_root_cid(record) else { // An unreadable record may still reference staged bytes, and its // root is unknowable. return Ok(Vec::new()); }; - let Some(root) = root else { continue }; + if let Some(root) = root { + roots.push(root); + } + } + for root in roots { // The drain removes each block as it uploads, so a root whose bytes are // gone is a finished upload with nothing left to expand. if let Some(block) = store.staged_bytes(&root).await? { @@ -397,6 +609,102 @@ mod tests { }); } + /// The dead-letter contract keeps a terminally unrebasable op's staged + /// bytes, but the abandonment removes its op record — so without a second + /// reference source GC reclaims exactly what was promised (#853). + #[test] + fn a_preserved_dead_letters_block_set_is_never_collected() { + let store = InMemoryStagingStore::default(); + block_on(async { + let (blocks, root_block, staged) = framed(b"forty bytes of content ------------------"); + put_blocks(&store, &blocks, &root_block, &staged).await; + preserve_staged_root(&store, &staged.root_cid) + .await + .unwrap(); + store.put_staged_bytes(b"orphan", b"stale").await.unwrap(); + + assert_eq!( + orphan_staging_keys(&store, &[]).await.unwrap(), + vec![b"orphan".to_vec()], + "the preserved root and every leaf it lists stay referenced" + ); + }); + } + + /// The fail-safe direction here is to preserve, so a preserved-roots record + /// this build cannot read must freeze the pass rather than read as empty. + #[test] + fn an_unreadable_preserved_roots_record_makes_orphan_gc_conservative() { + let store = InMemoryStagingStore::default(); + block_on(async { + store + .put_staged_bytes(PRESERVED_ROOTS_KEY, b"not a preserved-roots record") + .await + .unwrap(); + store.put_staged_bytes(b"orphan", b"stale").await.unwrap(); + + assert!(orphan_staging_keys(&store, &[]).await.unwrap().is_empty()); + assert!( + preserve_staged_root(&store, b"root").await.is_err(), + "overwriting it would drop the roots it already names" + ); + }); + } + + #[test] + fn preserved_roots_round_trip_and_prune_to_the_blocks_the_store_still_holds() { + let store = InMemoryStagingStore::default(); + block_on(async { + let (blocks, root_block, staged) = framed(b"forty bytes of content ------------------"); + put_blocks(&store, &blocks, &root_block, &staged).await; + preserve_staged_root(&store, &staged.root_cid) + .await + .unwrap(); + preserve_staged_root(&store, b"gone").await.unwrap(); + assert_eq!( + read_preserved_roots(&store).await.unwrap().unwrap(), + vec![staged.root_cid.clone(), b"gone".to_vec()] + ); + + prune_preserved_roots(&store).await; + assert_eq!( + read_preserved_roots(&store).await.unwrap().unwrap(), + vec![staged.root_cid.clone()], + "a root whose blocks are gone preserves nothing" + ); + + release_version_blocks(&store, &staged.root_cid).await; + prune_preserved_roots(&store).await; + assert!( + store + .staged_bytes(PRESERVED_ROOTS_KEY) + .await + .unwrap() + .is_none(), + "an empty list spends no staging budget" + ); + }); + } + + /// A release drops the whole set — every leaf the manifest lists, then the + /// root — so an abandoned version holds no budget. + #[test] + fn releasing_a_version_drops_every_block_of_it() { + let store = InMemoryStagingStore::default(); + block_on(async { + let (blocks, root_block, staged) = framed(b"forty bytes of content ------------------"); + put_blocks(&store, &blocks, &root_block, &staged).await; + store.put_staged_bytes(b"other", b"kept").await.unwrap(); + + release_version_blocks(&store, &staged.root_cid).await; + assert_eq!( + store.staged_keys().await.unwrap(), + vec![b"other".to_vec()], + "the version's whole block set goes, and nothing else" + ); + }); + } + /// A referenced root this build cannot decode hides an unknowable leaf set, /// so nothing may be classed an orphan against it. #[test] diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 4d62119e2..16d7ea1dc 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -374,11 +374,41 @@ fn secret() -> LoginSecret { LoginSecret::new(SECRET.to_vec()) } -/// Poll every spawned loop once with a no-op waker (the loops never yield -/// inside a pass over the synchronous fakes). +/// A waker that only records that it fired — enough to tell a cooperative +/// yield (which wakes itself) from a parked sleep (which does not). +struct WokenFlag(Mutex); + +impl std::task::Wake for WokenFlag { + fn wake(self: Arc) { + self.wake_by_ref(); + } + + fn wake_by_ref(self: &Arc) { + *self.0.lock().expect("lock") = true; + } +} + +/// Poll every spawned loop until each is parked on a timer rather than on the +/// drain's block-boundary yield, and report the last round's verdicts. fn poll_each(tasks: &mut [BoxedTask]) -> Vec> { + let flag = Arc::new(WokenFlag(Mutex::new(false))); + let waker = Waker::from(flag.clone()); + let mut cx = Context::from_waker(&waker); + loop { + *flag.0.lock().expect("lock") = false; + let polls: Vec<_> = tasks.iter_mut().map(|t| t.as_mut().poll(&mut cx)).collect(); + if !*flag.0.lock().expect("lock") { + return polls; + } + } +} + +/// Poll every spawned loop exactly once, leaving a yielded drain mid-pass. +fn poll_once(tasks: &mut [BoxedTask]) { let mut cx = Context::from_waker(Waker::noop()); - tasks.iter_mut().map(|t| t.as_mut().poll(&mut cx)).collect() + for task in tasks.iter_mut() { + let _ = task.as_mut().poll(&mut cx); + } } /// Run one resolve-tick interval, which is also one drain pass. @@ -3526,6 +3556,390 @@ fn a_publish_that_reached_the_transport_never_retires_its_head() { ); } +// --------------------------------------------------------------------------- +// Cancel, and the staged-byte lifetime around it (#824, #828, #853). +// --------------------------------------------------------------------------- + +/// The version an op has staged: its root first, then every leaf in file order. +fn queued_version(device: &FakeDevice, op_id: OpId) -> Vec> { + block_on(async { + let queued = device.staging_store.queued_ops().await.unwrap(); + let record = &queued + .iter() + .find(|(id, _)| *id == op_id) + .expect("the op is queued") + .1; + let root_cid = record_content_root_cid(record).unwrap().unwrap(); + let root_block = device + .staging_store + .staged_bytes(&root_cid) + .await + .unwrap() + .unwrap(); + core::iter::once(root_cid) + .chain(decode_root(&root_block).unwrap().leaf_cids) + .collect() + }) +} + +/// The acceptance case: a cancel that lands while the drain is mid-upload stops +/// it at the next block boundary, releases every block of the version, retires +/// what already reached the network, and publishes nothing. +#[test] +fn a_cancel_mid_upload_releases_every_block_and_returns_the_staging_budget() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, mut events, mut tasks) = boot(&world, &blocks, &alice, 42); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + &(0..200u8).collect::>(), + ) + .expect("the write commits"); + let version = queued_version(&alice, op_id); + let file = block_on(engine.view()).unwrap().children(ROOT)[0].id; + + // Each poll resumes the drain at its next block boundary, so a handful of + // them leave it parked with part of the version already on the network. + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + assert!( + uploads(&alice) > 0, + "the cancel must land mid-transfer, not before it started" + ); + + block_on(engine.command(Command::CancelUpload { op_id })).expect("the upload is cancellable"); + poll_each(&mut tasks); + + assert_no_blocks_staged(&alice, &version); + assert!( + block_on(alice.staging_store.staged_keys()) + .unwrap() + .iter() + .all(|key| key.as_slice() == DRAINED_OP_MARK_KEY || key.as_slice() == UPLOAD_MARK_KEY), + "the staging budget holds nothing but queue bookkeeping" + ); + assert!( + block_on(alice.staging_store.queued_ops()) + .unwrap() + .is_empty(), + "the cancelled op left the durable queue" + ); + assert!( + block_on(engine.view()).unwrap().children(ROOT).is_empty(), + "nothing published" + ); + let retired = retire_targets(&alice); + assert!( + version + .iter() + .all(|cid| retired.contains(&encode_content_cid_str(cid))), + "every block the version could have charged is retired, root and leaves" + ); + assert!( + events_so_far(&mut events).contains(&Event::OpProgress { + op_id: Some(op_id), + node: file, + phase: OpPhase::UploadCancelled, + progress: None, + error: None, + }), + "the host is told the upload was cancelled, keyed on its own op" + ); +} + +/// Cancel is guaranteed only until publish entry. Once the version's record has +/// published, the op has left the queue and a cancel is refused rather than +/// converted into a compensating delete of published state. +#[test] +fn a_cancel_after_the_version_published_is_refused() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + b"published bytes", + ) + .unwrap(); + tick(&world, &engine, &mut tasks); + + assert_eq!( + block_on(engine.command(Command::CancelUpload { op_id })), + Err(EngineError::TooLateToCancel { op_id }) + ); + assert_eq!( + block_on(engine.view()).unwrap().children(ROOT).len(), + 1, + "the published file is untouched" + ); + assert!( + retire_targets(&alice).is_empty(), + "a refused cancel unpins nothing" + ); +} + +/// Cancel is content-only: a metadata op is undone by a compensating mutation, +/// which costs neither the network nor the staging budget. +#[test] +fn a_cancel_of_a_metadata_op_is_refused() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, _tasks) = boot(&world, &blocks, &alice, 42); + let op_id = block_on(engine.command(Command::Create { + parent: ROOT, + name: "folder".into(), + kind: NodeKind::Folder, + })) + .unwrap() + .expect("the create queues"); + + assert_eq!( + block_on(engine.command(Command::CancelUpload { op_id })), + Err(EngineError::NotAnUpload { op_id }) + ); + assert_eq!( + block_on(alice.staging_store.queued_ops()).unwrap().len(), + 1, + "the refused cancel left the op queued" + ); +} + +/// A cancelled create takes every later queued op on the node it will never +/// bring into being; a cancelled version takes nothing, since versions are +/// independent full writes. +#[test] +fn a_cancelled_create_cascades_onto_its_node_and_a_cancelled_version_does_not() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + + // A landed file, so a later version of it has something to update. + write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "kept.bin".into(), + }, + b"kept bytes", + ) + .unwrap(); + tick(&world, &engine, &mut tasks); + let kept = child_id(&engine, ROOT, "kept.bin"); + + let create = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "doomed.bin".into(), + }, + b"doomed bytes", + ) + .unwrap(); + let doomed = child_id(&engine, ROOT, "doomed.bin"); + block_on(engine.command(Command::Rename { + node: doomed, + new_name: "renamed.bin".into(), + })) + .unwrap(); + let version = write_file( + &mut engine, + WriteTarget::Version { node: kept }, + b"a new version", + ) + .unwrap(); + + block_on(engine.command(Command::CancelUpload { op_id: create })).expect("the create cancels"); + let queued: Vec = block_on(alice.staging_store.queued_ops()) + .unwrap() + .into_iter() + .map(|(op_id, _)| op_id) + .collect(); + assert_eq!( + queued, + vec![version], + "the rename of the cancelled node went with it; the unrelated version stayed" + ); + + block_on(engine.command(Command::CancelUpload { op_id: version })) + .expect("the version cancels"); + assert!( + block_on(alice.staging_store.queued_ops()) + .unwrap() + .is_empty() + ); + tick(&world, &engine, &mut tasks); + assert_eq!( + published_names(&world.record_store, &blocks, ROOT), + vec!["kept.bin".to_owned()], + "neither cancelled op published" + ); +} + +/// A cancel that cannot carry out its removals must give the claim back: an op +/// left both queued and claimed would halt every pass behind it forever. +#[test] +fn a_cancel_that_cannot_dequeue_leaves_the_op_publishable() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + b"bytes that still publish", + ) + .unwrap(); + + alice.staging_store.fail_remove_op(); + assert!( + block_on(engine.command(Command::CancelUpload { op_id })).is_err(), + "the cancel could not remove the op, so it did not happen" + ); + + tick(&world, &engine, &mut tasks); + assert_eq!( + published_names(&world.record_store, &blocks, ROOT), + vec!["photo.bin".to_owned()], + "the op the cancel could not take is published, not wedged" + ); +} + +/// Orphan GC runs after each drain pass and reclaims blocks nothing references — +/// the residue of a crash between staging a version and journaling its op. +#[test] +fn orphan_residue_is_collected_and_a_live_handles_blocks_are_not() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let (leaves, root_block, root_cid) = frame_version(&(0..40u8).collect::>()); + stage_blocks(&alice, &leaves, &root_block, &root_cid); + + // A write handle mid-stream: its blocks are staged before any op references + // them, so only the live set keeps GC off them. + let handle = block_on(engine.begin_write( + WriteTarget::NewFile { + parent: ROOT, + name: "in-flight.bin".into(), + }, + 200, + )) + .unwrap(); + let plaintext: Vec = (0..200u8).collect(); + block_on(engine.push_chunk(handle, &plaintext[..64])).unwrap(); + + tick(&world, &engine, &mut tasks); + + let staged = block_on(alice.staging_store.staged_keys()).unwrap(); + for orphan in leaves.iter().map(|leaf| leaf.cid.clone()).chain([root_cid]) { + assert!( + !staged.contains(&orphan), + "unreferenced residue is collected" + ); + } + + // The handle finishing and publishing is the only assertion that proves the + // sweep left its blocks alone: a collected leaf fails the drain, not this. + block_on(engine.push_chunk(handle, &plaintext[64..])).unwrap(); + block_on(engine.commit_write(handle)).expect("the handle still holds every block it staged"); + tick(&world, &engine, &mut tasks); + let file = child_id(&engine, ROOT, "in-flight.bin"); + assert_eq!( + block_on(engine.read_content(file)).expect("the published version reads back"), + plaintext + ); +} + +/// A terminally unrebasable op keeps its staged bytes — and keeping them is only +/// real if they survive the cold start that drops the op record, and the GC pass +/// that runs there (#853). +#[test] +fn a_dead_lettered_ops_blocks_survive_a_cold_start_and_a_gc_pass() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (engine, mut events, mut tasks) = boot(&world, &blocks, &alice, 42); + // A version of a node no gate-passing state holds: terminally unrebasable. + let (leaves, root_block, root_cid) = frame_version(&(0..40u8).collect::>()); + stage_blocks(&alice, &leaves, &root_block, &root_cid); + stage( + &alice, + &Op::update_content( + NodeId([0xAB; 16]), + StagedContent { + root_cid: root_cid.clone(), + plaintext_size: 40, + sealed_content_key: b"never opened".to_vec(), + epoch: EPOCH, + }, + 1, + UnixMillis(4_242), + ), + Some(&root_block), + ); + tick(&world, &engine, &mut tasks); + + assert!( + events_so_far(&mut events).iter().any(|event| matches!( + event, + Event::DeadLetter { + reason: DeadLetterReason::TargetGone, + .. + } + )), + "the op is terminally unrebasable, not unrecoverable content" + ); + let version: Vec> = leaves + .iter() + .map(|leaf| leaf.cid.clone()) + .chain([root_cid]) + .collect(); + let after_drain = block_on(alice.staging_store.staged_keys()).unwrap(); + assert!( + version.iter().all(|cid| after_drain.contains(cid)), + "a dead letter preserves its staged bytes" + ); + drop(engine); + + let (engine, _events, mut tasks) = boot(&world, &blocks, &alice, 43); + tick(&world, &engine, &mut tasks); + let after_restart = block_on(alice.staging_store.staged_keys()).unwrap(); + assert!( + version.iter().all(|cid| after_restart.contains(cid)), + "and keeps them across the cold start that removed the op record" + ); +} + /// The upload a refusal lands on to halt a 200-byte version mid-set: past the /// first leaves, well short of the 13 the CI framing produces. const MID_SET_UPLOAD: usize = 8; diff --git a/crates/fuse/src/error.rs b/crates/fuse/src/error.rs index c2eff9c2e..f58851105 100644 --- a/crates/fuse/src/error.rs +++ b/crates/fuse/src/error.rs @@ -84,6 +84,8 @@ impl From for VfsError { | EngineError::ContentSizeMismatch { .. } | EngineError::UnknownWriteHandle | EngineError::ContentKeySealFailed { .. } + | EngineError::TooLateToCancel { .. } + | EngineError::NotAnUpload { .. } | EngineError::Unimplemented { .. }) => VfsError::Internal { message: error.to_string(), }, diff --git a/crates/wasm/src/host.rs b/crates/wasm/src/host.rs index 1b0f8839b..c66f8b49b 100644 --- a/crates/wasm/src/host.rs +++ b/crates/wasm/src/host.rs @@ -419,6 +419,8 @@ fn engine_error(error: EngineError) -> JsValue { EngineError::OverBudget { .. } => "overBudget", EngineError::ContentSizeMismatch { .. } => "contentSizeMismatch", EngineError::UnknownWriteHandle => "unknownWriteHandle", + EngineError::TooLateToCancel { .. } => "tooLateToCancel", + EngineError::NotAnUpload { .. } => "notAnUpload", EngineError::ContentTooLarge { .. } => "contentTooLarge", EngineError::ContentKeySealFailed { .. } => "contentKeySealFailed", EngineError::Seam { .. } => "seam", diff --git a/crates/wasm/src/lib.rs b/crates/wasm/src/lib.rs index 132b3b814..58e024217 100644 --- a/crates/wasm/src/lib.rs +++ b/crates/wasm/src/lib.rs @@ -531,6 +531,15 @@ impl Command { }) } + /// Cancel a queued upload by the op id `commitWrite` returned. Rejects with + /// `tooLateToCancel` once the version's record is publishing. + #[wasm_bindgen(js_name = cancelUpload)] + pub fn cancel_upload(op_id: u64) -> Command { + Self::wrap(facade::Command::CancelUpload { + op_id: cipherbox_engine::seams::OpId(op_id), + }) + } + /// Set the open folder driving the focus window (`undefined` clears it). #[wasm_bindgen(js_name = setFocus)] pub fn set_focus(node: Option) -> Command { diff --git a/packages/client/src/facade.ts b/packages/client/src/facade.ts index 14f4ae6c1..e573f7a4b 100644 --- a/packages/client/src/facade.ts +++ b/packages/client/src/facade.ts @@ -108,6 +108,14 @@ export class EngineFacade { return this.command({ kind: 'relink', node, newParent }); } + /** + * Cancel a queued upload by the op id `commitWrite` resolved with. Rejects + * with `tooLateToCancel` once the version's record is publishing. + */ + cancelUpload(opId: bigint): Promise { + return this.command({ kind: 'cancelUpload', opId }); + } + setFocus(node: Uint8Array | null): Promise { return this.command({ kind: 'setFocus', node }); } diff --git a/packages/client/src/worker/commandCodec.test.ts b/packages/client/src/worker/commandCodec.test.ts index 35d17f603..ce8a1d8eb 100644 --- a/packages/client/src/worker/commandCodec.test.ts +++ b/packages/client/src/worker/commandCodec.test.ts @@ -37,6 +37,23 @@ describe('buildCommand', () => { expect(calls[0][1]).toBe('a.txt'); expect(calls[0][2]).toBe(fakeWasmEnums.NodeKind.File); }); + + it('passes an upload cancel through as the bigint op id, not a number', () => { + const calls: unknown[][] = []; + const wasm = { + ...fakeWasmEnums, + Command: { + cancelUpload: (...args: unknown[]) => { + calls.push(args); + return {}; + }, + }, + } as unknown as EngineWasm; + + buildCommand(wasm, { kind: 'cancelUpload', opId: 2n ** 60n }); + + expect(calls).toEqual([[2n ** 60n]]); + }); }); describe('readEvent', () => { diff --git a/packages/client/src/worker/commandCodec.ts b/packages/client/src/worker/commandCodec.ts index 5c6fd06ed..245b79f60 100644 --- a/packages/client/src/worker/commandCodec.ts +++ b/packages/client/src/worker/commandCodec.ts @@ -54,6 +54,8 @@ export function buildCommand(wasm: EngineWasm, descriptor: CommandDescriptor): W return wasm.Command.rename(nodeId(wasm, descriptor.node), descriptor.newName); case 'relink': return wasm.Command.relink(nodeId(wasm, descriptor.node), nodeId(wasm, descriptor.newParent)); + case 'cancelUpload': + return wasm.Command.cancelUpload(descriptor.opId); case 'setFocus': return wasm.Command.setFocus( descriptor.node === null ? undefined : nodeId(wasm, descriptor.node) diff --git a/packages/client/src/worker/engineWasm.ts b/packages/client/src/worker/engineWasm.ts index 69849a495..60f0808ae 100644 --- a/packages/client/src/worker/engineWasm.ts +++ b/packages/client/src/worker/engineWasm.ts @@ -114,6 +114,7 @@ export interface EngineWasm { delete(node: WasmNodeId): WasmCommand; rename(node: WasmNodeId, newName: string): WasmCommand; relink(node: WasmNodeId, newParent: WasmNodeId): WasmCommand; + cancelUpload(opId: bigint): WasmCommand; setFocus(node?: WasmNodeId): WasmCommand; manualRefresh(): WasmCommand; importContact(contactCode: Uint8Array): WasmCommand; diff --git a/packages/client/src/worker/protocol.ts b/packages/client/src/worker/protocol.ts index daf065207..ca3260282 100644 --- a/packages/client/src/worker/protocol.ts +++ b/packages/client/src/worker/protocol.ts @@ -121,6 +121,7 @@ export type CommandDescriptor = | { kind: 'delete'; node: Uint8Array } | { kind: 'rename'; node: Uint8Array; newName: string } | { kind: 'relink'; node: Uint8Array; newParent: Uint8Array } + | { kind: 'cancelUpload'; opId: bigint } | { kind: 'setFocus'; node: Uint8Array | null } | { kind: 'manualRefresh' } | { kind: 'importContact'; contactCode: Uint8Array } From 78dc923567504ae7eb20f1d85949908dd43ef6d1 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 12:25:10 +0200 Subject: [PATCH 2/5] fix(engine): fold the review-gate findings into the cancel and GC slice Security review: - the orphan-GC live-set guard counted only handles opened, so an already-open handle staging its tail and root mid-sweep could have them collected; the counter now moves on every recorded key - the cancel retired the version's whole manifest, which in the window between an acked record PUT and the op's dequeue would unpin content a live record still names; it now retires only the blocks this session's drain confirmed, the publish claim is sticky across a failed attempt, and the dequeue runs before anything is unpinned Crypto/privacy review: - the cold-start undecodable path deleted the blocks named by a clear, unauthenticated header, so a co-tenant of the origin-shared store could plant a record bearing our public owner tag and destroy a queued version; removing the op record is enough, and orphan GC then reclaims on reference evidence - a preserved dead letter kept ciphertext no key could open, because the abandonment deleted the record carrying the version's content key; the preserved set now holds the whole op record Simplify: collapse staged_manifest into version_leaf_cids, fold the duplicated yield-then-check into cancel_checkpoint, collapse the publish claim and the confirmed-block list into one in-flight record, enumerate staged keys before reading the queue so an idle store short-circuits, and narrow the visibility widenings back to what is called. --- crates/engine/src/facade.rs | 65 +++--- crates/engine/src/sync/cancel.rs | 98 ++++++++- crates/engine/src/sync/drain.rs | 98 +++++---- crates/engine/src/sync/mod.rs | 5 +- crates/engine/src/sync/staging.rs | 324 +++++++++++++++++------------ crates/engine/tests/write_plane.rs | 107 +++++++++- 6 files changed, 458 insertions(+), 239 deletions(-) diff --git a/crates/engine/src/facade.rs b/crates/engine/src/facade.rs index a0113fe1b..1730ede58 100644 --- a/crates/engine/src/facade.rs +++ b/crates/engine/src/facade.rs @@ -21,6 +21,7 @@ use core::pin::Pin; use std::collections::{BTreeMap, BTreeSet}; use std::rc::Rc; +use cipherbox_core::content::encode_content_cid_str; use cipherbox_core::ipns::IpnsName; use cipherbox_core::seal::{ReadBody, seal_content_key}; use cipherbox_core::suite::ecdsa::EcdsaVerifier; @@ -51,7 +52,7 @@ use crate::session::SessionIdentity; use crate::storage_policy::StoragePolicy; use crate::sync::boot::{ColdStartError, ColdStartOutcome, ColdStartParams, cold_start}; use crate::sync::cancel::UploadCancels; -use crate::sync::drain::{Drain, DrainReport, DrainScope, registry_cids}; +use crate::sync::drain::{Drain, DrainReport, DrainScope}; use crate::sync::model::{NodeMeta, Snapshot, collation_key}; use crate::sync::op::{NewNode, Op, OpKind, Replaced, StagedContent}; use crate::sync::overlay::apply_overlay; @@ -61,10 +62,8 @@ use crate::sync::rebase::{QueueScan, QueueScanMemo, decode_queue}; pub use crate::sync::drain::BlockedOp; pub use crate::sync::rebase::DeadLetterReason; -use crate::sync::record::{RecordReader, RecordSeal, record_content_root_cid}; -use crate::sync::staging::{ - LiveBlocks, collect_orphans, release_version_blocks, stage_op, version_leaf_cids, -}; +use crate::sync::record::{RecordReader, RecordSeal}; +use crate::sync::staging::{LiveBlocks, collect_orphans, release_version_blocks, stage_op}; use crate::sync::staleness::{Connectivity, classify}; use crate::sync::tick::{FocusWindow, focus_folders, focus_folders_due}; @@ -1104,6 +1103,7 @@ struct LiveWrite { struct LiveWrites { ledger: StagingLedger, open: BTreeMap, + next: u64, } /// The re-point pointer-payload wire version the owner's own vault seals under @@ -1383,7 +1383,7 @@ impl Engine { // A crash between staging a version's blocks and journaling its op // leaves them referenced by nothing, so cold start is the first place // that residue can be reclaimed (#828). - self.collect_staging_orphans().await; + collect_orphans(&self.seams.staging_store, &self.live_blocks).await; self.spawn_liveness_loop(api.clone()); self.spawn_resolve_tick_loop(root_name, api.clone()); @@ -1469,17 +1469,6 @@ impl Engine { .remove_op(*op_id) .await .map_err(ColdStartError::Seam)?; - // Its intent body never opened, so the version's only content - // key is gone and its blocks are ciphertext nothing can ever - // read: released under the proven-unopenable rule rather than - // preserved (#818, #853). - if let Some(root_cid) = raw - .iter() - .find(|(id, _)| id == op_id) - .and_then(|(_, record)| record_content_root_cid(record).ok().flatten()) - { - release_version_blocks(&self.seams.staging_store, &root_cid).await; - } } } @@ -1959,7 +1948,9 @@ impl Engine { } }; - let handle = self.live_blocks.borrow_mut().open(); + writes.next += 1; + let handle = WriteHandle(writes.next); + self.live_blocks.borrow_mut().open(handle); writes.open.insert( handle, LiveWrite { @@ -2246,27 +2237,34 @@ impl Engine { Ok(()) } - /// Undo one queued upload: retire what of it reached the network, drop the - /// op, release its blocks, and tell the host. + /// Undo one queued upload: drop the op, retire what of it reached the + /// network ([`UploadCancels`]), release its blocks, and tell the host. /// - /// Retiring first is what makes the leaf CIDs recoverable at all — the - /// staged root is the only record of them, and releasing the blocks destroys - /// it (#916). The retire is best-effort: a refused batch leaves pin rows - /// charged, which is a leak, where leaving the op queued after the user - /// cancelled it would be a broken guarantee. + /// The dequeue goes first and is the only step allowed to fail the cancel: + /// an op that is still queued is still publishable, and unpinning its blocks + /// before it has left would publish a version whose leading leaves are gone. + /// The retire is then best-effort — a refused batch leaves pin rows charged, + /// which is a leak, where refusing the cancel over it would break the + /// guarantee the user was given. async fn discard_upload( &self, op_id: OpId, node: NodeId, root_cid: &[u8], ) -> Result<(), EngineError> { - if let Some(api) = &self.api { - let leaf_cids = version_leaf_cids(&self.seams.staging_store, root_cid).await; - let _ = retire(api, ®istry_cids(root_cid, &leaf_cids)).await; - } - // Dequeued before the blocks go, so no window leaves a durable op - // referencing content the store no longer holds. self.dequeue_op(op_id).await?; + let uploaded: Vec = self + .cancels + .borrow() + .uploaded_by(op_id) + .iter() + .map(|cid| encode_content_cid_str(cid)) + .collect(); + if let Some(api) = &self.api + && !uploaded.is_empty() + { + let _ = retire(api, &uploaded).await; + } release_version_blocks(&self.seams.staging_store, root_cid).await; let _ = self.events.unbounded_send(Event::OpProgress { op_id: Some(op_id), @@ -2286,11 +2284,6 @@ impl Engine { .map_err(EngineError::from_seam) } - /// One orphan-GC pass over the staged-byte plane (#828). - async fn collect_staging_orphans(&self) { - collect_orphans(&self.seams.staging_store, &self.live_blocks).await; - } - /// A rendered read of the current state — the gate-passing base snapshot ⊕ /// the pending-op overlay — for FUSE-shaped reads (children/lookup/attrs/ /// statfs). Fails `NotStarted` before [`start`](Self::start). diff --git a/crates/engine/src/sync/cancel.rs b/crates/engine/src/sync/cancel.rs index f00ba9207..a1baf2194 100644 --- a/crates/engine/src/sync/cancel.rs +++ b/crates/engine/src/sync/cancel.rs @@ -10,14 +10,30 @@ use std::collections::BTreeSet; use crate::seams::OpId; -/// Which uploads the user cancelled, and which one has passed publish entry. +/// The one op the drain is carrying, and how far it has got. +struct InFlight { + op_id: OpId, + /// The blocks confirmed on the network so far. + /// + /// Session-scoped by design: it is the only evidence a cancel has that a + /// block was charged by *this* upload rather than by a version that has + /// since published, and a retire without that evidence would unpin content + /// a live record still names (#916). + confirmed: Vec>, + /// Whether the version's record has been authored and PUT. Sticky across a + /// retry: a PUT that did not confirm may still be live at the name. + past_publish_entry: bool, +} + +/// Which uploads the user cancelled, and what the drain is doing with the one +/// op it carries. The drain is strictly FIFO and stops at the first op it +/// cannot finish, so exactly one upload is ever in flight. #[derive(Default)] pub(crate) struct UploadCancels { /// Cancelled op ids. Held for the session: an op leaves the durable queue /// with its cancel and never returns, so nothing here is ever reused. cancelled: BTreeSet, - /// The op whose version has finished uploading and is now being published. - publishing: Option, + in_flight: Option, } impl UploadCancels { @@ -25,7 +41,7 @@ impl UploadCancels { /// the caller must refuse the cancel rather than compensate a published /// mutation. pub(crate) fn request(&mut self, op_id: OpId) -> bool { - if self.publishing == Some(op_id) { + if self.carrying(op_id).is_some_and(|it| it.past_publish_entry) { return false; } self.cancelled.insert(op_id); @@ -45,18 +61,46 @@ impl UploadCancels { if self.cancelled.contains(&op_id) { return false; } - self.publishing = Some(op_id); + self.take_up(op_id).past_publish_entry = true; true } - /// The op has left the drain's hands, published or not. - pub(crate) fn leave_publish(&mut self) { - self.publishing = None; + /// The op published and left the durable queue. + pub(crate) fn published(&mut self, op_id: OpId) { + if self.carrying(op_id).is_some() { + self.in_flight = None; + } } pub(crate) fn is_cancelled(&self, op_id: OpId) -> bool { self.cancelled.contains(&op_id) } + + /// Record one more block of `op_id`'s version as confirmed on the network. + pub(crate) fn confirmed(&mut self, op_id: OpId, cid: &[u8]) { + self.take_up(op_id).confirmed.push(cid.to_vec()); + } + + /// The blocks a cancel of `op_id` may retire. + pub(crate) fn uploaded_by(&self, op_id: OpId) -> &[Vec] { + self.carrying(op_id).map_or(&[], |it| &it.confirmed) + } + + fn carrying(&self, op_id: OpId) -> Option<&InFlight> { + self.in_flight.as_ref().filter(|it| it.op_id == op_id) + } + + /// The in-flight record for `op_id`, starting a fresh one for a new op. + fn take_up(&mut self, op_id: OpId) -> &mut InFlight { + if self.carrying(op_id).is_none() { + self.in_flight = Some(InFlight { + op_id, + confirmed: Vec::new(), + past_publish_entry: false, + }); + } + self.in_flight.as_mut().expect("set just above") + } } #[cfg(test)] @@ -72,7 +116,6 @@ mod tests { "the record is publishing; cancel must be refused" ); - cancels.leave_publish(); let mut cancels = UploadCancels::default(); assert!(cancels.request(OpId(1))); assert!( @@ -81,6 +124,43 @@ mod tests { ); } + /// A publish that did not confirm may still be live at the name, so the + /// claim outlives the failed attempt and the op stays uncancellable. + #[test] + fn a_publish_claim_survives_an_attempt_that_did_not_confirm() { + let mut cancels = UploadCancels::default(); + cancels.enter_publish(OpId(1)); + assert!(!cancels.request(OpId(1))); + + cancels.published(OpId(1)); + assert!( + cancels.request(OpId(1)), + "once the op published and left the queue the claim is spent" + ); + } + + /// Only the blocks this session confirmed may be retired: a block an earlier + /// session sent is indistinguishable from one a published version names. + #[test] + fn only_this_sessions_confirmed_blocks_are_retirable() { + let mut cancels = UploadCancels::default(); + assert!(cancels.uploaded_by(OpId(1)).is_empty()); + + cancels.confirmed(OpId(1), b"leaf-0"); + cancels.confirmed(OpId(1), b"leaf-1"); + assert_eq!( + cancels.uploaded_by(OpId(1)), + [b"leaf-0".to_vec(), b"leaf-1".to_vec()] + ); + + cancels.confirmed(OpId(2), b"other-0"); + assert!( + cancels.uploaded_by(OpId(1)).is_empty(), + "a new upload starts the list over" + ); + assert_eq!(cancels.uploaded_by(OpId(2)), [b"other-0".to_vec()]); + } + /// The hold is per op: a cancel of a queued upload must not be refused /// because a different op happens to be publishing. #[test] diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 79c8b96de..a857ce4a9 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -65,7 +65,7 @@ use crate::sync::overlay::apply_overlay; use crate::sync::project::project_folder; use crate::sync::rebase::{AppliedOp, DeadLetterReason, decode_queue, replay}; use crate::sync::record::RecordReader; -use crate::sync::staging::{preserve_staged_root, release_version_blocks}; +use crate::sync::staging::{preserve_dead_letter, release_version_blocks, version_leaf_cids}; /// The staging key holding the drained-op high-water mark: every op id at or /// below the stored value has left this device's queue (#860). @@ -191,9 +191,8 @@ enum Halt { /// resume probe must find room for. needed_bytes: u64, }, - /// The user cancelled the upload. The facade already retired what it had - /// uploaded, released its blocks and dequeued it, so the valve does nothing - /// but stop the pass (#824). + /// The user cancelled the upload. The facade has already undone it, so the + /// valve does nothing but stop the pass (#824). Cancelled, } @@ -474,11 +473,10 @@ where continue; }; // A terminally unrebasable op keeps its staged bytes, and this is - // what keeps them reachable once its op record is gone (#853). - if let Some(root_cid) = op.content_root_cid() { - preserve_staged_root(self.staging, root_cid) - .await - .map_err(seam)?; + // what keeps them reachable — and openable — once the abandonment + // has dropped its record from the queue (#853). + if op.content_root_cid().is_some() { + self.preserve_dead_letter(*op_id).await?; } self.abandon(scope, *op_id, op).await?; report.dead_letters.push((*op_id, op.target, *reason)); @@ -492,16 +490,16 @@ where } for applied in &rebased.applied { - let outcome = self + if let Err(halt) = self .publish_applied(scope, &mut pass, applied, &rebased.rebased) - .await; - self.cancels.borrow_mut().leave_publish(); - if let Err(halt) = outcome { + .await + { self.apply_valve(scope, applied.op_id, &applied.op, halt, attempts, report) .await; return Err(halt); } self.dequeue_op(applied.op_id).await?; + self.cancels.borrow_mut().published(applied.op_id); report.published.push(applied.op_id); } Ok(()) @@ -1392,16 +1390,11 @@ where }; emit(OpPhase::UploadStarted, blocks(uploaded)); for (index, leaf_cid) in content.leaf_cids().iter().enumerate() { - // The block boundary a cancel gets to run at. Without it a whole - // version uploads inside one turn of the host's executor and the - // cancel guarantee collapses to "only before the op starts" (#824). - yield_now().await; - if self.cancels.borrow().is_cancelled(applied.op_id) { - return Err(Halt::Cancelled); - } + self.cancel_checkpoint(applied.op_id).await?; match self.staged_block(leaf_cid).await? { Some(block) => { self.upload_block(leaf_cid, &block).await?; + self.cancels.borrow_mut().confirmed(applied.op_id, leaf_cid); // A leaf a lost release left staged behind the mark is // re-uploaded here, and must not drag the mark back down // over the leaves past it — those are released, so an @@ -1421,14 +1414,14 @@ where None => {} } } - yield_now().await; - if self.cancels.borrow().is_cancelled(applied.op_id) { - return Err(Halt::Cancelled); - } + self.cancel_checkpoint(applied.op_id).await?; // The root goes up last and stays staged until the publish confirms: it // is the manifest every retry re-derives the plan from, so releasing it // before the record lands would strand a fully-uploaded version. self.upload_block(&staged.root_cid, &root_block).await?; + self.cancels + .borrow_mut() + .confirmed(applied.op_id, &staged.root_cid); emit(OpPhase::UploadCompleted, total); let content_cids = registry_cids(&staged.root_cid, content.leaf_cids()); @@ -1438,6 +1431,17 @@ where }) } + /// The block boundary a cancel gets to run at. Without the yield a whole + /// version uploads inside one turn of the host's executor, and the cancel + /// guarantee collapses to "only before the op starts" (#824). + async fn cancel_checkpoint(&self, op_id: OpId) -> Result<(), Halt> { + yield_now().await; + match self.cancels.borrow().is_cancelled(op_id) { + true => Err(Halt::Cancelled), + false => Ok(()), + } + } + /// Best-effort upload progress for the op driving this transfer (a dropped /// receiver is fine). fn emit_upload( @@ -1503,14 +1507,6 @@ where Ok(Some(block)) } - /// The version manifest a staged root block carries. `None` when the block - /// is gone or unreadable: both callers are reconciliation paths that must - /// still make progress on a store that has lost bytes. - async fn staged_manifest(&self, root_cid: &[u8]) -> Option { - let block = self.staged_block(root_cid).await.ok()??; - SealedContent::from_root_block(&block).ok() - } - /// Upload one block to the pin provider under `cid`, its staging key and /// own content address, so the ingress pins it where the published record /// points (#906). A block is only ever removed from staging on a confirmed @@ -1524,19 +1520,10 @@ where .map_err(|error| classify_upload(error, block.len() as u64)) } - /// Drop every staged block of an op's version. - /// - /// Called once its record publishes — the bytes are on the network — and on - /// a failure-valve abandonment, where the blocks are not the user's - /// recoverable work: the only copy of the version's content key rides the - /// op record, which the abandonment deletes, so what survives is ciphertext - /// nothing can ever open. Holding it would spend the staging budget forever - /// (#818; the dead-letter event is what surfaces the loss). A terminally - /// unrebasable op keeps its staged bytes instead (blueprint/engine.md, #33 - /// D6), so this is not called on that path. - /// - /// Best-effort: a failed removal is orphan residue a later GC pass collects, - /// never a reason to fail a landed publish. + /// Drop every staged block of an op's version — on a landed publish, and on + /// a failure-valve abandonment, where the only copy of the version's content + /// key rode the op record the abandonment deletes + /// (`crate::sync::staging` owns the release-or-preserve rule). async fn release_staged_blocks(&self, op: &Op) { let Some(root_cid) = op.content_root_cid() else { return; @@ -1814,6 +1801,19 @@ where } } + /// Copy one queued op's record into the preserved set before the + /// abandonment removes it, so the version it stages stays both referenced + /// and openable ([`preserve_dead_letter`]). + async fn preserve_dead_letter(&self, op_id: OpId) -> Result<(), Halt> { + let queued = self.staging.queued_ops().await.map_err(seam)?; + let Some((_, record)) = queued.iter().find(|(id, _)| *id == op_id) else { + return Ok(()); + }; + preserve_dead_letter(self.staging, record) + .await + .map_err(seam) + } + /// Abandon one op: retire what its publish registered, then drop it from /// the queue (#819 as amended by #824). async fn abandon(&self, scope: &DrainScope<'_>, op_id: OpId, op: &Op) -> Result<(), Halt> { @@ -1861,11 +1861,7 @@ where }); let content = match op.content_root_cid() { Some(root_cid) => { - let manifest = self.staged_manifest(root_cid).await; - registry_cids( - root_cid, - manifest.as_ref().map_or(&[], SealedContent::leaf_cids), - ) + registry_cids(root_cid, &version_leaf_cids(self.staging, root_cid).await) } None => Vec::new(), }; @@ -2097,7 +2093,7 @@ fn upload_failure(halt: Halt) -> Option<&'static str> { /// batch names exactly what a register batch claimed — and every block is its /// own accountable pin row (blueprint/api.md "Pin/name registry"), so leaving /// a leaf out of a retirement spends account quota forever. -pub(crate) fn registry_cids(root_cid: &[u8], leaf_cids: &[Vec]) -> Vec { +fn registry_cids(root_cid: &[u8], leaf_cids: &[Vec]) -> Vec { core::iter::once(root_cid) .chain(leaf_cids.iter().map(Vec::as_slice)) .map(encode_content_cid_str) diff --git a/crates/engine/src/sync/mod.rs b/crates/engine/src/sync/mod.rs index 854a065ee..5c2daedab 100644 --- a/crates/engine/src/sync/mod.rs +++ b/crates/engine/src/sync/mod.rs @@ -48,10 +48,7 @@ pub use record::{ OpRecordError, RecordClass, RecordReader, RecordSeal, RetainedReason, encode_op_record, record_content_root_cid, }; -pub use staging::{ - PRESERVED_ROOTS_KEY, orphan_staging_keys, preserve_staged_root, release_version_blocks, - stage_op, -}; +pub use staging::{orphan_staging_keys, stage_op}; pub use staleness::{Connectivity, classify, withheld_escalation}; pub use tick::{ FocusTarget, FocusWindow, ResolveMode, TickCause, TickControl, focus_folders, diff --git a/crates/engine/src/sync/staging.rs b/crates/engine/src/sync/staging.rs index 3b3ca6c48..9d730095f 100644 --- a/crates/engine/src/sync/staging.rs +++ b/crates/engine/src/sync/staging.rs @@ -9,6 +9,12 @@ //! storage. That bound is admitted whole at `beginWrite` //! ([`crate::content::StagingLedger`]); this module owns the journal entry and //! the staged-byte hygiene that outlives it. +//! +//! One rule decides a staged version's fate, and it lives here: **preserved when +//! the engine gave up on the op, released when the user did** (#853). Preserved +//! on a terminally unrebasable dead letter ([`preserve_dead_letter`]); released +//! on a cancel, on a version proven unopenable (#818), and on a staged root that +//! cannot be expanded ([`release_version_blocks`]). use std::collections::{BTreeMap, HashSet}; @@ -50,31 +56,28 @@ pub async fn stage_op( store.enqueue_op(&record).await } -/// The staging key holding the roots of dead-lettered versions whose bytes are -/// **preserved**: the op record that referenced them is gone from the queue, and -/// this record is what keeps orphan GC off them (#853). +/// The staging key holding the **op records** of dead letters whose staged bytes +/// are preserved, `u32`-length-prefixed behind a one-byte format tag. /// -/// It lives in the staging store for the same reason the drain's other -/// bookkeeping does — the roots and the blocks they name share one durability -/// domain — and [`orphan_staging_keys`] treats it as referenced. -pub const PRESERVED_ROOTS_KEY: &[u8] = b"cipherbox/preserved-roots"; - -/// The preserved-roots record's format tag. The staging store is shared with -/// whatever build wrote it, so bytes that merely happen to be well-shaped must -/// not parse as roots. +/// It holds the whole record, not just the root CID, because the record is the +/// only carrier of the version's content key — a KDF non-edge nothing can +/// re-derive (#818). Preserving the blocks without it would keep ciphertext no +/// key ever opens, which is the condition that *releases* a version, not the one +/// that preserves it. Orphan GC reads each entry's root keylessly through the +/// same frozen clear header a queue entry exposes ([`record_content_root_cid`]), +/// and treats this key as referenced. +pub(crate) const PRESERVED_DEAD_LETTERS_KEY: &[u8] = b"cipherbox/preserved-dead-letters"; + +/// The preserved record's format tag. The staging store is shared with whatever +/// build wrote it, so bytes that merely happen to be well-shaped must not parse. const PRESERVED_FORMAT_V1: u8 = 1; /// Drop every staged block of one version: the leaves its root manifest lists, -/// in file order, then the root itself. -/// -/// Staged bytes are released on exactly three verdicts — the user cancelled the -/// upload, the version is proven unopenable (#818), or its staged root cannot be -/// expanded — and preserved on the fourth, a terminally unrebasable op -/// (blueprint/engine.md; #33 D6). File order keeps the blocks that remain a -/// suffix at every step, which is the invariant the drain's resume reads. +/// in file order, then the root itself. File order keeps the blocks that remain +/// a suffix at every step, which is the invariant the drain's resume reads. /// /// Best-effort: a failed removal is orphan residue a later GC pass collects. -pub async fn release_version_blocks(store: &S, root_cid: &[u8]) { +pub(crate) async fn release_version_blocks(store: &S, root_cid: &[u8]) { for leaf_cid in version_leaf_cids(store, root_cid).await { let _ = store.remove_staged_bytes(&leaf_cid).await; } @@ -85,7 +88,7 @@ pub async fn release_version_blocks(store: &S, root_cid: &[u8]) /// is gone, fails its own CID, or does not decode — every caller is a /// reconciliation path that must still make progress on a store that has lost /// bytes. -pub async fn version_leaf_cids(store: &S, root_cid: &[u8]) -> Vec> { +pub(crate) async fn version_leaf_cids(store: &S, root_cid: &[u8]) -> Vec> { let Ok(Some(block)) = store.staged_bytes(root_cid).await else { return Vec::new(); }; @@ -95,61 +98,79 @@ pub async fn version_leaf_cids(store: &S, root_cid: &[u8]) -> V decode_root(&block).map(|m| m.leaf_cids).unwrap_or_default() } -/// Record `root_cid` as a preserved dead letter, so orphan GC keeps its blocks -/// once the op that referenced them leaves the durable queue. +/// Keep one dead letter's op record after it leaves the durable queue, so orphan +/// GC keeps the blocks it names and the version stays openable (#853). /// /// Fails closed on a preserved record this build cannot read: overwriting it -/// would drop the roots it already names, and those blocks are exactly what the -/// dead-letter contract promises to keep. -pub async fn preserve_staged_root(store: &S, root_cid: &[u8]) -> SeamResult<()> { - let mut roots = read_preserved_roots(store) +/// would drop the dead letters it already holds, and those are exactly what the +/// contract promises to keep. +pub(crate) async fn preserve_dead_letter( + store: &S, + record: &[u8], +) -> SeamResult<()> { + let mut kept = read_preserved_dead_letters(store) .await? - .ok_or_else(|| SeamError::new("preserve_staged_root: unreadable preserved-roots record"))?; - if roots.iter().any(|root| root == root_cid) { + .ok_or_else(|| SeamError::new("preserve_dead_letter: unreadable preserved record"))?; + if kept.iter().any(|held| held == record) { return Ok(()); } - roots.push(root_cid.to_vec()); - write_preserved_roots(store, &roots).await + kept.push(record.to_vec()); + write_preserved_dead_letters(store, &kept).await } -/// The preserved roots the store holds. `None` when the record is present but -/// not one this build wrote — the fail-safe direction is to preserve, so a -/// caller must freeze rather than treat it as empty. -async fn read_preserved_roots(store: &S) -> SeamResult>>> { - let Some(stored) = store.staged_bytes(PRESERVED_ROOTS_KEY).await? else { +/// The dead letters the store holds. `None` when the record is present but not +/// one this build wrote — the fail-safe direction is to preserve, so a caller +/// must freeze rather than treat it as empty. +async fn read_preserved_dead_letters( + store: &S, +) -> SeamResult>>> { + let Some(stored) = store.staged_bytes(PRESERVED_DEAD_LETTERS_KEY).await? else { return Ok(Some(Vec::new())); }; let Some(mut rest) = stored.strip_prefix(&[PRESERVED_FORMAT_V1]) else { return Ok(None); }; - let mut roots = Vec::new(); - while let Some((&len, tail)) = rest.split_first() { - let Some((cid, next)) = tail.split_at_checked(len as usize) else { + let mut kept = Vec::new(); + while !rest.is_empty() { + let Some((len, tail)) = rest.split_at_checked(4) else { return Ok(None); }; - roots.push(cid.to_vec()); + let len = u32::from_be_bytes(len.try_into().expect("4 bytes")) as usize; + // A zero-length entry is not a record, and would loop forever. + let Some((record, next)) = tail.split_at_checked(len).filter(|_| len > 0) else { + return Ok(None); + }; + kept.push(record.to_vec()); rest = next; } - Ok(Some(roots)) + Ok(Some(kept)) } -/// Fails closed on a root too long to length-prefix: writing a truncated record -/// would silently unpin every root behind it (AGENTS.md rule 8). +/// Fails closed on a record too long to length-prefix, and on an empty one: +/// writing either would silently unpin every dead letter behind it, which the +/// reader hard-rejects (AGENTS.md rule 8). /// /// An empty list removes the key rather than storing a tag-only record, so a /// device that has never dead-lettered spends no staging budget on this. -async fn write_preserved_roots(store: &S, roots: &[Vec]) -> SeamResult<()> { - if roots.is_empty() { - return store.remove_staged_bytes(PRESERVED_ROOTS_KEY).await; +async fn write_preserved_dead_letters( + store: &S, + kept: &[Vec], +) -> SeamResult<()> { + if kept.is_empty() { + return store.remove_staged_bytes(PRESERVED_DEAD_LETTERS_KEY).await; } let mut bytes = vec![PRESERVED_FORMAT_V1]; - for root in roots { - let len = u8::try_from(root.len()) - .map_err(|_| SeamError::new("preserved root is not length-prefixable"))?; - bytes.push(len); - bytes.extend_from_slice(root); - } - store.put_staged_bytes(PRESERVED_ROOTS_KEY, &bytes).await + for record in kept { + let len = u32::try_from(record.len()) + .ok() + .filter(|len| *len > 0) + .ok_or_else(|| SeamError::new("preserved dead letter is not length-prefixable"))?; + bytes.extend_from_slice(&len.to_be_bytes()); + bytes.extend_from_slice(record); + } + store + .put_staged_bytes(PRESERVED_DEAD_LETTERS_KEY, &bytes) + .await } /// The staging keys write handles hold outside the durable queue. @@ -162,22 +183,23 @@ async fn write_preserved_roots(store: &S, roots: &[Vec]) -> #[derive(Default)] pub(crate) struct LiveBlocks { by_handle: BTreeMap>>, - /// Handles opened so far — the handle-id source, and the evidence a GC sweep - /// reads to tell that no handle opened while it was running. - opened: u64, + /// Bumped by every open and every recorded key. A GC sweep spans many + /// awaits, so it reads this to tell that the live set it read is still the + /// whole live set — counting only handles would miss an already-open one + /// staging its tail and root mid-sweep. + generation: u64, } impl LiveBlocks { - /// Mint the next handle and start holding its staging keys. - pub(crate) fn open(&mut self) -> WriteHandle { - self.opened += 1; - let handle = WriteHandle(self.opened); + /// Start holding `handle`'s staging keys. + pub(crate) fn open(&mut self, handle: WriteHandle) { + self.generation += 1; self.by_handle.insert(handle, Vec::new()); - handle } /// Hold one more staging key for `handle`, before its bytes are staged. pub(crate) fn record(&mut self, handle: WriteHandle, key: &[u8]) { + self.generation += 1; if let Some(keys) = self.by_handle.get_mut(&handle) { keys.push(key.to_vec()); } @@ -194,9 +216,8 @@ impl LiveBlocks { self.by_handle.values().flatten().cloned().collect() } - /// How many handles have ever been opened. - pub(crate) fn opened(&self) -> u64 { - self.opened + pub(crate) fn generation(&self) -> u64 { + self.generation } } @@ -205,42 +226,47 @@ impl LiveBlocks { /// pass. /// /// Best-effort throughout — a store that cannot enumerate or remove leaves its -/// residue for the next pass — and it also prunes [`PRESERVED_ROOTS_KEY`] of +/// residue for the next pass — and it also prunes [`PRESERVED_DEAD_LETTERS_KEY`] of /// roots whose bytes are already gone, so that record cannot grow without bound. pub(crate) async fn collect_orphans( store: &S, live: &core::cell::RefCell, ) { - let opened_before = live.borrow().opened(); - let live_keys = live.borrow().keys(); + let (generation, live_keys) = { + let live = live.borrow(); + (live.generation(), live.keys()) + }; let Ok(orphans) = orphan_staging_keys(store, &live_keys).await else { return; }; for key in orphans { - // A handle opened after the scan may be staging exactly these bytes. - if live.borrow().opened() != opened_before { + // A block staged since the scan is not in the live set this pass read, + // and its owning handle has not journaled an op that references it — + // abandoning the sweep is the only safe reading of that. + if live.borrow().generation() != generation { return; } let _ = store.remove_staged_bytes(&key).await; } - prune_preserved_roots(store).await; + prune_preserved_dead_letters(store).await; } -/// Drop preserved roots whose blocks the store no longer holds. -async fn prune_preserved_roots(store: &S) { - let Ok(Some(roots)) = read_preserved_roots(store).await else { +/// Drop preserved dead letters whose blocks the store no longer holds. +async fn prune_preserved_dead_letters(store: &S) { + let Ok(Some(kept)) = read_preserved_dead_letters(store).await else { return; }; - let mut live = Vec::with_capacity(roots.len()); - for root in &roots { - if matches!(store.staged_bytes(root).await, Ok(Some(_))) { - live.push(root.clone()); + let mut live = Vec::with_capacity(kept.len()); + for record in &kept { + let Ok(Some(root)) = record_content_root_cid(record) else { + continue; + }; + if matches!(store.staged_bytes(&root).await, Ok(Some(_))) { + live.push(record.clone()); } } - // An unchanged record is not rewritten, so an idle sweep makes no store - // write. - if live.len() != roots.len() { - let _ = write_preserved_roots(store, &live).await; + if live.len() != kept.len() { + let _ = write_preserved_dead_letters(store, &live).await; } } @@ -250,37 +276,48 @@ async fn prune_preserved_roots(store: &S) { /// Three things reference a block. A **queued record**'s content root rides its /// clear header, and since a version stages one block per key, a root is /// expanded into the leaf keys its own manifest lists — so a foreign account's -/// or a forward-version record's whole block set is retained. A **preserved -/// dead letter**'s root is expanded the same way, which is what keeps the bytes -/// the dead-letter contract promises once its op record is gone (#853). An -/// **open write handle**'s blocks are staged before any op is journaled, so they -/// are unreferenced by construction and must be passed in as `live`; collecting -/// them mid-write would publish a version whose manifest names blocks nothing -/// holds. +/// or a forward-version record's whole block set is retained. A **preserved dead +/// letter**'s record is read the same way, which is what keeps the bytes the +/// contract promises once that record has left the queue (#853). An **open write +/// handle**'s blocks are staged before any op is journaled, so they are +/// unreferenced by construction and must be passed in as `live`; collecting them +/// mid-write would publish a version whose manifest names blocks nothing holds. /// -/// Fail-closed: an unreadable queue entry, an unreadable preserved-roots record, -/// or a referenced root the store cannot produce or this build cannot decode, -/// classes **nothing** an orphan. A root that cannot be expanded therefore -/// freezes the whole pass — self-clearing, because the drain classifies that -/// same root permanent and releases the version. +/// Fail-closed: an unreadable queue entry, an unreadable preserved record, or a +/// referenced root the store cannot produce or this build cannot decode, classes +/// **nothing** an orphan. A root that cannot be expanded therefore freezes the +/// whole pass — self-clearing, because the drain classifies that same root +/// permanent and releases the version. pub async fn orphan_staging_keys( store: &S, live: &[Vec], ) -> SeamResult>> { - let queued = store.queued_ops().await?; // The drain's own queue bookkeeping is not upload residue. let mut referenced = HashSet::from([ DRAINED_OP_MARK_KEY.to_vec(), OP_ATTEMPTS_KEY.to_vec(), UPLOAD_MARK_KEY.to_vec(), - PRESERVED_ROOTS_KEY.to_vec(), + PRESERVED_DEAD_LETTERS_KEY.to_vec(), ]); referenced.extend(live.iter().cloned()); - let Some(preserved) = read_preserved_roots(store).await? else { + // Enumerated first, so an idle store answers without reading the queue at + // all, and a version journaled mid-pass is decided by a queue read that + // already covers it. + let candidates: Vec> = store + .staged_keys() + .await? + .into_iter() + .filter(|key| !referenced.contains(key)) + .collect(); + if candidates.is_empty() { + return Ok(candidates); + } + let Some(preserved) = read_preserved_dead_letters(store).await? else { return Ok(Vec::new()); }; - let mut roots = preserved; - for (_, record) in &queued { + let queued = store.queued_ops().await?; + let mut roots = Vec::new(); + for record in preserved.iter().chain(queued.iter().map(|(_, r)| r)) { let Ok(root) = record_content_root_cid(record) else { // An unreadable record may still reference staged bytes, and its // root is unknowable. @@ -301,13 +338,10 @@ pub async fn orphan_staging_keys( } referenced.insert(root); } - let orphans = store - .staged_keys() - .await? + Ok(candidates .into_iter() .filter(|key| !referenced.contains(key)) - .collect(); - Ok(orphans) + .collect()) } #[cfg(test)] @@ -317,6 +351,7 @@ mod tests { use crate::facade::NodeId; use crate::seams::UnixMillis; use crate::sync::op::{NewNode, StagedContent}; + use crate::sync::record::{RecordClass, RecordReader}; use crate::testkit::fakes::InMemoryStagingStore; use crate::testkit::{block_on, frame_version}; use cipherbox_core::suite::aead::KEY_LEN; @@ -618,9 +653,8 @@ mod tests { block_on(async { let (blocks, root_block, staged) = framed(b"forty bytes of content ------------------"); put_blocks(&store, &blocks, &root_block, &staged).await; - preserve_staged_root(&store, &staged.root_cid) - .await - .unwrap(); + let record = encode_op_record(seal(1), &content_op(1, staged)).unwrap(); + preserve_dead_letter(&store, &record).await.unwrap(); store.put_staged_bytes(b"orphan", b"stale").await.unwrap(); assert_eq!( @@ -631,53 +665,85 @@ mod tests { }); } - /// The fail-safe direction here is to preserve, so a preserved-roots record - /// this build cannot read must freeze the pass rather than read as empty. + /// Preserving the whole record, not just the root, is what keeps the version + /// openable: the sealed content key is a KDF non-edge and the record is its + /// only carrier (#818). #[test] - fn an_unreadable_preserved_roots_record_makes_orphan_gc_conservative() { + fn a_preserved_dead_letter_still_carries_the_key_that_opens_its_version() { let store = InMemoryStagingStore::default(); block_on(async { - store - .put_staged_bytes(PRESERVED_ROOTS_KEY, b"not a preserved-roots record") + let (_, _, staged) = framed(b"forty bytes of content ------------------"); + let op = content_op(1, staged); + preserve_dead_letter(&store, &encode_op_record(seal(1), &op).unwrap()) .await .unwrap(); - store.put_staged_bytes(b"orphan", b"stale").await.unwrap(); - assert!(orphan_staging_keys(&store, &[]).await.unwrap().is_empty()); - assert!( - preserve_staged_root(&store, b"root").await.is_err(), - "overwriting it would drop the roots it already names" + let kept = read_preserved_dead_letters(&store).await.unwrap().unwrap(); + assert_eq!( + RecordReader::new(&OWNER).classify(&kept[0]), + RecordClass::Mine(op), + "the preserved bytes reopen to the intent, sealed key included" ); }); } + /// The fail-safe direction here is to preserve, so a preserved record this + /// build cannot read must freeze the pass rather than read as empty. + #[test] + fn an_unreadable_preserved_record_makes_orphan_gc_conservative() { + let store = InMemoryStagingStore::default(); + block_on(async { + store.put_staged_bytes(b"orphan", b"stale").await.unwrap(); + // A wrong tag, a length prefix past the end, and a zero-length entry + // that would otherwise loop forever. + for stored in [ + b"not a preserved record".to_vec(), + vec![PRESERVED_FORMAT_V1, 0, 0, 0, 9, 1, 2], + vec![PRESERVED_FORMAT_V1, 0, 0, 0, 0], + ] { + store + .put_staged_bytes(PRESERVED_DEAD_LETTERS_KEY, &stored) + .await + .unwrap(); + assert!(read_preserved_dead_letters(&store).await.unwrap().is_none()); + assert!(orphan_staging_keys(&store, &[]).await.unwrap().is_empty()); + assert!( + preserve_dead_letter(&store, b"record").await.is_err(), + "overwriting it would drop the dead letters it already holds" + ); + } + }); + } + #[test] - fn preserved_roots_round_trip_and_prune_to_the_blocks_the_store_still_holds() { + fn preserved_dead_letters_round_trip_and_prune_to_the_blocks_still_held() { let store = InMemoryStagingStore::default(); block_on(async { let (blocks, root_block, staged) = framed(b"forty bytes of content ------------------"); put_blocks(&store, &blocks, &root_block, &staged).await; - preserve_staged_root(&store, &staged.root_cid) - .await - .unwrap(); - preserve_staged_root(&store, b"gone").await.unwrap(); + let root_cid = staged.root_cid.clone(); + let held = encode_op_record(seal(1), &content_op(1, staged)).unwrap(); + let (_, _, gone) = framed(b"another forty bytes of content ----------"); + let collected = encode_op_record(seal(2), &content_op(2, gone)).unwrap(); + preserve_dead_letter(&store, &held).await.unwrap(); + preserve_dead_letter(&store, &collected).await.unwrap(); assert_eq!( - read_preserved_roots(&store).await.unwrap().unwrap(), - vec![staged.root_cid.clone(), b"gone".to_vec()] + read_preserved_dead_letters(&store).await.unwrap().unwrap(), + vec![held.clone(), collected] ); - prune_preserved_roots(&store).await; + prune_preserved_dead_letters(&store).await; assert_eq!( - read_preserved_roots(&store).await.unwrap().unwrap(), - vec![staged.root_cid.clone()], - "a root whose blocks are gone preserves nothing" + read_preserved_dead_letters(&store).await.unwrap().unwrap(), + vec![held], + "a dead letter whose blocks are gone preserves nothing" ); - release_version_blocks(&store, &staged.root_cid).await; - prune_preserved_roots(&store).await; + release_version_blocks(&store, &root_cid).await; + prune_preserved_dead_letters(&store).await; assert!( store - .staged_bytes(PRESERVED_ROOTS_KEY) + .staged_bytes(PRESERVED_DEAD_LETTERS_KEY) .await .unwrap() .is_none(), diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 16d7ea1dc..6298e22cd 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -3637,13 +3637,19 @@ fn a_cancel_mid_upload_releases_every_block_and_returns_the_staging_budget() { block_on(engine.view()).unwrap().children(ROOT).is_empty(), "nothing published" ); - let retired = retire_targets(&alice); + // Every block that reached the network is a charged pin row with no + // reachable record behind it, so the cancel must retire exactly those and + // nothing the version never sent (#916). + let charged: Vec = version + .iter() + .map(|cid| encode_content_cid_str(cid)) + .filter(|cid| blocks.get(cid).is_some()) + .collect(); assert!( - version - .iter() - .all(|cid| retired.contains(&encode_content_cid_str(cid))), - "every block the version could have charged is retired, root and leaves" + (1..version.len()).contains(&charged.len()), + "the cancel landed mid-set: part of the version is charged, not all of it" ); + assert_eq!(retire_targets(&alice), charged); assert!( events_so_far(&mut events).contains(&Event::OpProgress { op_id: Some(op_id), @@ -3796,38 +3802,119 @@ fn a_cancelled_create_cascades_onto_its_node_and_a_cancelled_version_does_not() ); } -/// A cancel that cannot carry out its removals must give the claim back: an op -/// left both queued and claimed would halt every pass behind it forever. +/// A cancel that cannot carry out its removals must give the claim back and +/// unpin nothing: an op left both queued and claimed would halt every pass +/// behind it forever, and one left queued with its leading leaves retired would +/// publish a version whose blocks are gone. #[test] -fn a_cancel_that_cannot_dequeue_leaves_the_op_publishable() { +fn a_cancel_that_cannot_dequeue_retires_nothing_and_leaves_the_op_publishable() { let world = FakeWorld::new(); let blocks = Blocks::default(); seed_account(&world, &blocks); let alice = world.device(b"alice"); let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let plaintext: Vec = (0..200u8).collect(); let op_id = write_file( &mut engine, WriteTarget::NewFile { parent: ROOT, name: "photo.bin".into(), }, - b"bytes that still publish", + &plaintext, ) .unwrap(); + // Part of the version is already on the network when the cancel arrives, so + // there is a retire batch to get wrong. + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + assert!(uploads(&alice) > 0); + alice.staging_store.fail_remove_op(); assert!( block_on(engine.command(Command::CancelUpload { op_id })).is_err(), "the cancel could not remove the op, so it did not happen" ); + poll_each(&mut tasks); - tick(&world, &engine, &mut tasks); + assert!( + retire_targets(&alice).is_empty(), + "an op that is still publishable keeps every pin its upload charged" + ); assert_eq!( published_names(&world.record_store, &blocks, ROOT), vec!["photo.bin".to_owned()], "the op the cancel could not take is published, not wedged" ); + let file = child_id(&engine, ROOT, "photo.bin"); + assert_eq!( + block_on(engine.read_content(file)).expect("the published version reads back"), + plaintext + ); +} + +/// The op-record header is clear and unauthenticated, and the owner tag on it is +/// a public key any co-tenant of the origin-shared store can copy. A record that +/// bears our tag but never opens is dead-lettered and dropped at cold start — it +/// must not also authorize deleting the blocks its header names, or planting one +/// would destroy a queued version whose key is intact. +#[test] +fn an_undecodable_record_never_authorizes_deleting_the_blocks_its_header_names() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, tasks) = boot(&world, &blocks, &alice, 42); + let plaintext: Vec = (0..200u8).collect(); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + &plaintext, + ) + .unwrap(); + let version = queued_version(&alice, op_id); + + // The forgery: the real op's record with its sealed body corrupted, so the + // header — our owner tag, and the real op's content root — still reads. + let mut forged = block_on(alice.staging_store.queued_ops()).unwrap()[0] + .1 + .clone(); + let last = forged.len() - 1; + forged[last] ^= 1; + block_on(alice.staging_store.enqueue_op(&forged)).unwrap(); + drop(engine); + drop(tasks); + + let (engine, mut events, mut tasks) = boot(&world, &blocks, &alice, 43); + assert!( + events_so_far(&mut events).iter().any(|event| matches!( + event, + Event::DeadLetter { + reason: DeadLetterReason::Undecodable, + .. + } + )), + "the forgery must reach the path under test, not be retained short of it" + ); + let staged = block_on(alice.staging_store.staged_keys()).unwrap(); + assert!( + version.iter().all(|cid| staged.contains(cid)), + "the forged record was dropped; the version it named was not" + ); + + tick(&world, &engine, &mut tasks); + let file = child_id(&engine, ROOT, "photo.bin"); + assert_eq!( + block_on(engine.read_content(file)).expect("the real op still publishes"), + plaintext + ); } /// Orphan GC runs after each drain pass and reclaims blocks nothing references — From bd4cc0ee56ada42a6879d6fd0fcf9475ab544c46 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 19:24:56 +0200 Subject: [PATCH 3/5] fix(engine): close the retire gap a cancel leaves against the reordered mark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebasing onto #941 put the cancel interlock and the mark-before-release ordering in the same match arm. Both are kept: a confirmed block is recorded for the retire set, then the mark advances only when it is allowed to, then the leaf is released. The window that reorder widens is not the one it looks like. A cancel cannot strand a marked-but-unreleased leaf — the facade releases every leaf the root manifest lists, mark or no mark — and the mark it leaves behind names a root nothing can upload again. What is reachable is the upload the drain was already awaiting when the cancel landed: it confirms after the facade snapshotted its retire batch, so it stayed charged with nothing left to reach it. The drain now retires the complete confirmed set on its cancelled arm, which is idempotent against the facade's. Tests: the cancel retire is asserted as a set over both batches; a cancelled version's stale mark is shown not to count towards the next one; and a leaf a lost release strands on the cancel path is shown to be reclaimed by the sweep, since nothing re-runs that release. --- crates/engine/src/sync/drain.rs | 25 +++++- crates/engine/tests/write_plane.rs | 136 ++++++++++++++++++++++++++++- 2 files changed, 158 insertions(+), 3 deletions(-) diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index a857ce4a9..57b1ba243 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -516,7 +516,14 @@ where report: &mut DrainReport, ) { match halt { - Halt::Unclassified | Halt::Cancelled => {} + Halt::Unclassified => {} + // The facade undid the op against the blocks it could see when the + // cancel landed. One more can confirm inside that window — the + // upload the drain was already awaiting — and it would be charged + // with nothing left to reach it, so the complete set is retired + // here. Idempotent, so the overlap with the facade's batch is a + // no-op (#916). + Halt::Cancelled => self.retire_cancelled(op_id).await, Halt::Attempt | Halt::UploadAttempt => { if attempts.charge(op_id) < ATTEMPT_BUDGET { return; @@ -1801,6 +1808,22 @@ where } } + /// Retire every block a cancelled op put on the network. Best-effort: a + /// refused batch leaves pin rows charged, which is a leak, where failing the + /// pass over an op that is already gone would be a stuck queue. + async fn retire_cancelled(&self, op_id: OpId) { + let cids: Vec = self + .cancels + .borrow() + .uploaded_by(op_id) + .iter() + .map(|cid| encode_content_cid_str(cid)) + .collect(); + if !cids.is_empty() { + let _ = retire(self.api, &cids).await; + } + } + /// Copy one queued op's record into the preserved set before the /// abandonment removes it, so the version it stages stays both referenced /// and openable ([`preserve_dead_letter`]). diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 6298e22cd..271c837e5 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -3649,7 +3649,22 @@ fn a_cancel_mid_upload_releases_every_block_and_returns_the_staging_budget() { (1..version.len()).contains(&charged.len()), "the cancel landed mid-set: part of the version is charged, not all of it" ); - assert_eq!(retire_targets(&alice), charged); + // Both halves of the retire fire — the facade's, against what it could see + // when the cancel landed, and the drain's, against the complete confirmed + // set once it stops. Their union is the invariant; the overlap is an + // idempotent replay, which is why this compares sets and not batches. + let batches = retire_batches(&alice); + assert_eq!( + batches.len(), + 2, + "a block confirming inside the facade's window is only covered by the drain's batch" + ); + let mut retired = retire_targets(&alice); + retired.sort(); + retired.dedup(); + let mut expected = charged.clone(); + expected.sort(); + assert_eq!(retired, expected); assert!( events_so_far(&mut events).contains(&Event::OpProgress { op_id: Some(op_id), @@ -3662,6 +3677,68 @@ fn a_cancel_mid_upload_releases_every_block_and_returns_the_staging_budget() { ); } +/// A cancel releases leaves the durable mark still covers, and leaves that mark +/// behind naming a root nothing will ever upload again. That residue must not +/// reach the next version: a mark read as this version's progress would skip +/// leaves it never sent and publish a manifest naming blocks nobody holds +/// (#924's mark, #824's release). +#[test] +fn a_cancelled_versions_upload_mark_never_counts_towards_the_next_one() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let cancelled = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "abandoned.bin".into(), + }, + &(0..200u8).collect::>(), + ) + .unwrap(); + let root_cid = queued_version(&alice, cancelled)[0].clone(); + + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + block_on(engine.command(Command::CancelUpload { op_id: cancelled })).unwrap(); + poll_each(&mut tasks); + + let mark = block_on(alice.staging_store.staged_bytes(UPLOAD_MARK_KEY)) + .unwrap() + .expect("the cancelled pass left its progress mark behind"); + assert!( + mark.starts_with(&root_cid), + "the residue names the cancelled root, so the next version must not read it as progress" + ); + + let plaintext: Vec = (0..200u8).rev().collect(); + write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "kept.bin".into(), + }, + &plaintext, + ) + .unwrap(); + tick(&world, &engine, &mut tasks); + + let bob = world.device(b"alice-second-device"); + serve_http(&bob, &blocks, 400); + let (mut engine_b, _events_b) = engine_on(&bob, 7); + block_on(engine_b.start(secret())).unwrap(); + let kept = child_id(&engine_b, ROOT, "kept.bin"); + assert_eq!( + block_on(engine_b.read_content(kept)).expect("every leaf of the next version was sent"), + plaintext + ); +} + /// Cancel is guaranteed only until publish entry. Once the version's record has /// published, the op has left the queue and a cancel is refused rather than /// converted into a compensating delete of published state. @@ -3965,6 +4042,56 @@ fn orphan_residue_is_collected_and_a_live_handles_blocks_are_not() { ); } +/// A release that reports done without dropping the bytes strands a staged leaf +/// (#924's residue shape). On the cancel path nothing re-runs that release — the +/// op is gone from the queue — so orphan GC is the only thing that reclaims it, +/// and it does so precisely because nothing references it any more. +#[test] +fn a_leaf_a_lost_release_stranded_on_a_cancel_is_reclaimed_by_the_next_sweep() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + &(0..200u8).collect::>(), + ) + .unwrap(); + // The last leaf: far past where the cancel interrupts the upload, so the + // drain never removes it and the facade's release is its only cleaner. + let version = queued_version(&alice, op_id); + let stranded = version.last().expect("a multi-leaf version").clone(); + + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + alice.staging_store.drop_staged_removal_after(&stranded, 0); + block_on(engine.command(Command::CancelUpload { op_id })).unwrap(); + assert!( + block_on(alice.staging_store.staged_keys()) + .unwrap() + .contains(&stranded), + "the fixture must actually strand a leaf, or the sweep has nothing to prove" + ); + + // The pass that notices the cancel sweeps behind itself, so the residue does + // not wait a whole cadence. + poll_each(&mut tasks); + assert!( + !block_on(alice.staging_store.staged_keys()) + .unwrap() + .contains(&stranded), + "nothing references it once its op is gone, so the sweep takes it" + ); +} + /// A terminally unrebasable op keeps its staged bytes — and keeping them is only /// real if they survive the cold start that drops the op record, and the GC pass /// that runs there (#853). @@ -4224,12 +4351,17 @@ fn uploaded_cids(device: &FakeDevice) -> Vec { /// Every target this device has asked the registry to retire, in order. fn retire_targets(device: &FakeDevice) -> Vec { + retire_batches(device).into_iter().flatten().collect() +} + +/// The retire calls this device made, one entry per batch. +fn retire_batches(device: &FakeDevice) -> Vec> { device .http .requests() .iter() .filter(|request| request.url.ends_with("/registry/retire")) - .flat_map(|request| { + .map(|request| { let body = request .body .as_deref() From ed61e9f3dfd4c86ba50c110b89e6ecb64a67aa5c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 07:17:54 +0000 Subject: [PATCH 4/5] fix(engine): gate the drain's cancel retire on the op leaving the queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The facade publishes the cancel claim before its durable removal commits, so a pass that stops on that claim cannot assume the op is gone. It retired the confirmed set unconditionally, and a removal that then failed put the op back in play with its leading leaves unpinned — the version publishes naming blocks nobody holds. The retire now follows the facade's own rule and runs only behind a removal of its own. Also emit SnapshotUpdated when a cancel's cascade step fails, guard the mid-transfer precondition in the two cancellation tests that lacked it, cover the TooLateToCancel and NotAnUpload host mappings, and document the notAnUpload rejection on the cancelUpload contract. --- crates/engine/src/facade.rs | 17 ++++++---- crates/engine/src/sync/drain.rs | 12 ++++++- crates/engine/tests/write_plane.rs | 54 ++++++++++++++++++++++++++++++ crates/fuse/src/error.rs | 4 +++ crates/wasm/src/lib.rs | 1 + packages/client/src/facade.ts | 3 +- 6 files changed, 82 insertions(+), 9 deletions(-) diff --git a/crates/engine/src/facade.rs b/crates/engine/src/facade.rs index 1730ede58..57f9290cf 100644 --- a/crates/engine/src/facade.rs +++ b/crates/engine/src/facade.rs @@ -2224,17 +2224,20 @@ impl Engine { self.cancels.borrow_mut().withdraw(op_id); return Err(error); } + // The primary op is already gone, so the overlay is stale either way: + // the host is told even when a cascade step fails part way. + let mut cascaded = Ok(()); for (later_id, later) in cascade { - match later.content_root_cid() { - Some(root_cid) => { - self.discard_upload(later_id, later.target, root_cid) - .await? - } - None => self.dequeue_op(later_id).await?, + cascaded = match later.content_root_cid() { + Some(root_cid) => self.discard_upload(later_id, later.target, root_cid).await, + None => self.dequeue_op(later_id).await, + }; + if cascaded.is_err() { + break; } } let _ = self.events.unbounded_send(Event::SnapshotUpdated); - Ok(()) + cascaded } /// Undo one queued upload: drop the op, retire what of it reached the diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 57b1ba243..7d6142eb1 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -523,7 +523,17 @@ where // with nothing left to reach it, so the complete set is retired // here. Idempotent, so the overlap with the facade's batch is a // no-op (#916). - Halt::Cancelled => self.retire_cancelled(op_id).await, + // + // The dequeue gates the retire on the rule the facade's own path + // follows: the claim is published before that removal commits, so + // an op reaching here may still be queued — and unpinning the + // leading leaves of something still publishable would land a + // version whose blocks are gone (#824). + Halt::Cancelled => { + if self.dequeue_op(op_id).await.is_ok() { + self.retire_cancelled(op_id).await; + } + } Halt::Attempt | Halt::UploadAttempt => { if attempts.charge(op_id) < ATTEMPT_BUDGET { return; diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 271c837e5..7af42dc01 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -3705,6 +3705,10 @@ fn a_cancelled_versions_upload_mark_never_counts_towards_the_next_one() { for _ in 0..4 { poll_once(&mut tasks); } + assert!( + uploads(&alice) > 0, + "only a partial upload leaves a mark, so the cancel must land mid-transfer" + ); block_on(engine.command(Command::CancelUpload { op_id: cancelled })).unwrap(); poll_each(&mut tasks); @@ -3933,6 +3937,52 @@ fn a_cancel_that_cannot_dequeue_retires_nothing_and_leaves_the_op_publishable() ); } +/// The facade publishes the cancel claim before its removal commits, so the pass +/// that stops on that claim cannot assume the op has left the queue. Its retire +/// is gated on a removal of its own: proving the op is gone is what makes +/// unpinning its leaves safe, and a removal it cannot make means it retires +/// nothing rather than stranding a still-publishable version (#824). +#[test] +fn the_drains_cancel_retire_is_gated_on_the_op_leaving_the_durable_queue() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + let alice = world.device(b"alice"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 42); + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: ROOT, + name: "photo.bin".into(), + }, + &(0..200u8).collect::>(), + ) + .unwrap(); + + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + assert!( + uploads(&alice) > 0, + "the cancel must land mid-transfer, not before it started" + ); + block_on(engine.command(Command::CancelUpload { op_id })).expect("the upload is cancellable"); + let facade_batches = retire_batches(&alice).len(); + + // The drain now stops on the claim with no removal available to it, which + // is indistinguishable from the op never having left the queue. + alice.staging_store.fail_remove_op(); + poll_each(&mut tasks); + + assert_eq!( + retire_batches(&alice).len(), + facade_batches, + "a pass that cannot prove the op left the queue unpins nothing" + ); +} + /// The op-record header is clear and unauthenticated, and the owner tag on it is /// a public key any co-tenant of the origin-shared store can copy. A record that /// bears our tag but never opens is dead-lettered and dropped at cold start — it @@ -4072,6 +4122,10 @@ fn a_leaf_a_lost_release_stranded_on_a_cancel_is_reclaimed_by_the_next_sweep() { for _ in 0..4 { poll_once(&mut tasks); } + assert!( + uploads(&alice) > 0, + "the cancel must land mid-transfer, not before it started" + ); alice.staging_store.drop_staged_removal_after(&stranded, 0); block_on(engine.command(Command::CancelUpload { op_id })).unwrap(); assert!( diff --git a/crates/fuse/src/error.rs b/crates/fuse/src/error.rs index f58851105..7a3806971 100644 --- a/crates/fuse/src/error.rs +++ b/crates/fuse/src/error.rs @@ -122,6 +122,8 @@ impl std::error::Error for VfsError {} #[cfg(test)] mod tests { + use cipherbox_engine::seams::OpId; + use super::*; #[test] @@ -201,6 +203,8 @@ mod tests { EngineError::NotStarted, EngineError::AlreadyStarted, EngineError::InvalidSecret, + EngineError::TooLateToCancel { op_id: OpId(1) }, + EngineError::NotAnUpload { op_id: OpId(2) }, EngineError::Unimplemented { command: "grant" }, EngineError::Seam { message: "fsync failed".into(), diff --git a/crates/wasm/src/lib.rs b/crates/wasm/src/lib.rs index 58e024217..452972424 100644 --- a/crates/wasm/src/lib.rs +++ b/crates/wasm/src/lib.rs @@ -532,6 +532,7 @@ impl Command { } /// Cancel a queued upload by the op id `commitWrite` returned. Rejects with + /// `notAnUpload` when the op carries no content, and with /// `tooLateToCancel` once the version's record is publishing. #[wasm_bindgen(js_name = cancelUpload)] pub fn cancel_upload(op_id: u64) -> Command { diff --git a/packages/client/src/facade.ts b/packages/client/src/facade.ts index e573f7a4b..be3f2f72b 100644 --- a/packages/client/src/facade.ts +++ b/packages/client/src/facade.ts @@ -110,7 +110,8 @@ export class EngineFacade { /** * Cancel a queued upload by the op id `commitWrite` resolved with. Rejects - * with `tooLateToCancel` once the version's record is publishing. + * with `notAnUpload` when the op carries no content, and with + * `tooLateToCancel` once the version's record is publishing. */ cancelUpload(opId: bigint): Promise { return this.command({ kind: 'cancelUpload', opId }); From bed26ae70cd9e8acc8af3adf002d331aa757f999 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sun, 2 Aug 2026 20:16:13 +0200 Subject: [PATCH 5/5] test(engine): pin the focus refresh against a cancelled upload The focus window's folder refresh runs before the drain each pass and merges a folder's published children into the base, so it is worth holding the line that it cannot carry a cancelled upload back into the folder the user is looking at. The fixture authors the folder on one device and focuses it on another, because the refresh needs the folder's own ipnsName and only a resolved parent ref supplies it. A second writer's child is planted between the two passes so the assertion fails if the refresh never ran at all. --- crates/engine/tests/write_plane.rs | 77 ++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 7af42dc01..1342f53b2 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -3582,6 +3582,83 @@ fn queued_version(device: &FakeDevice, op_id: OpId) -> Vec> { }) } +/// The focus window's folder refresh runs before the drain each pass (#945) and +/// merges a folder's *published* children into the base. A cancelled upload was +/// never published, so the refresh cannot carry it back into the folder the user +/// is looking at. +#[test] +fn a_focus_refresh_never_renders_back_an_upload_the_user_cancelled() { + let world = FakeWorld::new(); + let blocks = Blocks::default(); + seed_account(&world, &blocks); + + // Authored on one device and resolved on another, so the focusing device + // knows the folder's own name and its refresh really descends into it. + let author = world.device(b"alice"); + let (mut engine_a, _events_a, mut tasks_a) = boot(&world, &blocks, &author, 42); + block_on(engine_a.command(Command::Create { + parent: ROOT, + name: "photos".into(), + kind: NodeKind::Folder, + })) + .unwrap(); + tick(&world, &engine_a, &mut tasks_a); + let photos = child_id(&engine_a, ROOT, "photos"); + + let alice = world.device(b"alice-second-device"); + let (mut engine, _events, mut tasks) = boot(&world, &blocks, &alice, 7); + block_on(engine.command(Command::SetFocus { node: Some(photos) })).unwrap(); + + let op_id = write_file( + &mut engine, + WriteTarget::NewFile { + parent: photos, + name: "holiday.bin".into(), + }, + &(0..200u8).collect::>(), + ) + .unwrap(); + assert_eq!( + listed_names(&engine, photos), + vec!["holiday.bin".to_owned()] + ); + let version = queued_version(&alice, op_id); + + world.scheduler.advance(engine.profile().poll_cadence); + for _ in 0..4 { + poll_once(&mut tasks); + } + assert!(uploads(&alice) > 0, "the cancel lands mid-transfer"); + block_on(engine.command(Command::CancelUpload { op_id })).unwrap(); + // Finish the pass the cancel interrupted: drain, then sweep. + tick(&world, &engine, &mut tasks); + + // Another writer's child, discoverable only by the focus refresh — without + // it this test could pass with the refresh never running at all. + concurrent_add( + &world.record_store, + &blocks, + photos, + file_ref([0xC1; 16], "from-elsewhere.bin"), + ); + // The next pass refreshes the focused folder before its drain, which is the + // ordering the overlap turns on (#945). + tick(&world, &engine, &mut tasks); + + assert_eq!( + listed_names(&engine, photos), + vec!["from-elsewhere.bin".to_owned()], + "the refresh merged what published and left the cancelled upload gone" + ); + assert!( + published_names(&world.record_store, &blocks, photos) + .iter() + .all(|name| name != "holiday.bin"), + "the cancelled version never reaches the record plane" + ); + assert_no_blocks_staged(&alice, &version); +} + /// The acceptance case: a cancel that lands while the drain is mid-upload stops /// it at the next block boundary, releases every block of the version, retires /// what already reached the network, and publishes nothing.