From 3a8e396899753c542688ac7610e22e00fa5a9005 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 11:48:20 +0200 Subject: [PATCH 1/4] fix(engine): retire the head block a publish orphaned before the transport Every record publish uploads its head block through the same charged POST /content/upload a content block goes through, so each attempt creates its own accountable pin row. A publish that failed before the record reached the transport left that row charged against an account that could never reach the block again - and because the drain draws a fresh seal nonce per pass, every retry orphaned another one. Nothing retired any of them, for any op kind: #916 taught abandonment to retire the write name and the content blocks, and folder creates, deletes, renames and moves have no content blocks at all. The drain now notes the head CID of any publish that stopped short of the PUT fan-out - register-first, the floor read, the head-CID echo, a refused upload aside - and retires the set at the end of the pass that orphaned it, independent of the op's fate. An op that retries and then succeeds charged those rows just as surely as one that abandons. A fan-out that acknowledged nothing is deliberately not one of them: AllEndpointsFailed says no endpoint acked, not that none stored, and unpinning the head of a record a lost ack left resolvable is the loss #916 refuses, where leaving the row charged is only a leak. Closes #921 --- blueprint/engine.md | 8 ++- crates/contract/tests/contract.rs | 79 ++++++++++++++++++++++ crates/engine/src/facade.rs | 7 ++ crates/engine/src/sync/drain.rs | 66 ++++++++++++++++-- crates/engine/tests/write_plane.rs | 103 ++++++++++++++++++++++++++++- 5 files changed, 257 insertions(+), 6 deletions(-) diff --git a/blueprint/engine.md b/blueprint/engine.md index c8baf9788..678772444 100644 --- a/blueprint/engine.md +++ b/blueprint/engine.md @@ -142,7 +142,13 @@ bytes (#28 D2). target that never landed costs nothing. An op whose record PUT was **acknowledged** retires nothing: the record may be resolvable at its name, and unpinning content a live record still references is loss, where leaving the - rows charged is only a leak. + rows charged is only a leak. A publish that fails **before the record reaches + the transport** — register-first, the floor read, the head-CID echo — is the + mirror case: its head block already uploaded under its own charged row, no + record can name it, and the retry re-authors under a fresh seal nonce, so the + drain retires that head at the end of the pass that orphaned it, per attempt. + A fan-out that acknowledged nothing does **not** qualify: no ack is not proof + nothing stored. ## Adoption gate and floors diff --git a/crates/contract/tests/contract.rs b/crates/contract/tests/contract.rs index 168e209f2..a70a5693f 100644 --- a/crates/contract/tests/contract.rs +++ b/crates/contract/tests/contract.rs @@ -482,6 +482,12 @@ fn leaf_cid(bytes: &[u8]) -> String { encode_content_cid_str(&compute_cid(CONTENT_CID_CODEC, bytes)) } +/// The engine's own content address for a head block — the `dag-cbor` codec a +/// record `Value` spells, the address `PreflightedHead` publishes under. +fn head_cid(bytes: &[u8]) -> String { + encode_content_cid_str(&compute_cid(DAG_ROOT_CODEC, bytes)) +} + /// Fetch one block back from the stack's trustless gateway by CID — the read /// path's view of what the ingress actually pinned. async fn fetch_block(cid: &str) -> Vec { @@ -799,6 +805,79 @@ async fn an_abandoned_versions_whole_block_set_retires_back_to_the_pre_upload_fi ); } +/// A record's head block goes up through the same charged ingress a content +/// block does, and every publish attempt authors its own under a fresh seal +/// nonce — so an op that retried left one charged, unreferenced head row per +/// attempt. Retiring only the last of them leaves the rest spending the quota +/// that refuses later uploads (#921). +#[tokio::test] +async fn every_head_block_a_retrying_publish_orphaned_retires_back_to_the_pre_upload_figure() { + let base = require_stack!( + "every_head_block_a_retrying_publish_orphaned_retires_back_to_the_pre_upload_figure" + ); + let client = fresh_account(&base).await; + + // Three attempts at one record: byte-different heads under distinct + // addresses, each uploaded and then registered under the same name the way + // register-first composes it. + let name = "k51contractOrphanedHeads".to_owned(); + let attempts: Vec> = (0..3u8).map(|i| vec![0xC0 | i; 96]).collect(); + let mut heads = Vec::new(); + for block in &attempts { + let declared = head_cid(block); + let uploaded = client + .upload(&declared, block) + .await + .unwrap_or_else(|e| panic!("a head block uploads: {e:?}")); + assert_eq!( + uploaded.cid, declared, + "a head block pins under the address the drain computed" + ); + client + .register(&[NameRegistration { + ipns_name: name.clone(), + head_cid: Some(declared.clone()), + content_cids: Vec::new(), + }]) + .await + .expect("register-first names the head the attempt authored"); + heads.push(declared); + } + assert_eq!( + client.quota().await.expect("quota after upload").used_bytes, + (attempts.len() * 96) as u64, + "each attempt's head charges the account on its own" + ); + + // Retiring only the head the last attempt registered is what the leak looks + // like: the earlier two stay charged. + client + .retire(&heads[2..]) + .await + .expect("retire the last attempt's head"); + assert_eq!( + client + .quota() + .await + .expect("quota after a partial retire") + .used_bytes, + (2 * 96) as u64, + "the heads the earlier attempts orphaned are still charged" + ); + + let mut targets = heads[..2].to_vec(); + targets.push(name); + client + .retire(&targets) + .await + .expect("retire every head the retries orphaned"); + assert_eq!( + client.quota().await.expect("quota after retire").used_bytes, + 0, + "retiring every orphaned head returns the account to its pre-upload figure" + ); +} + /// The retire batch is bounded fail-closed (blueprint/api.md, "Batch bounds"): /// an oversize array is refused, never truncated or partially applied. That /// refusal is what makes the engine's client-side chunking mandatory rather than diff --git a/crates/engine/src/facade.rs b/crates/engine/src/facade.rs index 0152fd882..fae30a448 100644 --- a/crates/engine/src/facade.rs +++ b/crates/engine/src/facade.rs @@ -1146,6 +1146,10 @@ pub struct Engine { /// [`snapshot`](Self::snapshot). In-memory: a restart re-derives it from the /// next drain attempt's own 413 rather than trusting a stale verdict. blocked: Rc>>, + /// Head blocks the drain uploaded for a publish that never reached the + /// record transport, pending retirement. Session-lived so a retire the + /// registry refused goes out again on a later pass (#921). + orphan_heads: Rc>>, /// Session-alive latch: cleared on drop so the spawned liveness loop /// stops at its next wake instead of re-PUTting after the engine is gone. alive: Rc>, @@ -1198,6 +1202,7 @@ impl Engine { dead_letters: Rc::new(RefCell::new(BTreeMap::new())), queue_scan: RefCell::new(QueueScanMemo::default()), blocked: Rc::new(RefCell::new(None)), + orphan_heads: Rc::new(RefCell::new(Vec::new())), alive: Rc::new(Cell::new(true)), session: None, api: None, @@ -1510,6 +1515,7 @@ impl Engine { let scope_write_seeds = self.scope_write_seeds.clone(); let dead_letters = self.dead_letters.clone(); let blocked = self.blocked.clone(); + let orphan_heads = self.orphan_heads.clone(); let transport = self.seams.record_transport.clone(); let snapshot_cache = self.seams.snapshot_cache.clone(); let floors = self.seams.floor_store.clone(); @@ -1630,6 +1636,7 @@ impl Engine { base: &base, held: &held, blocked: &blocked, + orphan_heads: &orphan_heads, events: &events, } .run(&DrainScope { diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 149856b11..8c0f360af 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -48,8 +48,8 @@ use crate::net::record_publish::{ }; use crate::net::retire::retire; use crate::net::{ - Adopter, ChildAdopter, HeldRecord, HeldRecords, LocalHead, ResolveOutcome, RootAdopter, - assemble_head_envelope, fanout_get_verify, resolve, + Adopter, ChildAdopter, HeldRecord, HeldRecords, LocalHead, REGISTRY_BATCH_MAX, ResolveOutcome, + RootAdopter, assemble_head_envelope, fanout_get_verify, resolve, }; use crate::profile::SyncTimingProfile; use crate::rotation::derive_write_name; @@ -293,6 +293,9 @@ pub(crate) struct Drain<'a, T, H: Http, C: CredentialStore, F, S, St, Sch> { /// The over-quota hold, shared with the facade's read surface. It clears /// only here, on a quota probe reporting room. pub(crate) blocked: &'a RefCell>, + /// Head blocks this session's publishes orphaned, pending retirement + /// ([`Drain::retire_orphan_heads`]). + pub(crate) orphan_heads: &'a RefCell>, /// The facade's outbound event stream, for upload progress. pub(crate) events: &'a mpsc::UnboundedSender, } @@ -412,8 +415,15 @@ where Sch: Scheduler + Clone + 'static, { /// Run one pass: rebase the queue onto gate-passing state and publish every - /// applied op it can, stopping at the first it cannot. + /// applied op it can, stopping at the first it cannot, then clear what the + /// pass orphaned. pub(crate) async fn run(&self, scope: &DrainScope<'_>) -> DrainReport { + let report = self.drain_queue(scope).await; + self.retire_orphan_heads().await; + report + } + + async fn drain_queue(&self, scope: &DrainScope<'_>) -> DrainReport { let mut report = DrainReport::default(); let Ok(Queue { mine, all_ids }) = self.queued_ops(scope, &mut report).await else { return report; @@ -1686,7 +1696,12 @@ where }, ) .await - .map_err(|error| classify_publish(error, head.block.len() as u64))?; + .map_err(|error| { + if orphaned_head(&error) { + self.record_orphan_head(preflighted.cid()); + } + classify_publish(error, head.block.len() as u64) + })?; match outcome { PublishOutcome::Published { .. } => Ok(record_bytes), // Both burned a CAS sequence at this name without a record we could @@ -1725,6 +1740,28 @@ where self.staging.remove_op(op_id).await.map_err(seam) } + /// Note one head block as orphaned, capped at [`REGISTRY_BATCH_MAX`] so a + /// session whose retires keep failing bounds its leak, not its memory. + fn record_orphan_head(&self, cid: &str) { + let mut orphans = self.orphan_heads.borrow_mut(); + if orphans.len() < REGISTRY_BATCH_MAX { + orphans.push(cid.to_owned()); + } + } + + /// Retire the head blocks this session's publishes orphaned + /// ([`orphaned_head`]). A refused retire keeps them pending for the next + /// pass rather than losing the only record of what to retire. + async fn retire_orphan_heads(&self) { + let pending = self.orphan_heads.borrow().clone(); + if pending.is_empty() { + return; + } + if retire(self.api, &pending).await.is_ok() { + self.orphan_heads.borrow_mut().drain(..pending.len()); + } + } + /// 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> { @@ -1913,6 +1950,27 @@ fn classify_register(error: ApiError) -> Halt { } } +/// Whether a failed publish left its head block charged and unreachable: the +/// upload landed under its own pin row, no record naming it reached the +/// transport, and the retry re-authors under a fresh seal nonce +/// (blueprint/engine.md "Resolve/publish pipeline: Retirement", #921). +fn orphaned_head(error: &RecordPublishError) -> bool { + match error { + // A refused upload charged no row. + RecordPublishError::Upload(_) => false, + RecordPublishError::HeadCidMismatch { .. } => true, + RecordPublishError::Publish(error) => match error { + PublishError::Register(_) | PublishError::FloorRead(_) | PublishError::EmptyHeadCid => { + true + } + // No ack is not proof nothing stored: unpinning a head a live + // record may still name is loss, where the row is only a leak + // (#916). + PublishError::AllEndpointsFailed => false, + }, + } +} + /// Classify a content-upload failure for the valve. The same server verdicts a /// head-block upload can carry, since content blocks and head blocks go through /// one endpoint. diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 8c28534e6..4d62119e2 100644 --- a/crates/engine/tests/write_plane.rs +++ b/crates/engine/tests/write_plane.rs @@ -191,10 +191,16 @@ impl Blocks { } /// Answer every registration with a 400 carrying `body` instead of acking. + /// Retirement keeps answering, so a pass can still clear what it orphaned. fn refuse_register(&self, body: Vec) { *self.register_refusal.lock().expect("lock") = Some(body); } + /// Let every registration through again. + fn accept_registrations(&self) { + *self.register_refusal.lock().expect("lock") = None; + } + /// Answer one engine HTTP call: a content upload lands its bytes here and /// echoes their address, a registry call acks, and a gateway GET serves the /// block back. Enqueued as many times as the pass needs, so no test depends @@ -3438,6 +3444,88 @@ fn an_unconfirmed_publish_never_retires_the_version_it_may_already_name() { ); } +/// Register-first stops a publish only after its head block has uploaded and +/// charged its own pin row, and each attempt re-authors under a fresh seal +/// nonce — so a retrying op orphans a byte-different head every pass. Each +/// leaves the inventory on the pass that orphaned it, and the abandonment still +/// owes back the name on top (#921). The refusal is an intermediary's `400`, so +/// it is charged rather than permanent and the op survives to retry. +#[test] +fn every_head_block_a_retrying_op_orphaned_leaves_the_inventory() { + 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); + + create(&mut engine, "photos"); + let photos = child_id(&engine, ROOT, "photos"); + blocks.refuse_register(proxy_400()); + let mut heads = Vec::new(); + for attempt in 1..=3 { + tick(&world, &engine, &mut tasks); + heads = uploaded_cids(&alice); + assert_eq!(heads.len(), attempt, "one head block per attempt"); + assert_eq!( + retire_targets(&alice), + heads, + "an orphaned head leaves the inventory on its own pass, not on a retry that \ + re-authors" + ); + } + assert_eq!( + heads + .iter() + .collect::>() + .len(), + heads.len(), + "every attempt authored its own head under a fresh nonce" + ); + + // The op kept its place; a permanent refusal now abandons it. + blocks.accept_registrations(); + blocks.refuse_upload(Box::new(|_| Some(upload_413(Some("UPLOAD_TOO_LARGE"))))); + tick(&world, &engine, &mut tasks); + + let mut expected = heads; + expected.push(write_name(photos).as_str().to_owned()); + assert_eq!( + retire_targets(&alice), + expected, + "the abandonment owes back the name on top of every head the retries charged" + ); +} + +/// A PUT fan-out that acknowledges nothing is not proof that nothing stored: +/// an endpoint may hold the record and have lost its ack. Retiring that head +/// would unpin the block a record still resolvable at the name points at — +/// loss, where leaving the row charged is only a leak (#916 as extended by +/// #921). +#[test] +fn a_publish_that_reached_the_transport_never_retires_its_head() { + 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); + + create(&mut engine, "photos"); + for endpoint in world.record_store.endpoints() { + world.record_store.fail_put_endpoint(&endpoint); + } + tick(&world, &engine, &mut tasks); + + assert_eq!( + uploads(&alice), + 1, + "the head block went up before the fan-out refused it" + ); + assert!( + retire_targets(&alice).is_empty(), + "a head the record plane may already hold is not ours to unpin" + ); +} + /// 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; @@ -3612,12 +3700,25 @@ fn events_so_far(events: &mut EventStream) -> Vec { /// How many content uploads this device has sent. fn uploads(device: &FakeDevice) -> usize { + uploaded_cids(device).len() +} + +/// The address each of this device's uploads declared, in order. +fn uploaded_cids(device: &FakeDevice) -> Vec { device .http .requests() .iter() .filter(|request| request.url.ends_with("/content/upload")) - .count() + .map(|request| { + request + .headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case("X-Content-Cid")) + .map(|(_, value)| value.clone()) + .expect("an upload declares its CID") + }) + .collect() } /// Every target this device has asked the registry to retire, in order. From 318eea5f191b8f2c6f6034bec90e59c41e5f3db3 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 11:51:28 +0200 Subject: [PATCH 2/4] fix(engine): orphan a head whose upload ack never came back A refused upload charged no pin row, but a dropped connection or an unreadable 2xx body may have left one behind: the server committed the pin and the client never learned the address was live. The retry authors a fresh head, so nothing revisits that CID and the row is charged forever. Only a status answer is the server's own refusal. Transport and decode failures now orphan their head like every other pre-transport failure, and a unit test pins each arm - the AllEndpointsFailed arm above all, where retiring would be loss rather than a leak. --- crates/engine/src/sync/drain.rs | 57 +++++++++++++++++++++++++++++++-- 1 file changed, 55 insertions(+), 2 deletions(-) diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 8c0f360af..4c45130d1 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -1956,8 +1956,11 @@ fn classify_register(error: ApiError) -> Halt { /// (blueprint/engine.md "Resolve/publish pipeline: Retirement", #921). fn orphaned_head(error: &RecordPublishError) -> bool { match error { - // A refused upload charged no row. - RecordPublishError::Upload(_) => false, + // A status answer is the server's own refusal, so it charged no row; a + // dropped connection or an unreadable 2xx may have left one behind. + RecordPublishError::Upload(error) => { + matches!(error, ApiError::Transport(_) | ApiError::Decode(_)) + } RecordPublishError::HeadCidMismatch { .. } => true, RecordPublishError::Publish(error) => match error { PublishError::Register(_) | PublishError::FloorRead(_) | PublishError::EmptyHeadCid => { @@ -2116,6 +2119,56 @@ mod tests { } } + /// The destruction-critical arm: a fan-out that acked nothing may still + /// have stored the record, so its head stays pinned. Everything else here + /// stopped short of the transport with a charged row behind it, or with no + /// row at all. + #[test] + fn only_a_publish_that_never_reached_the_transport_orphans_its_head() { + use RecordPublishError::Upload; + for (error, orphaned) in [ + ( + RecordPublishError::Publish(PublishError::AllEndpointsFailed), + false, + ), + ( + RecordPublishError::Publish(PublishError::Register(ApiError::NotAuthenticated)), + true, + ), + ( + RecordPublishError::Publish(PublishError::EmptyHeadCid), + true, + ), + ( + RecordPublishError::HeadCidMismatch { + expected: "a".to_owned(), + returned: "b".to_owned(), + }, + true, + ), + ( + Upload(ApiError::Status { + status: 413, + message: None, + code: Some(UPLOAD_TOO_LARGE.to_owned()), + }), + false, + ), + (Upload(ApiError::NotAuthenticated), false), + ( + Upload(ApiError::Transport(crate::seams::SeamError::new("dropped"))), + true, + ), + (Upload(ApiError::Decode("short body".to_owned())), true), + ] { + assert_eq!( + orphaned_head(&error), + orphaned, + "{error:?} orphans its head block: {orphaned}" + ); + } + } + /// Every other publish failure is availability: retried indefinitely and /// charged nothing, so an unreachable network never abandons an op. #[test] From c641990e17800b0ba4851b30c1a30f99d4574941 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 12:00:45 +0200 Subject: [PATCH 3/4] fix(engine): check the live set before queueing a head for retirement The orphan queue's only consumer physically unpins at refcount zero, so deciding from the publish error alone put a destructive step one careless caller away from unpinning a head a resolvable record names. The check now runs where the damage would be done: a CID the held set carries is refused outright, whatever the error said. Also drops EmptyHeadCid from the orphan set - it means nothing was ever addressed, so there is no CID to retire - and bounds the post-retire drain by the queue's own length. The contract leg declared its stand-in head blocks under the dag-cbor codec, which the ingress could not pin because 96 bytes of filler is not valid CBOR. It uses the same raw address the neighbouring version-set leg uses; the byte accounting the test asserts does not turn on the codec. --- blueprint/engine.md | 12 ++++++------ crates/contract/tests/contract.rs | 8 +------- crates/engine/src/sync/drain.rs | 28 ++++++++++++++++++++++++---- 3 files changed, 31 insertions(+), 17 deletions(-) diff --git a/blueprint/engine.md b/blueprint/engine.md index 678772444..99debbf2a 100644 --- a/blueprint/engine.md +++ b/blueprint/engine.md @@ -143,12 +143,12 @@ bytes (#28 D2). **acknowledged** retires nothing: the record may be resolvable at its name, and unpinning content a live record still references is loss, where leaving the rows charged is only a leak. A publish that fails **before the record reaches - the transport** — register-first, the floor read, the head-CID echo — is the - mirror case: its head block already uploaded under its own charged row, no - record can name it, and the retry re-authors under a fresh seal nonce, so the - drain retires that head at the end of the pass that orphaned it, per attempt. - A fan-out that acknowledged nothing does **not** qualify: no ack is not proof - nothing stored. + the transport** — register-first, the floor read, the head-CID echo, or an + upload whose ack never came back — is the mirror case: its head block may + already be pinned under its own charged row, no record can name it, and the + retry re-authors under a fresh seal nonce, so the drain retires that head at + the end of the pass that orphaned it, per attempt. A fan-out that + acknowledged nothing does **not** qualify: no ack is not proof nothing stored. ## Adoption gate and floors diff --git a/crates/contract/tests/contract.rs b/crates/contract/tests/contract.rs index a70a5693f..2c7f0258a 100644 --- a/crates/contract/tests/contract.rs +++ b/crates/contract/tests/contract.rs @@ -482,12 +482,6 @@ fn leaf_cid(bytes: &[u8]) -> String { encode_content_cid_str(&compute_cid(CONTENT_CID_CODEC, bytes)) } -/// The engine's own content address for a head block — the `dag-cbor` codec a -/// record `Value` spells, the address `PreflightedHead` publishes under. -fn head_cid(bytes: &[u8]) -> String { - encode_content_cid_str(&compute_cid(DAG_ROOT_CODEC, bytes)) -} - /// Fetch one block back from the stack's trustless gateway by CID — the read /// path's view of what the ingress actually pinned. async fn fetch_block(cid: &str) -> Vec { @@ -824,7 +818,7 @@ async fn every_head_block_a_retrying_publish_orphaned_retires_back_to_the_pre_up let attempts: Vec> = (0..3u8).map(|i| vec![0xC0 | i; 96]).collect(); let mut heads = Vec::new(); for block in &attempts { - let declared = head_cid(block); + let declared = leaf_cid(block); let uploaded = client .upload(&declared, block) .await diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 4c45130d1..4d45d6f23 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -1742,7 +1742,19 @@ where /// Note one head block as orphaned, capped at [`REGISTRY_BATCH_MAX`] so a /// session whose retires keep failing bounds its leak, not its memory. + /// + /// A head the live set still names is refused outright: this queue's only + /// consumer physically unpins, so the destructive step checks the live set + /// itself rather than trusting the caller to hand it a fresh head. fn record_orphan_head(&self, cid: &str) { + if self + .held + .borrow() + .values() + .any(|record| record.head_cid == cid) + { + return; + } let mut orphans = self.orphan_heads.borrow_mut(); if orphans.len() < REGISTRY_BATCH_MAX { orphans.push(cid.to_owned()); @@ -1758,7 +1770,9 @@ where return; } if retire(self.api, &pending).await.is_ok() { - self.orphan_heads.borrow_mut().drain(..pending.len()); + let mut orphans = self.orphan_heads.borrow_mut(); + let sent = pending.len().min(orphans.len()); + orphans.drain(..sent); } } @@ -1963,9 +1977,9 @@ fn orphaned_head(error: &RecordPublishError) -> bool { } RecordPublishError::HeadCidMismatch { .. } => true, RecordPublishError::Publish(error) => match error { - PublishError::Register(_) | PublishError::FloorRead(_) | PublishError::EmptyHeadCid => { - true - } + PublishError::Register(_) | PublishError::FloorRead(_) => true, + // Nothing was ever addressed, so there is no CID to retire. + PublishError::EmptyHeadCid => false, // No ack is not proof nothing stored: unpinning a head a live // record may still name is loss, where the row is only a leak // (#916). @@ -2137,6 +2151,12 @@ mod tests { ), ( RecordPublishError::Publish(PublishError::EmptyHeadCid), + false, + ), + ( + RecordPublishError::Publish(PublishError::FloorRead(crate::seams::SeamError::new( + "floor", + ))), true, ), ( From 275fda6508a70afb90116c826c72fb7bd5e519b1 Mon Sep 17 00:00:00 2001 From: Michael Yankelev Date: Sat, 1 Aug 2026 19:23:03 +0200 Subject: [PATCH 4/4] docs(engine): say where the orphan queue's live-set check actually runs The doc comment claimed the destructive step checks the live set. It does not: the check runs at enqueue, in record_orphan_head. State the guard the code performs and why refusing a live head matters, and drop the claim about a check that lives nowhere. --- crates/engine/src/sync/drain.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/engine/src/sync/drain.rs b/crates/engine/src/sync/drain.rs index 4d45d6f23..9e93c2caa 100644 --- a/crates/engine/src/sync/drain.rs +++ b/crates/engine/src/sync/drain.rs @@ -1743,9 +1743,9 @@ where /// Note one head block as orphaned, capped at [`REGISTRY_BATCH_MAX`] so a /// session whose retires keep failing bounds its leak, not its memory. /// - /// A head the live set still names is refused outright: this queue's only - /// consumer physically unpins, so the destructive step checks the live set - /// itself rather than trusting the caller to hand it a fresh head. + /// A head the live set still names never enters the queue: its only + /// consumer physically unpins, and unpinning a head a live record names is + /// loss, where leaving the row charged is only a leak. fn record_orphan_head(&self, cid: &str) { if self .held