fix(storage): harden ODM and scanner publication (#7187)

* fix(storage): harden ODM and scanner publication * fix(app): simplify absent SSE configuration matching * test(heal): settle PUT rename tails before disk-wipe fixtures * fix(ecstore): remove duplicate local rename implementation Keep the canonical commit module after concurrent storage changes merged. The control-write and rollback changes are already present there. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ci): satisfy new clippy lints * style(scanner): order merged test imports * fix(scanner): invalidate bucket work after namespace completion * fix(scanner): fence cached snapshots by scan execution --------- Co-authored-by: houseme <housemecn@gmail.com> Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-07 20:46:11 +00:00 · 2026-09-05 21:47:12 +08:00
parent 447f3c704b
commit e2a921bc16
35 changed files with 2252 additions and 274 deletions
@@ -25,8 +25,8 @@
 //!   [`BACKFILL_SAVE_INTERVAL`], and at every page end, with an `If-Match`
 //!   compare-and-set so a concurrent cancel or takeover is never overwritten.
 //! - The `continuation_token` only advances once every pull queued from the
-//!   page before it has reported back, so a crash re-lists at most one page
-//!   (already-present keys are then skipped, never re-pulled).
+//!   page before it has succeeded. After a failure it stays at that page,
+//!   so crash recovery cannot skip failed pulls (existing keys are skipped).
 //! - The owner holds a lease of [`BACKFILL_LEASE`] renewed by every save. The
 //!   recovery loop ([`run_backfill_recovery_loop`]) scans the buckets this
 //!   node has an ODM state for every [`BACKFILL_RECOVERY_INTERVAL`] and takes
@@ -367,9 +367,8 @@ pub struct LocalBackfillObject {
    pub source_etag: Option<String>,
 }

-/// Receiver of one queued pull's report; `None` when the pull was coalesced
-/// into one already running.
-pub type PullReport = Option<oneshot::Receiver<QueuedPullOutcome>>;
+/// Shared report of a new or coalesced pull; absent only when not admitted.
+pub type PullReport = Option<super::pull::QueuedPullReport>;

 /// Everything the job needs from its bucket, so the loop can run against a
 /// mock in unit tests. Production: [`BucketBackfillContext`].
