fix(ecstore): reject stalled ODM pagination before merging pages (#7164)

* fix(odm): reject non-progressing listing cursors * docs(odm): clarify folded source probe pagination * test(odm): match SDK bucket-root listing requests * test(ecstore): match sealed context fixture map type * test(odm): use app facade for listing wire types * fix(odm): resolve pagination Clippy failures
2026-09-07 12:35:54 +00:00 · 2026-09-05 15:17:25 +08:00
parent a6589c19e3
commit 971f9acdf4
6 changed files with 763 additions and 37 deletions
@@ -86,7 +86,12 @@ impl BreakerVerdict {
            Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => {
                BreakerVerdict::Failure
            }
-            Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral,
+            Some(
+                SourceError::AccessDenied
+                | SourceError::Unsupported(_)
+                | SourceError::InvalidPagination(_)
+                | SourceError::Other(_),
+            ) => BreakerVerdict::Neutral,
        }
    }
 }
@@ -189,8 +189,8 @@ pub enum SourceListPlan {
    /// delimiter — the source's own roll-up boundary matches the request's.
    Page { prefix: String },
    /// `filter.prefix` reaches past a delimiter, so every key the source could
-    /// contribute rolls into this one common prefix. One bounded probe listing
-    /// decides whether it exists; there is nothing to paginate.
+    /// contribute rolls into this one common prefix. Bounded probes follow
+    /// empty progressing pages until a key proves existence or the source ends.
    Folded { probe_prefix: String, common_prefix: String },
 }

@@ -279,6 +279,29 @@ pub struct FetchRequest {
    pub token: Option<String>,
 }

