diff --git a/blueprint/engine.md b/blueprint/engine.md index c8baf97888..99debbf2a0 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, 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 168e209f2f..2c7f0258a7 100644 --- a/crates/contract/tests/contract.rs +++ b/crates/contract/tests/contract.rs @@ -799,6 +799,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 = leaf_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 0152fd882b..fae30a4485 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 149856b111..9e93c2caad 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,42 @@ 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. + /// + /// 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 + .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()); + } + } + + /// 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() { + let mut orphans = self.orphan_heads.borrow_mut(); + let sent = pending.len().min(orphans.len()); + orphans.drain(..sent); + } + } + /// 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 +1964,30 @@ 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 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(_) => 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). + 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. @@ -2058,6 +2133,62 @@ 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), + false, + ), + ( + RecordPublishError::Publish(PublishError::FloorRead(crate::seams::SeamError::new( + "floor", + ))), + 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] diff --git a/crates/engine/tests/write_plane.rs b/crates/engine/tests/write_plane.rs index 8c28534e6c..4d62119e23 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.