@@ -1191,9 +1190,11 @@ impl Job {
    }

    async fn main_loop(&mut self) -> Result<(), Stop> {
+        let mut cursor = self.checkpoint.continuation_token.clone();
+        let failed_at_resume = self.checkpoint.failed;
        loop {
            self.check_cancel()?;
-            let page = self.list_page().await?;
+            let page = self.list_page(cursor.as_deref()).await?;
            for object in &page.objects {
                self.check_cancel()?;
                self.checkpoint.listed += 1;
@@ -1205,10 +1206,13 @@ impl Job {
                self.drain_ready();
                self.tick(false).await?;
            }
-            // Only advance the cursor once every pull of this page reported
-            // back, so a takeover re-lists at most this page.
+            // A persisted cursor certifies successful work, not just listing
+            // progress. Keep it at the first failed page for crash recovery.
            self.drain_all().await?;
-            self.checkpoint.continuation_token = page.next_continuation_token.clone();
+            cursor = page.next_continuation_token;
+            if self.checkpoint.failed == failed_at_resume {
+                self.checkpoint.continuation_token = cursor.clone();
+            }
            self.tick(true).await?;
            if !page.is_truncated {
                return Ok(());
@@ -1223,7 +1227,7 @@ impl Job {
        }
    }

-    async fn list_page(&mut self) -> Result<SourcePage, Stop> {
+    async fn list_page(&mut self, cursor: Option<&str>) -> Result<SourcePage, Stop> {
        let mut attempt = 0;
        loop {
            while !self.context.source_available() {
@@ -1231,7 +1235,7 @@ impl Job {
                self.tick(false).await?;
            }
            let prefix = self.checkpoint.prefix.clone();
-            let token = self.checkpoint.continuation_token.clone();
+            let token = cursor.map(str::to_string);
            match self
                .context
                .list_page(prefix.as_deref(), token.as_deref(), BACKFILL_LIST_PAGE_SIZE)
@@ -1305,9 +1309,10 @@ impl Job {
        }
        loop {
            match self.context.enqueue(key) {
-                (EnqueueOutcome::Enqueued, report) => {
+                (EnqueueOutcome::Enqueued | EnqueueOutcome::Coalesced, report) => {
                    self.checkpoint.enqueued += 1;
-                    if let Some(rx) = report {
+                    let rx = report.ok_or(Stop::Unavailable)?;
+                    {
                        let key = key.to_string();
                        self.outstanding.push(Box::pin(async move { (key, rx.await) }));
                    }
@@ -1322,11 +1327,6 @@ impl Job {
                    );
                    return Ok(());
                }
-                (EnqueueOutcome::Coalesced, _) => {
-                    // Someone else pulls it; its result is not ours to count.
-                    self.checkpoint.enqueued += 1;
-                    return Ok(());
-                }
                (EnqueueOutcome::QueueFull, _) => {
                    // Wait, never drop: one completion frees a slot.
                    if self.outstanding.is_empty() {
@@ -1640,6 +1640,7 @@ mod tests {
        queue_capacity: usize,
        pending: Mutex<Vec<(String, oneshot::Sender<QueuedPullOutcome>)>>,
        fail_keys: HashSet<String>,
+        coalesced: bool,
        auto_complete: AtomicBool,
        cancel: CancellationToken,
        config_updated_at: Mutex<Option<OffsetDateTime>>,
@@ -1667,6 +1668,7 @@ mod tests {
                queue_capacity: usize::MAX,
                pending: Mutex::new(Vec::new()),
                fail_keys: HashSet::new(),
+                coalesced: false,
                auto_complete: AtomicBool::new(true),
                cancel: CancellationToken::new(),
                config_updated_at: Mutex::new(Some(ts(1_700_000_000))),
@@ -1746,7 +1748,12 @@ mod tests {
            } else {
                self.pending.lock().push((key.to_string(), tx));
            }
-            (EnqueueOutcome::Enqueued, Some(rx))
+            let outcome = if self.coalesced {
+                EnqueueOutcome::Coalesced
+            } else {
+                EnqueueOutcome::Enqueued
+            };
+            (outcome, Some(futures::FutureExt::shared(rx)))
        }

        fn cancel_token(&self) -> CancellationToken {
@@ -1912,7 +1919,7 @@ mod tests {
    #[tokio::test]
    async fn failed_pulls_are_counted_hashed_and_finish_with_failures() {
        let bucket = "backfill-failed";
-        let mut context = MockContext::new(5, 1000);
+        let mut context = MockContext::new(5, 2);
        Arc::get_mut(&mut context)
            .expect("unshared")
            .fail_keys
@@ -1927,12 +1934,52 @@ mod tests {
            .checkpoint;
        assert_eq!(cp.state, BackfillState::CompletedWithFailures);
        assert_eq!((cp.pulled, cp.failed), (4, 1));
+        assert_eq!(cp.continuation_token.as_deref(), Some("2"), "retain the first failed page for recovery");
        assert_eq!(cp.failed_keys, vec![key_hash("k/00002")]);
        let last = cp.last_error.expect("last error");
        assert_eq!(last.class, "local_write");
        assert_eq!(last.key_hash.as_deref(), Some(key_hash("k/00002").as_str()));
    }

+    #[tokio::test]
+    async fn coalesced_pulls_block_the_checkpoint_and_report_failures() {
+        let bucket = "backfill-coalesced";
+        let mut context = MockContext::new(1, 1);
+        {
+            let ctx = Arc::get_mut(&mut context).expect("unshared");
+            ctx.coalesced = true;
+            ctx.auto_complete = AtomicBool::new(false);
+            ctx.fail_keys.insert("k/00000".to_string());
+        }
+        let (_dirs, store, runner) = runner_with("node-a", bucket, Arc::clone(&context)).await;
+        runner.start(bucket, BackfillRequest::default()).await.expect("start");
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while context.pending.lock().is_empty() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("job enqueued");
+        assert!(runner.is_running_locally(bucket), "coalescing is not completion");
+        let cp = read_checkpoint(&store, bucket)
+            .await
+            .expect("read")
+            .expect("checkpoint")
+            .checkpoint;
+        assert!(cp.state.is_active());
+        assert!(cp.continuation_token.is_none());
+        context.complete_pending();
+        runner.wait_until_idle(bucket).await;
+        let cp = read_checkpoint(&store, bucket)
+            .await
+            .expect("read")
+            .expect("checkpoint")
+            .checkpoint;
+        assert_eq!(cp.state, BackfillState::CompletedWithFailures);
+        assert_eq!((cp.enqueued, cp.pulled, cp.failed), (1, 0, 1));
+        assert_eq!(cp.failed_keys, vec![key_hash("k/00000")]);
+    }
+
    #[tokio::test]
    async fn listing_failure_marks_the_job_failed_with_the_error_class() {
        let bucket = "backfill-list-error";
@@ -2145,6 +2192,68 @@ mod tests {
        assert_eq!(runner.recover_once().await.taken_over, 0, "a finished job is not recovered");
    }

+    #[tokio::test]
+    async fn recovery_advances_past_historical_failures_but_pins_new_failures() {
+        let bucket = "backfill-takeover-failed";
+        let mut context = MockContext::new(8, 2);
+        {
+            let ctx = Arc::get_mut(&mut context).expect("unshared");
+            ctx.auto_complete = AtomicBool::new(false);
+            ctx.fail_keys.insert("k/00004".to_string());
+        }
+        let (_dirs, store, runner) = runner_with("node-b", bucket, Arc::clone(&context)).await;
+        let crashed_at = OffsetDateTime::now_utc() - Duration::from_secs(300);
+        let mut crashed = BackfillCheckpoint::new(&BackfillRequest::default(), ts(1_700_000_000), "node-a", crashed_at);
+        crashed.continuation_token = Some("2".to_string());
+        crashed.failed = 1;
+        crashed.record_failure("local_write", Some("k/00002"), crashed_at);
+        write_checkpoint(&store, bucket, &crashed, None)
+            .await
+            .expect("seed failed page with an expired lease");
+
+        assert_eq!(runner.recover_once().await.taken_over, 1);
+        for (page_start, durable_token, failures) in [(2, "2", 1), (4, "4", 1), (6, "4", 2)] {
+            tokio::time::timeout(Duration::from_secs(10), async {
+                loop {
+                    if context.pending.lock().len() == 2 {
+                        break;
+                    }
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("resumed page enqueued before its reports complete");
+            assert_eq!(
+                context.pending.lock().iter().map(|(key, _)| key.clone()).collect::<Vec<_>>(),
+                vec![format!("k/{page_start:05}"), format!("k/{:05}", page_start + 1)]
+            );
+            let cp = read_checkpoint(&store, bucket)
+                .await
+                .expect("read persisted page boundary")
+                .expect("checkpoint")
+                .checkpoint;
+            assert_eq!(cp.job_id, crashed.job_id);
+            assert_eq!(cp.owner.as_ref().map(|owner| owner.node.as_str()), Some("node-b"));
+            assert_eq!(cp.continuation_token.as_deref(), Some(durable_token));
+            assert_eq!(cp.failed, failures);
+            context.complete_pending();
+        }
+        runner.wait_until_idle(bucket).await;
+        let cp = read_checkpoint(&store, bucket)
+            .await
+            .expect("read completed checkpoint")
+            .expect("checkpoint")
+            .checkpoint;
+        assert_eq!(cp.state, BackfillState::CompletedWithFailures);
+        assert_eq!((cp.pulled, cp.failed), (5, 2));
+        assert_eq!(cp.continuation_token.as_deref(), Some("4"));
+        assert_eq!(cp.failed_keys, vec![key_hash("k/00002"), key_hash("k/00004")]);
+        assert_eq!(
+            context.list_requests.lock().as_slice(),
+            &[Some("2".to_string()), Some("4".to_string()), Some("6".to_string())]
+        );
+    }
+
    #[tokio::test]
    async fn recovery_cancels_a_job_whose_config_changed_and_reclaims_own_node_jobs() {
        let bucket = "backfill-recovery-config";
@@ -37,6 +37,9 @@ pub const MAX_LIST_NO_PROGRESS_PAGES: u8 = 16;
 /// listing's own marker, so the decoder needs a positive signal before it
 /// treats an opaque token as a merged one.
 const LIST_THROUGH_TOKEN_TAG: &str = "odm-list";
+// Object keys cannot contain NUL (bucket::utils::is_valid_object_prefix),
+// so this framing cannot collide with a local key used as an opaque marker.
+const LIST_THROUGH_TOKEN_PREFIX: &str = "\0odm-list:";

 /// Pages fetched per side per request: the first page, plus at most one refill
 /// when the first one was mostly consumed by the previous page. Two pages of
@@ -91,8 +94,7 @@ pub struct MergePick {
 }

 /// The continuation-token envelope. Opaque to clients: it is serialized as
-/// JSON and then base64-encoded by the same helper that encodes a plain local
-/// marker, so the wire shape is `base64(json)`.
+/// framed JSON and then base64-encoded by the same helper as a local marker.
 ///
 /// A `null` cursor with `done = false` means "list that side from the start";
 /// `done = true` means the side is finished and must not be listed again.
@@ -139,7 +141,7 @@ impl ListThroughToken {
    pub fn encode(&self) -> String {
        // The envelope is built here from owned strings, so serialization
        // cannot fail; the fallback keeps the signature infallible.
-        serde_json::to_string(self).unwrap_or_default()
+        format!("{LIST_THROUGH_TOKEN_PREFIX}{}", serde_json::to_string(self).unwrap_or_default())
    }
 }

@@ -163,21 +165,18 @@ pub enum ListThroughTokenError {

 /// Classifies an already base64-decoded continuation token.
 ///
-/// Only a JSON object carrying the envelope marker is read as a merged token;
+/// Only a framed JSON object is read as a merged token;
 /// anything else is a local marker, so a bucket that turns `list_through` off
 /// keeps paginating with the tokens it handed out. A token that *is* an
 /// envelope but was tampered with (unknown version, unknown field, truncated
 /// JSON) is an error, never a silent fallback.
 pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, ListThroughTokenError> {
-    if !decoded.starts_with('{') {
-        return Ok(ListThroughCursor::Local(decoded.to_string()));
-    }
-    let Ok(value) = serde_json::from_str::<serde_json::Value>(decoded) else {
-        // Not JSON at all: an object key may legitimately start with '{'.
+    let Some(payload) = decoded.strip_prefix(LIST_THROUGH_TOKEN_PREFIX) else {
        return Ok(ListThroughCursor::Local(decoded.to_string()));
    };
+    let value = serde_json::from_str::<serde_json::Value>(payload).map_err(|_| ListThroughTokenError::Malformed)?;
    if value.get("t").and_then(serde_json::Value::as_str) != Some(LIST_THROUGH_TOKEN_TAG) {
-        return Ok(ListThroughCursor::Local(decoded.to_string()));
+        return Err(ListThroughTokenError::Malformed);
    }
    match value.get("v").and_then(serde_json::Value::as_u64) {
        Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
@@ -1052,9 +1051,9 @@ mod tests {
        assert_eq!(decode_continuation_token(&extra), Err(ListThroughTokenError::Malformed));

        let truncated = &encoded[..encoded.len() - 3];
-        assert_eq!(decode_continuation_token(truncated), Ok(ListThroughCursor::Local(truncated.to_string())));
+        assert_eq!(decode_continuation_token(truncated), Err(ListThroughTokenError::Malformed));

-        let no_version = "{\"t\":\"odm-list\"}";
+        let no_version = "\0odm-list:{\"t\":\"odm-list\"}";
        assert_eq!(decode_continuation_token(no_version), Err(ListThroughTokenError::Malformed));
    }

@@ -1311,6 +1310,13 @@ mod tests {

    #[test]
    fn a_plain_local_marker_stays_local() {
+        for marker in [
+            r#"{"t":"odm-list","v":1}"#,
+            r#"{"t":"odm-list","v":2,"local_done":true}"#,
+            r#"{"t":"odm-list"}"#,
+        ] {
+            assert_eq!(decode_continuation_token(marker), Ok(ListThroughCursor::Local(marker.to_string())));
+        }
        assert_eq!(
            decode_continuation_token("photos/2024/01.jpg"),
            Ok(ListThroughCursor::Local("photos/2024/01.jpg".to_string()))
@@ -46,10 +46,10 @@ use super::stats::{PullFailureReason, PullPath};
 use super::sys::{BucketOdmState, OnDemandMigrationSys, PullError, PullOutcome, PullSlot};
 use async_trait::async_trait;
 use bytes::Bytes;
-use futures::{Stream, StreamExt};
+use futures::{FutureExt, Stream, StreamExt, future::Shared};
 use parking_lot::Mutex;
 use rand::RngExt;
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::fmt;
 use std::io;
 use std::pin::Pin;
@@ -133,6 +133,8 @@ pub enum QueuedPullOutcome {
    Failed(PullError),
 }

+pub type QueuedPullReport = Shared<oneshot::Receiver<QueuedPullOutcome>>;
+
 /// Result of [`PullQueue::enqueue`].
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub enum EnqueueOutcome {
@@ -251,6 +253,7 @@ pub struct WriteBackRequest {
    pub preserve_etag: bool,
    /// `policy.emit_events`.
    pub emit_events: bool,
+    pub respect_delete_marker: bool,
    /// Source tags to copy (`policy.copy_tags`), `None` to skip.
    pub tags: Option<HashMap<String, String>>,
 }
@@ -266,6 +269,7 @@ impl WriteBackRequest {
            pulled_at: OffsetDateTime::now_utc(),
            preserve_etag: config.policy.preserve_etag,
            emit_events: config.policy.emit_events,
+            respect_delete_marker: config.policy.respect_local_delete_marker,
            tags,
        }
    }
@@ -830,7 +834,7 @@ pub struct PullQueue {
    bucket: String,
    tx: mpsc::Sender<PullJob>,
    /// Keys queued or running; the job removes its key when it ends.
-    pending: Mutex<HashSet<String>>,
+    pending: Mutex<HashMap<String, QueuedPullReport>>,
    capacity: usize,
    cancel: CancellationToken,
    stats: Arc<super::stats::OdmStats>,
@@ -869,7 +873,7 @@ impl PullQueue {
        let queue = Arc::new(Self {
            bucket: state.bucket().to_string(),
            tx,
-            pending: Mutex::new(HashSet::new()),
+            pending: Mutex::new(HashMap::new()),
            capacity,
            cancel: state.cancel_token(),
            stats: Arc::clone(state.stats()),
@@ -903,29 +907,24 @@ impl PullQueue {
        self.enqueue_with_report(key, reason).0
    }

-    /// [`Self::enqueue`] that also hands back the job's report channel when
-    /// a new job was queued (`Coalesced` pulls report to their first
-    /// requester only).
-    pub fn enqueue_with_report(
-        &self,
-        key: &str,
-        reason: PullReason,
-    ) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
+    /// [`Self::enqueue`] with a shared report, including for coalesced pulls.
+    pub fn enqueue_with_report(&self, key: &str, reason: PullReason) -> (EnqueueOutcome, Option<QueuedPullReport>) {
        if self.cancel.is_cancelled() {
            return (EnqueueOutcome::Unavailable, None);
        }
        let mut pending = self.pending.lock();
-        if pending.contains(key) {
-            return (EnqueueOutcome::Coalesced, None);
+        if let Some(report) = pending.get(key) {
+            return (EnqueueOutcome::Coalesced, Some(report.clone()));
        }
        let (report_tx, report_rx) = oneshot::channel();
+        let report_rx = report_rx.shared();
        match self.tx.try_send(PullJob {
            key: key.to_string(),
            reason,
            report: Some(report_tx),
        }) {
            Ok(()) => {
-                pending.insert(key.to_string());
+                pending.insert(key.to_string(), report_rx.clone());
                (EnqueueOutcome::Enqueued, Some(report_rx))
            }
            Err(TrySendError::Full(_)) => {
@@ -1072,7 +1071,7 @@ impl BucketOdmState {
        self: &Arc<Self>,
        key: &str,
        reason: PullReason,
-    ) -> (EnqueueOutcome, Option<oneshot::Receiver<QueuedPullOutcome>>) {
+    ) -> (EnqueueOutcome, Option<QueuedPullReport>) {
        match self.pull_queue() {
            Some(queue) => queue.enqueue_with_report(key, reason),
            None => (EnqueueOutcome::Unavailable, None),
@@ -1399,13 +1398,21 @@ mod tests {
        assert_eq!(queue.capacity(), 1024);

        let mut outcomes = HashMap::new();
+        let mut shared_report = None;
        for _ in 0..100 {
-            *outcomes.entry(queue.enqueue("a", PullReason::RangeGet)).or_insert(0) += 1;
+            let (outcome, report) = queue.enqueue_with_report("a", PullReason::RangeGet);
+            *outcomes.entry(outcome).or_insert(0) += 1;
+            shared_report = report;
        }
        assert_eq!(outcomes.get(&EnqueueOutcome::Enqueued), Some(&1));
        assert_eq!(outcomes.get(&EnqueueOutcome::Coalesced), Some(&99));
        assert_eq!(queue.pending_keys(), 1);

+        assert_eq!(
+            shared_report.expect("coalesced report").await,
+            Ok(QueuedPullOutcome::Stored { size: 1000 })
+        );
+
        wait_until("first pull to finish", || queue.pending_keys() == 0).await;
        assert_eq!(source.head_calls.load(Ordering::SeqCst), 1);
        assert_eq!(source.get_calls.load(Ordering::SeqCst), 1);
@@ -1438,6 +1445,23 @@ mod tests {
        assert_eq!(queue.enqueue("a", PullReason::RangeGet), EnqueueOutcome::Unavailable);
    }

+    #[tokio::test]
+    async fn coalesced_enqueues_share_failure_reports() {
+        let sys = OnDemandMigrationSys::new();
+        let state = enabled_state(&sys, &config()).await;
+        let source = MockSource::with_object("missing", 1000, BodyKind::Bytes(body_bytes(1000)));
+        let queue = PullQueue::start(Arc::clone(&state), source, Arc::new(MockWriteBack::default()));
+        let (first, first_report) = queue.enqueue_with_report("absent", PullReason::RangeGet);
+        let (second, second_report) = queue.enqueue_with_report("absent", PullReason::Backfill);
+        assert_eq!(first, EnqueueOutcome::Enqueued);
+        assert_eq!(second, EnqueueOutcome::Coalesced);
+        let (first, second) = tokio::join!(first_report.expect("leader report"), second_report.expect("coalesced report"));
+        assert_eq!(first, second);
+        assert!(matches!(first, Ok(QueuedPullOutcome::Failed(_))));
+        sys.remove(BUCKET);
+        queue.wait_until_stopped().await;
+    }
+
    #[tokio::test]
    async fn queue_full_is_reported_and_cancel_drains_without_leaking_tasks() {
        let sys = OnDemandMigrationSys::new();
@@ -1467,7 +1491,8 @@ mod tests {
        wait_until("dispatcher to wait for a slot", || state.stats().queue_depth() == 1).await;
        assert_eq!(queue.enqueue("c", PullReason::LargeObject), EnqueueOutcome::Enqueued);
        assert_eq!(queue.enqueue("d", PullReason::LargeObject), EnqueueOutcome::QueueFull);
-        assert_eq!(queue.enqueue("c", PullReason::LargeObject), EnqueueOutcome::Coalesced);
+        let (coalesced, canceled_report) = queue.enqueue_with_report("c", PullReason::LargeObject);
+        assert_eq!(coalesced, EnqueueOutcome::Coalesced);
        assert_eq!(queue.pending_keys(), 3);
        assert_eq!(failures(&state).get("queue_full"), Some(&1));
        assert!(!queue.is_stopped());
@@ -1477,6 +1502,12 @@ mod tests {
            .await
            .expect("dispatcher and in-flight job must exit after cancel");
        assert!(queue.is_stopped());
+        assert!(
+            tokio::time::timeout(Duration::from_secs(5), canceled_report.expect("coalesced cancellation report"))
+                .await
+                .expect("cancellation closes the report")
+                .is_err()
+        );
        assert_eq!(queue.pending_keys(), 0);
        assert_eq!(state.inflight_keys(), 0);
        assert_eq!(state.stats().inflight_pulls(), 0);
@@ -153,8 +153,8 @@ pub struct SourceClientSpec {
    /// Wire requests one logical source call may cost. The pull pipeline and
    /// the backfill job own the retry budget (`pull.rs` `PULL_MAX_RETRIES`,
    /// `backfill.rs` `LIST_MAX_RETRIES`) and the breaker counts logical calls,
-    /// so ODM declares [`RemoteS3RetryPolicy::Disabled`] and keeps one counted
-    /// failure equal to one request against a struggling source.
+    /// so ODM declares [`RemoteS3RetryPolicy::Disabled`]. An ambiguous HEAD
+    /// 404 additionally probes the bucket before declaring a key absent.
    pub retry: RemoteS3RetryPolicy,
    /// Bytes per second the pull pipeline may consume from this source;
    /// `None` means unlimited. Enforced by the consumer, not by this client.
@@ -262,7 +262,7 @@ const THROTTLE_CODES: &[&str] = &[
    "TooManyRequests",
    "RequestThrottled",
 ];
-const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "NotFound", "NoSuchBucket", "NoSuchVersion"];
+const NOT_FOUND_CODES: &[&str] = &["NoSuchKey"];
 const ACCESS_DENIED_CODES: &[&str] = &[
    "AccessDenied",
    "InvalidAccessKeyId",
@@ -285,7 +285,6 @@ fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceEr
        }
    }
    match status {
-        404 => SourceError::NotFound,
        401 | 403 => SourceError::AccessDenied,
        429 | 503 => SourceError::Throttled,
        500..=599 => SourceError::ServerError(status),
@@ -631,8 +630,7 @@ impl SourceClient {
    }

    /// `config` must come from [`SourceClientSpec::endpoint_spec`], which is
-    /// where the retry policy that keeps one logical call equal to one wire
-    /// request is declared.
+    /// where the policy disabling SDK-level retries is declared.
    fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
        let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
        Self {
@@ -754,15 +752,16 @@ impl SourceClient {
 #[async_trait::async_trait]
 impl SourceBackend for S3SourceBackend {
    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
-        let output = self
-            .client
-            .head_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_head_from_head_output(output)
+        match self.client.head_object().bucket(&self.bucket).key(key).send().await {
+            Ok(output) => source_head_from_head_output(output),
+            Err(err) if err.raw_response().is_some_and(|response| response.status().as_u16() == 404) => {
+                // HEAD has no error body: a missing bucket must not poison
+                // the per-key negative cache as though only the key was absent.
+                self.probe().await?;
+                Err(SourceError::NotFound)
+            }
+            Err(err) => Err(classify_sdk_error(err)),
+        }
    }

    /// Streams the object; `range` is passed through as an HTTP `Range`
@@ -809,8 +808,8 @@ impl SourceBackend for S3SourceBackend {
            .contents
            .unwrap_or_default()
            .into_iter()
-            .filter_map(s3_source_object)
-            .collect();
+            .map(s3_source_object)
+            .collect::<Result<Vec<_>, _>>()?;
        let common_prefixes = output
            .common_prefixes
            .unwrap_or_default()
@@ -849,14 +848,20 @@ impl SourceBackend for S3SourceBackend {
    }
 }

-fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
-    let key = object.key?;
+fn s3_source_object(object: SdkObject) -> Result<SourceObject, SourceError> {
+    let key = object
+        .key
+        .ok_or_else(|| SourceError::Other("source listing object has no key".to_string()))?;
+    let size = object
+        .size
+        .and_then(|size| u64::try_from(size).ok())
+        .ok_or_else(|| SourceError::Other("source listing object has no valid size".to_string()))?;
    let etag = normalize_etag(object.e_tag);
    let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
-    Some(SourceObject {
+    Ok(SourceObject {
        key,
        etag,
-        size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
+        size,
        last_modified: system_time(object.last_modified),
        storage_class: object.storage_class.map(|class| class.as_str().to_string()),
        is_multipart_etag,
@@ -1489,7 +1494,10 @@ mod tests {
    #[tokio::test]
    async fn source_error_classification_covers_every_class() {
        let cases: Vec<(Scripted, &str, bool)> = vec![
-            (status(404, ""), "not_found", false),
+            (status(404, ""), "other", false),
+            (status(404, "<Error><Code>NoSuchKey</Code></Error>"), "not_found", false),
+            (status(404, "<Error><Code>NoSuchBucket</Code></Error>"), "other", false),
+            (status(404, "<Error><Code>NoSuchVersion</Code></Error>"), "other", false),
            (status(403, ACCESS_DENIED_BODY), "access_denied", false),
            (status(401, ""), "access_denied", false),
            (status(429, ""), "throttled", true),
@@ -1512,14 +1520,35 @@ mod tests {
            }
        }

-        // HEAD carries no error body, so the classification must work from the
-        // status alone as well.
-        let (client, _) = scripted_client(&spec(None), vec![status(404, "")]).await;
+        let (client, requests) = scripted_client(&spec(None), vec![status(404, ""), status(200, "")]).await;
        assert!(matches!(client.head_object("missing").await, Err(SourceError::NotFound)));
+        assert_eq!(recorded(&requests).len(), 2, "ambiguous HEAD 404 must check the bucket");
+        let (client, _) = scripted_client(&spec(None), vec![status(404, ""), status(404, "")]).await;
+        assert!(matches!(client.head_object("missing").await, Err(SourceError::Other(_))));
+        let (client, _) = scripted_client(&spec(None), vec![status(404, ""), status(403, "")]).await;
+        assert!(matches!(client.head_object("missing").await, Err(SourceError::AccessDenied)));
        let (client, _) = scripted_client(&spec(None), vec![status(403, "")]).await;
        assert!(matches!(client.head_object("secret").await, Err(SourceError::AccessDenied)));
    }

+    #[test]
+    fn source_listing_rejects_missing_and_negative_sizes() {
+        for size in [None, Some(-1)] {
+            let object = SdkObject::builder().key("key").set_size(size).build();
+            assert!(matches!(s3_source_object(object), Err(SourceError::Other(_))));
+        }
+        assert!(matches!(
+            s3_source_object(SdkObject::builder().size(0).build()),
+            Err(SourceError::Other(_))
+        ));
+        assert_eq!(
+            s3_source_object(SdkObject::builder().key("empty").size(0).build())
+                .expect("empty object")
+                .size,
+            0
+        );
+    }
+
    #[tokio::test]
    async fn source_client_debug_redacts_credentials() {
        let (client, _) = scripted_client(&spec(Some("data/")), Vec::new()).await;