+/// Invalid pagination metadata. Opaque cursor values are never included in errors.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
+pub enum ListPageError {
+    #[error("truncated listing has no continuation token")]
+    Missing,
+    #[error("truncated listing has an empty continuation token")]
+    Empty,
+    #[error("truncated listing repeats a continuation token")]
+    Repeated,
+}
+
+pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> {
+    if is_truncated {
+        match next_token {
+            None => return Err(ListPageError::Missing),
+            Some("") => return Err(ListPageError::Empty),
+            Some(next) if Some(next) == token => return Err(ListPageError::Repeated),
+            Some(_) => {}
+        }
+    }
+    Ok(())
+}
+
 #[derive(Debug, Default)]
 struct SideState {
    start: SideCursor,
@@ -364,6 +387,11 @@ impl ListThroughMerger {
    /// or `filter.prefix` excludes it.
    pub fn disable_source(&mut self) {
        self.source.disabled = true;
+        // A refill can fail after a valid first page. A local-only response
+        // must discard both that source payload and its ordering horizon.
+        self.source.entries.clear();
+        self.source.pages.clear();
+        self.source.more = false;
    }

    pub fn next_fetch(&self) -> Option<FetchRequest> {
@@ -378,7 +406,13 @@ impl ListThroughMerger {
    /// Records one fetched page. `entries` must be sorted by `name` and already
    /// filtered with [`Self::accepts`]; the caller keeps the matching payloads
    /// in the same order.
-    pub fn push_page(&mut self, side: MergeSide, entries: Vec<ListEntryKey>, is_truncated: bool, next_token: Option<String>) {
+    pub fn push_page(
+        &mut self,
+        side: MergeSide,
+        entries: Vec<ListEntryKey>,
+        is_truncated: bool,
+        next_token: Option<String>,
+    ) -> Result<(), ListPageError> {
        let state = match side {
            MergeSide::Local => &mut self.local,
            MergeSide::Source => &mut self.source,
@@ -387,15 +421,19 @@ impl ListThroughMerger {
            Some(last) => last.next_token.clone(),
            None => state.start.token.clone(),
        };
-        // A truncated page without a cursor cannot be continued; treating the
-        // side as finished is the only alternative to looping on it forever.
-        state.more = is_truncated && next_token.is_some();
+        validate_list_page(is_truncated, token.as_deref(), next_token.as_deref())?;
+        // Also reject a cycle through an earlier page in this bounded fetch.
+        if is_truncated && state.pages.iter().any(|page| page.token == next_token) {
+            return Err(ListPageError::Repeated);
+        }
+        state.more = is_truncated;
        state.pages.push(FetchedPage {
            token,
            count: entries.len(),
            next_token: is_truncated.then_some(next_token).flatten(),
        });
        state.entries.extend(entries);
+        Ok(())
    }

    pub fn finish(self) -> MergeOutcome {
@@ -599,9 +637,15 @@ mod tests {
                let (entries, truncated, next) = reference_page(keys, prefix, delimiter, fetch.token.as_deref(), max_keys);
                let kept: Vec<ListEntryKey> = entries.into_iter().filter(|entry| merger.accepts(&entry.name)).collect();
                buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.iter().cloned());
-                merger.push_page(fetch.side, kept, truncated, next);
+                merger
+                    .push_page(fetch.side, kept, truncated, next)
+                    .expect("reference provider pages must advance");
            }
            let outcome = merger.finish();
+            assert_eq!(outcome.is_truncated, outcome.next_token.is_some());
+            if outcome.is_truncated {
+                assert_ne!(outcome.next_token, token, "every truncated merged page must make progress");
+            }
            page_sizes.push(outcome.picks.len());
            for pick in &outcome.picks {
                let entry = buffers[usize::from(pick.side == MergeSide::Source)][pick.index].clone();
@@ -616,11 +660,25 @@ mod tests {
    }

    fn expected(local: &[String], source: &[String], prefix: &str, delimiter: Option<&str>) -> Vec<ListEntryKey> {
-        let mut all: Vec<String> = local.iter().chain(source.iter()).cloned().collect();
-        all.sort();
-        all.dedup();
-        let (entries, _, _) = reference_page(&all, prefix, delimiter, None, usize::MAX);
-        entries
+        // This oracle builds the complete namespace independently of the
+        // provider's page/marker helper and the production merger.
+        let mut namespace = std::collections::BTreeMap::new();
+        for key in local.iter().chain(source) {
+            let Some(suffix) = key.strip_prefix(prefix) else {
+                continue;
+            };
+            if let Some(delimiter) = delimiter.filter(|delimiter| !delimiter.is_empty())
+                && let Some((directory, _)) = suffix.split_once(delimiter)
+            {
+                namespace.insert(format!("{prefix}{directory}{delimiter}"), true);
+                continue;
+            }
+            namespace.insert(key.clone(), false);
+        }
+        namespace
+            .into_iter()
+            .map(|(name, is_prefix)| ListEntryKey { name, is_prefix })
+            .collect()
    }

    #[test]
@@ -662,7 +720,9 @@ mod tests {
                token: None
            })
        );
-        merger.push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None)
+            .expect("local EOF is valid");
        assert_eq!(merger.next_fetch(), None);
        let outcome = merger.finish();
        assert_eq!(outcome.picks.len(), 1);
@@ -683,12 +743,14 @@ mod tests {
        };
        let mut merger = ListThroughMerger::new(1, Some(&resume));
        merger.disable_source();
-        merger.push_page(
-            MergeSide::Local,
-            vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
-            true,
-            Some("local-2".to_string()),
-        );
+        merger
+            .push_page(
+                MergeSide::Local,
+                vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
+                true,
+                Some("local-2".to_string()),
+            )
+            .expect("local cursor advances");
        let outcome = merger.finish();
        assert!(outcome.is_truncated);
        let token = outcome.next_token.expect("truncated page carries a token");
@@ -698,6 +760,212 @@ mod tests {
        assert_eq!(token.local.as_deref(), Some("local-1"), "a partly read page is re-listed");
    }

+    #[test]
+    fn truncated_pages_require_a_nonempty_advancing_cursor() {
+        for side in [MergeSide::Local, MergeSide::Source] {
+            for entries in [vec![], vec![ListEntryKey::object("a")]] {
+                for (next, expected) in [
+                    (None, Err(ListPageError::Missing)),
+                    (Some(""), Err(ListPageError::Empty)),
+                    (Some("stuck"), Err(ListPageError::Repeated)),
+                    (Some("advances"), Ok(())),
+                ] {
+                    let resume = ListThroughToken::new(
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        None,
+                    );
+                    let mut merger = ListThroughMerger::new(2, Some(&resume));
+                    let result = merger.push_page(side, entries.clone(), true, next.map(str::to_string));
+                    assert_eq!(result, expected, "{side:?}, {entries:?}, {next:?}");
+                    let state = if side == MergeSide::Local {
+                        &merger.local
+                    } else {
+                        &merger.source
+                    };
+                    assert_eq!(state.pages.len(), usize::from(result.is_ok()), "invalid page must not be accepted");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn repeated_empty_cursor_is_rejected_before_an_identical_page_can_escape() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("stuck".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        assert_eq!(
+            merger.next_fetch(),
+            Some(FetchRequest {
+                side: MergeSide::Source,
+                token: Some("stuck".into())
+            })
+        );
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn empty_pages_may_advance_within_the_fetch_budget_until_eof() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger.push_page(MergeSide::Local, vec![], false, None).expect("local EOF");
+        for next in ["opaque-z", "opaque-a"] {
+            assert_eq!(merger.next_fetch().expect("bounded source fetch").side, MergeSide::Source);
+            merger
+                .push_page(MergeSide::Source, vec![], true, Some(next.into()))
+                .expect("opaque cursor advances regardless of sort order");
+        }
+        assert!(merger.next_fetch().is_none(), "two source fetches exhaust the request budget");
+        let outcome = merger.finish();
+        assert!(outcome.picks.is_empty());
+        assert!(outcome.is_truncated);
+        let token = outcome.next_token.expect("empty progressing page has a cursor");
+        assert_eq!(token.source.as_deref(), Some("opaque-a"));
+        let mut merger = ListThroughMerger::new(2, Some(&token));
+        assert_eq!(merger.next_fetch().expect("source resumes").token.as_deref(), Some("opaque-a"));
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
+            .expect("source EOF");
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Source,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn a_cursor_cycle_inside_the_fetch_budget_is_rejected() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("first".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger
+            .push_page(MergeSide::Source, vec![], true, Some("second".into()))
+            .expect("first page advances");
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("first".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn source_refill_failure_discards_buffered_source_entries_and_horizon() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("z")], false, None)
+            .expect("local EOF");
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("a")], true, Some("stuck".into()))
+            .expect("first source page advances");
+        assert_eq!(merger.next_fetch().expect("source refill is required").token.as_deref(), Some("stuck"));
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+        merger.disable_source();
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Local,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn list_through_static_namespace_boundary_matrix() {
+        let corpus = [
+            "a",
+            "a/",
+            "a/b",
+            "a/b/child",
+            "a0",
+            "b",
+            "b/leaf",
+            "quote\"&<",
+            "space key",
+            "z",
+            "é",
+            "中/文",
+        ];
+        for count in [0, 1, 3, 4, corpus.len()] {
+            let keys: Vec<String> = corpus[..count].iter().map(|key| (*key).to_string()).collect();
+            for placement in 0..3 {
+                let (local, source): (Vec<_>, Vec<_>) =
+                    keys.iter()
+                        .enumerate()
+                        .fold((vec![], vec![]), |(mut local, mut source), (index, key)| {
+                            if placement != 1 || index % 2 == 0 {
+                                local.push(key.clone());
+                            }
+                            if placement != 0 || index % 2 == 0 {
+                                source.push(key.clone());
+                            }
+                            (local, source)
+                        });
+                for prefix in ["", "a", "a/", "中/"] {
+                    for delimiter in [None, Some("/")] {
+                        for max_keys in [1, 3, 4] {
+                            let oracle = expected(&local, &source, prefix, delimiter);
+                            let (emitted, sizes) = walk(&local, &source, prefix, delimiter, max_keys);
+                            assert_eq!(
+                                emitted.iter().map(|(entry, _)| entry.clone()).collect::<Vec<_>>(),
+                                oracle,
+                                "count={count}, placement={placement}, prefix={prefix}, delimiter={delimiter:?}, max={max_keys}"
+                            );
+                            let expected_sizes: Vec<_> = if oracle.is_empty() {
+                                vec![0]
+                            } else {
+                                oracle.chunks(max_keys).map(<[ListEntryKey]>::len).collect()
+                            };
+                            assert_eq!(sizes, expected_sizes, "exact max and max+1 boundaries must agree");
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn list_through_large_overlap_walk_keeps_all_5300_keys() {
+        let source: Vec<_> = (0..5000).map(|index| format!("k{index:05}")).collect();
+        let local: Vec<_> = (4800..5300).map(|index| format!("k{index:05}")).collect();
+        let (emitted, sizes) = walk(&local, &source, "", None, 333);
+        assert_eq!(emitted.len(), 5300);
+        for (index, (entry, side)) in emitted.iter().enumerate() {
+            assert_eq!(entry.name, format!("k{index:05}"));
+            assert_eq!(*side, if index >= 4800 { MergeSide::Local } else { MergeSide::Source });
+        }
+        assert_eq!(sizes, [vec![333; 15], vec![305]].concat());
+    }
+
    #[test]
    fn token_round_trips_and_rejects_tampering() {
        let token = ListThroughToken::new(
@@ -796,7 +1064,10 @@ mod tests {
    }

    proptest! {
-        #![proptest_config(ProptestConfig::with_cases(256))]
+        #![proptest_config(ProptestConfig {
+            rng_seed: proptest::test_runner::RngSeed::Fixed(0xec5706),
+            ..ProptestConfig::with_cases(256)
+        })]

        /// Full pagination of a merged listing equals the sorted, deduplicated
        /// union of both sides, with every shared key served by local, and no
@@ -25,6 +25,7 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.

+use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
    PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
 };
@@ -223,6 +224,8 @@ pub enum SourceError {
    ServerError(u16),
    #[error("unsupported source object: {0}")]
    Unsupported(String),
+    #[error("invalid source listing: {0}")]
+    InvalidPagination(#[from] ListPageError),
    #[error("source request failed: {0}")]
    Other(String),
 }
@@ -245,6 +248,7 @@ impl SourceError {
            SourceError::Connect(_) => "connect",
            SourceError::ServerError(_) => "server_error",
            SourceError::Unsupported(_) => "unsupported",
+            SourceError::InvalidPagination(_) => "invalid_pagination",
            SourceError::Other(_) => "other",
        }
    }
@@ -714,6 +718,7 @@ impl SourceClient {
                ..*request
            })
            .await?;
+        validate_list_page(page.is_truncated, request.continuation_token, page.next_continuation_token.as_deref())?;
        page.objects = page
            .objects
            .into_iter()
@@ -800,11 +805,6 @@ impl SourceBackend for S3SourceBackend {

        let is_truncated = output.is_truncated.unwrap_or(false);
        let next_continuation_token = output.next_continuation_token;
-        if is_truncated && next_continuation_token.is_none() {
-            return Err(SourceError::Other(
-                "source reported a truncated listing without a continuation token".to_string(),
-            ));
-        }
        let objects = output
            .contents
            .unwrap_or_default()
@@ -1274,7 +1274,9 @@ mod tests {
 <CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
 <CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
 </ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
+        let next_body = body.replace("data/opaque", "data/next");
+        let (client, requests) =
+            scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), &next_body)]).await;
        let first = client
            .list_page(&SourceListRequest {
                prefix: Some("photos/"),
@@ -1336,7 +1338,104 @@ mod tests {
            .list_objects_v2(None, None, 10)
            .await
            .expect_err("truncated page without token is corrupt");
-        assert!(matches!(err, SourceError::Other(_)), "{err:?}");
+        assert!(matches!(err, SourceError::InvalidPagination(ListPageError::Missing)), "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_s3_cursor_progress_before_mapping_entries() {
+        for contents in ["", "<Contents><Key>data/a</Key><Size>1</Size></Contents>"] {
+            for (truncated, next, expected) in [
+                (true, None, Some(ListPageError::Missing)),
+                (true, Some(""), Some(ListPageError::Empty)),
+                (true, Some("stuck"), Some(ListPageError::Repeated)),
+                (true, Some("opaque-next"), None),
+                (false, None, None),
+                (false, Some("stuck"), None),
+            ] {
+                let next_xml = next
+                    .map(|next| format!("<NextContinuationToken>{next}</NextContinuationToken>"))
+                    .unwrap_or_default();
+                let body = format!(
+                    "<ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><IsTruncated>{truncated}</IsTruncated>{next_xml}{contents}</ListBucketResult>"
+                );
+                let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), &body)]).await;
+                let result = client
+                    .list_page(&SourceListRequest {
+                        continuation_token: Some("stuck"),
+                        max_keys: 2,
+                        ..Default::default()
+                    })
+                    .await;
+                match expected {
+                    Some(expected) => {
+                        let error = result.expect_err("malformed pagination must fail at the provider boundary");
+                        assert!(
+                            matches!(&error, SourceError::InvalidPagination(actual) if *actual == expected),
+                            "{error:?}"
+                        );
+                        assert_eq!(error.class_label(), "invalid_pagination");
+                        assert!(!error.is_retryable());
+                        assert!(!error.to_string().contains("stuck"), "errors must not echo opaque tokens");
+                    }
+                    None => {
+                        let page = result.expect("progressing empty/nonempty pages and EOF are valid");
+                        assert_eq!(page.is_truncated, truncated);
+                        assert_eq!(page.next_continuation_token.as_deref(), next);
+                        assert_eq!(page.objects.len(), usize::from(!contents.is_empty()));
+                        if let Some(object) = page.objects.first() {
+                            assert_eq!(object.key, "a");
+                        }
+                    }
+                }
+                let requests = recorded(&requests);
+                assert_eq!(requests.len(), 1, "invalid pagination must not be retried");
+                assert!(requests[0].uri.contains("continuation-token=stuck"));
+            }
+        }
+    }
+
+    struct ListOnlyBackend(SourcePage);
+
+    #[async_trait::async_trait]
+    impl SourceBackend for ListOnlyBackend {
+        async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+            assert_eq!(request.continuation_token, Some("stuck"), "opaque cursors reach every provider unchanged");
+            Ok(self.0.clone())
+        }
+
+        async fn head(&self, _key: &str) -> Result<SourceHead, SourceError> {
+            panic!("unexpected HEAD in list test")
+        }
+        async fn get(&self, _key: &str, _range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+            panic!("unexpected GET in list test")
+        }
+        async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
+            panic!("unexpected tagging in list test")
+        }
+        async fn probe(&self) -> Result<(), SourceError> {
+            panic!("unexpected probe in list test")
+        }
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_non_s3_provider_cursors_at_the_common_boundary() {
+        for (next, expected) in [
+            (None, ListPageError::Missing),
+            (Some(""), ListPageError::Empty),
+            (Some("stuck"), ListPageError::Repeated),
+        ] {
+            let mut client = prefix_client(Some("data/".into()));
+            client.backend = Box::new(ListOnlyBackend(SourcePage {
+                is_truncated: true,
+                next_continuation_token: next.map(str::to_string),
+                ..Default::default()
+            }));
+            let error = client
+                .list_objects_v2(None, Some("stuck"), 2)
+                .await
+                .expect_err("all providers must advance pagination");
+            assert!(matches!(error, SourceError::InvalidPagination(actual) if actual == expected));
+        }
    }

    const TAGGING_BODY: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -177,7 +177,7 @@ impl From<&SourceError> for PullFailureReason {
            SourceError::Connect(_) => PullFailureReason::SourceConnect,
            SourceError::ServerError(_) => PullFailureReason::SourceServerError,
            SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported,
-            SourceError::Other(_) => PullFailureReason::SourceOther,
+            SourceError::InvalidPagination(_) | SourceError::Other(_) => PullFailureReason::SourceOther,
        }
    }
 }