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..57f9290cf 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; @@ -37,6 +38,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 +51,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::cancel::UploadCancels; use crate::sync::drain::{Drain, DrainReport, DrainScope}; 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; @@ -60,7 +63,7 @@ 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::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}; @@ -364,6 +367,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 +465,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 +650,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 +815,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}]" @@ -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). + 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()); self.api = Some(api); @@ -1516,6 +1570,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 +1693,7 @@ impl Engine { held: &held, blocked: &blocked, orphan_heads: &orphan_heads, + cancels: &cancels, events: &events, } .run(&DrainScope { @@ -1650,6 +1707,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 +1827,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): @@ -1889,6 +1950,7 @@ impl Engine { writes.next += 1; let handle = WriteHandle(writes.next); + self.live_blocks.borrow_mut().open(handle); writes.open.insert( handle, LiveWrite { @@ -1951,11 +2013,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 +2035,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 +2053,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 +2070,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 +2119,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 +2190,103 @@ 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); + } + // 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 { + 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); + cascaded + } + + /// Undo one queued upload: drop the op, retire what of it reached the + /// network ([`UploadCancels`]), release its blocks, and tell the host. + /// + /// 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> { + 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), + 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) + } + /// 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..a1baf2194 --- /dev/null +++ b/crates/engine/src/sync/cancel.rs @@ -0,0 +1,174 @@ +//! 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; + +/// 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, + in_flight: 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.carrying(op_id).is_some_and(|it| it.past_publish_entry) { + 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.take_up(op_id).past_publish_entry = true; + true + } + + /// 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)] +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" + ); + + let mut cancels = UploadCancels::default(); + assert!(cancels.request(OpId(1))); + assert!( + !cancels.enter_publish(OpId(1)), + "the user cancelled first; the publish must abandon" + ); + } + + /// 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] + 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..7d6142eb1 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_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). @@ -189,6 +191,9 @@ enum Halt { /// resume probe must find room for. needed_bytes: u64, }, + /// The user cancelled the upload. The facade has already undone 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 +301,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 +472,12 @@ 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 — 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)); } @@ -486,6 +499,7 @@ where 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(()) @@ -503,6 +517,23 @@ where ) { match halt { 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). + // + // 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; @@ -1291,19 +1322,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,9 +1407,11 @@ where }; emit(OpPhase::UploadStarted, blocks(uploaded)); for (index, leaf_cid) in content.leaf_cids().iter().enumerate() { + 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 @@ -1382,10 +1431,14 @@ where None => {} } } + 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()); @@ -1395,6 +1448,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( @@ -1460,14 +1524,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 @@ -1481,29 +1537,15 @@ 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(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 @@ -1776,6 +1818,35 @@ 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`]). + 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> { @@ -1823,11 +1894,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(), }; @@ -1931,6 +1998,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 +2106,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. diff --git a/crates/engine/src/sync/mod.rs b/crates/engine/src/sync/mod.rs index a29fbcf56..5c2daedab 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; diff --git a/crates/engine/src/sync/staging.rs b/crates/engine/src/sync/staging.rs index d3d109f21..9d730095f 100644 --- a/crates/engine/src/sync/staging.rs +++ b/crates/engine/src/sync/staging.rs @@ -9,10 +9,19 @@ //! 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}; 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 +56,278 @@ pub async fn stage_op( store.enqueue_op(&record).await } +/// 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 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. 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(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; + } + 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(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(); + }; + if verify_cid(root_cid, &block).is_err() { + return Vec::new(); + } + decode_root(&block).map(|m| m.leaf_cids).unwrap_or_default() +} + +/// 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 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_dead_letter: unreadable preserved record"))?; + if kept.iter().any(|held| held == record) { + return Ok(()); + } + kept.push(record.to_vec()); + write_preserved_dead_letters(store, &kept).await +} + +/// 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 kept = Vec::new(); + while !rest.is_empty() { + let Some((len, tail)) = rest.split_at_checked(4) else { + return Ok(None); + }; + 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(kept)) +} + +/// 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_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 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. +/// +/// 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>>, + /// 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 { + /// Start holding `handle`'s staging keys. + pub(crate) fn open(&mut self, handle: WriteHandle) { + self.generation += 1; + self.by_handle.insert(handle, Vec::new()); + } + + /// 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()); + } + } + + /// 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() + } + + pub(crate) fn generation(&self) -> u64 { + self.generation + } +} + +/// 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_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 (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 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_dead_letters(store).await; +} + +/// 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(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()); + } + } + if live.len() != kept.len() { + let _ = write_preserved_dead_letters(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 +/// 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, 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 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_DEAD_LETTERS_KEY.to_vec(), ]); referenced.extend(live.iter().cloned()); - for (_, record) in &queued { + // 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 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. 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? { @@ -89,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)] @@ -105,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; @@ -397,6 +644,133 @@ 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; + 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!( + orphan_staging_keys(&store, &[]).await.unwrap(), + vec![b"orphan".to_vec()], + "the preserved root and every leaf it lists stay referenced" + ); + }); + } + + /// 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 a_preserved_dead_letter_still_carries_the_key_that_opens_its_version() { + let store = InMemoryStagingStore::default(); + block_on(async { + 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(); + + 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_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; + 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_dead_letters(&store).await.unwrap().unwrap(), + vec![held.clone(), collected] + ); + + prune_preserved_dead_letters(&store).await; + assert_eq!( + read_preserved_dead_letters(&store).await.unwrap().unwrap(), + vec![held], + "a dead letter whose blocks are gone preserves nothing" + ); + + release_version_blocks(&store, &root_cid).await; + prune_preserved_dead_letters(&store).await; + assert!( + store + .staged_bytes(PRESERVED_DEAD_LETTERS_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..1342f53b2 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,735 @@ 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 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. +#[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" + ); + // 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!( + (1..version.len()).contains(&charged.len()), + "the cancel landed mid-set: part of the version is charged, not all of it" + ); + // 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), + node: file, + phase: OpPhase::UploadCancelled, + progress: None, + error: None, + }), + "the host is told the upload was cancelled, keyed on its own op" + ); +} + +/// 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); + } + 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); + + 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. +#[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 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_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(), + }, + &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); + + 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 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 +/// 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 — +/// 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 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); + } + 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!( + 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). +#[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; @@ -3723,12 +4482,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() diff --git a/crates/fuse/src/error.rs b/crates/fuse/src/error.rs index c2eff9c2e..7a3806971 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(), }, @@ -120,6 +122,8 @@ impl std::error::Error for VfsError {} #[cfg(test)] mod tests { + use cipherbox_engine::seams::OpId; + use super::*; #[test] @@ -199,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/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..452972424 100644 --- a/crates/wasm/src/lib.rs +++ b/crates/wasm/src/lib.rs @@ -531,6 +531,16 @@ 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 { + 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..be3f2f72b 100644 --- a/packages/client/src/facade.ts +++ b/packages/client/src/facade.ts @@ -108,6 +108,15 @@ export class EngineFacade { return this.command({ kind: 'relink', node, newParent }); } + /** + * Cancel a queued upload by the op id `commitWrite` resolved with. Rejects + * 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 }); + } + 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 }