fix(engine): retire the head block a publish orphaned before the transport

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
 

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<Vec<u8>> = (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

crates/engine/src/facade.rs

@@ -1146,6 +1146,10 @@ pub struct Engine<T: SeamTypes> {
     /// [`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<RefCell<Option<BlockedOp>>>,
+    /// 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<RefCell<Vec<String>>>,
     /// 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<Cell<bool>>,
@@ -1198,6 +1202,7 @@ impl<T: SeamTypes> Engine<T> {
                 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<T: SeamTypes> Engine<T> {
         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<T: SeamTypes> Engine<T> {
                         base: &base,
                         held: &held,
                         blocked: &blocked,
+                        orphan_heads: &orphan_heads,
                         events: &events,
                     }
                     .run(&DrainScope {

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<Option<BlockedOp>>,
+    /// Head blocks this session's publishes orphaned, pending retirement
+    /// ([`Drain::retire_orphan_heads`]).
+    pub(crate) orphan_heads: &'a RefCell<Vec<String>>,
     /// The facade's outbound event stream, for upload progress.
     pub(crate) events: &'a mpsc::UnboundedSender<Event>,
 }
@@ -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]