fix(odm): bound empty pagination chains with staged tokens (#7197)

* fix(odm): add staged cross-request pagination progress budgets

* 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(app): simplify absent SSE configuration matching

* fix(tests): satisfy new clippy lints

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

.github/workflows/rustfs-functional-chain.yml

@@ -14,8 +14,8 @@
 
 # Functional chain driver: runs the ten functional suites in a fixed order
 # (upgrade -> s3 -> kms -> tier -> storage -> heal -> pool -> security ->
-# replication -> performance). Each suite attempts the next handoff even
-# when its tests fail.
+# replication, with performance on its own runner in parallel) and guarantees
+# the chain keeps moving even when individual suites fail.
 #
 # Each suite workflow can still be dispatched standalone (workflow_dispatch);
 # only chain-triggered runs forward to the next suite via repository_dispatch,
@@ -59,3 +59,16 @@ jobs:
           gh api --method POST repos/rustfs/rustfs/dispatches \
             -f event_type='rustfs-chain-upgrade' \
             -F 'client_payload[from_suite]=nightly-build'
+
+      - name: Dispatch performance suite (parallel, own runner)
+        env:
+          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch performance" >&2
+            exit 1
+          fi
+          gh api --method POST repos/rustfs/rustfs/dispatches \
+            -f event_type='rustfs-chain-performance' \
+            -F 'client_payload[from_suite]=nightly-build'

.github/workflows/rustfs-performance-test.yml

@@ -49,16 +49,17 @@ on:
         type: boolean
         default: true
   repository_dispatch:
-    # Chain handoff: dispatched when the replication suite finishes.
+    # Chain entry: dispatched by rustfs-functional-chain.yml (runs on its own
+    # pf-testing runner, in parallel with the shared-VM chain).
     types: [rustfs-chain-performance]
 
 permissions:
   contents: read
 
-# The default performance nodes overlap the other suites' remote VMs, even
-# though the runner differs. Hold the shared lock through cleanup as well.
+# Dedicated pf-testing runner/environment: own concurrency group so perf runs
+# never block (or are blocked by) the pool-expansion / heal tests.
 concurrency:
-  group: rustfs-shared-functional-tests
+  group: rustfs-performance-test
   cancel-in-progress: false
 
 defaults:

.github/workflows/rustfs-replication-test.yml

@@ -34,7 +34,8 @@ on:
           - site
         default: all
   repository_dispatch:
-    # Chain handoff: dispatched when the security suite finishes.
+    # Chain handoff: dispatched when the security suite finishes. This is the
+    # last link of the functional chain.
     types: [rustfs-chain-replication]
 
 permissions:
@@ -61,6 +62,9 @@ env:
 jobs:
   replication-test:
     runs-on: smoke-testing
+    # A failed replication run must not break the chain or the workflow: the
+    # failure is reported to rustfs/backlog instead (see the issue step).
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
@@ -345,50 +349,13 @@ jobs:
             '
           done
 
-      - name: "Continue functional chain (next: Performance)"
+      - name: Chain complete
+        # Replication is the last link of the functional chain: nothing to
+        # dispatch after it. This step just records that the chain finished.
         if: ${{ always() && github.event_name == 'repository_dispatch' }}
-        env:
-          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
-          set -uo pipefail
-          if [ -z "${GH_TOKEN:-}" ]; then
-            echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
-            exit 1
-          fi
-          DISPATCHED=0
-          for attempt in 1 2 3; do
-            if gh api --method POST repos/rustfs/rustfs/dispatches \
-              -f event_type='rustfs-chain-performance' \
-              -F 'client_payload[from_suite]=replication'; then
-              echo "dispatched next suite Performance (attempt ${attempt})"
-              DISPATCHED=1
-              break
-            fi
-            echo "dispatch attempt ${attempt} failed; retrying in ${attempt}0s" >&2
-            sleep "${attempt}0"
-          done
-          if [ "${DISPATCHED:-0}" -ne 1 ]; then
-            echo "ERROR: functional chain stalled: could not dispatch Performance after 3 attempts" >&2
-            TITLE="[functional][chain] stalled after replication (run ${GITHUB_RUN_ID})"
-            BODY_FILE="$(mktemp)"
-            trap 'rm -f "${BODY_FILE}"' EXIT
-            {
-              echo "The functional chain could not hand off from **replication** to **Performance** after 3 attempts."
-              echo ""
-              echo "- Failed suite job: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
-              echo "- Expected next event: 'rustfs-chain-performance'"
-              echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
-              echo "- Recovery: re-dispatch manually with"
-              FENCE="$(printf "\x60\x60\x60")"; echo "  ${FENCE}"
-              echo "  gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-performance'"
-              FENCE="$(printf "\x60\x60\x60")"; echo "  ${FENCE}"
-            } > "${BODY_FILE}"
-            gh issue create -R rustfs/backlog --title "${TITLE}" \
-              --body-file "${BODY_FILE}" --label functional-test \
-              || gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}" \
-              || echo "could not file the stall alert issue either; check the token" >&2
-            exit 1
-          fi
+          echo "Functional chain complete: replication (final suite) finished."
+          echo "from_suite=security trigger=${{ github.event_name }} outcome=${{ steps.test.outcome }}"
 
       - name: Notify on failure
         if: failure()

crates/ecstore/src/api/mod.rs

@@ -167,9 +167,10 @@ pub mod bucket {
             idle_guarded_body,
         };
         pub use crate::bucket::on_demand_migration::{
-            FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
-            ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MergeOutcome, MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT,
-            SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter, decode_continuation_token, source_list_plan,
+            FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger,
+            ListThroughToken, ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MAX_LIST_NO_PROGRESS_PAGES, MergeOutcome,
+            MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter,
+            decode_continuation_token, source_list_plan,
         };
         pub mod backfill {
             pub use crate::bucket::on_demand_migration::backfill::{

crates/ecstore/src/bucket/on_demand_migration/list_through.rs

@@ -25,8 +25,13 @@ use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use std::time::{Duration, Instant};
 
-/// The only continuation-token envelope version this build reads and writes.
+/// The continuation-token version used by ordinary progressing pages.
 pub const LIST_THROUGH_TOKEN_VERSION: u32 = 1;
+const LIST_THROUGH_PROGRESS_TOKEN_VERSION: u32 = 2;
+
+/// The sixteenth consecutive merged page without a key or new EOF fails.
+/// This also bounds legitimate sparse listings; it is not a cycle detector.
+pub const MAX_LIST_NO_PROGRESS_PAGES: u8 = 16;
 
 /// Envelope marker. A bucket that is *not* merging hands out the local
 /// listing's own marker, so the decoder needs a positive signal before it
@@ -111,6 +116,10 @@ pub struct ListThroughToken {
     /// common prefix compares as itself, never as its members.
     #[serde(default)]
     pub last_key: Option<String>,
+    /// Consecutive empty truncated merged pages, present only in v2 tokens.
+    /// Ordinary v1 tokens retain their original serialized shape.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub no_progress: Option<u8>,
 }
 
 impl ListThroughToken {
@@ -123,6 +132,7 @@ impl ListThroughToken {
             source: source.token,
             source_done: source.done,
             last_key,
+            no_progress: None,
         }
     }
 
@@ -170,7 +180,21 @@ pub fn decode_continuation_token(decoded: &str) -> Result<ListThroughCursor, Lis
         return Ok(ListThroughCursor::Local(decoded.to_string()));
     }
     match value.get("v").and_then(serde_json::Value::as_u64) {
-        Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {}
+        Some(version) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {
+            // v1 readers reject this field even when it is null or zero.
+            if value.get("no_progress").is_some() {
+                return Err(ListThroughTokenError::Malformed);
+            }
+        }
+        Some(version) if version == u64::from(LIST_THROUGH_PROGRESS_TOKEN_VERSION) => {
+            if !value
+                .get("no_progress")
+                .and_then(serde_json::Value::as_u64)
+                .is_some_and(|count| (1..u64::from(MAX_LIST_NO_PROGRESS_PAGES)).contains(&count))
+            {
+                return Err(ListThroughTokenError::Malformed);
+            }
+        }
         Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
         None => return Err(ListThroughTokenError::Malformed),
     }
@@ -288,6 +312,8 @@ pub enum ListPageError {
     Empty,
     #[error("truncated listing repeats a continuation token")]
     Repeated,
+    #[error("listing exhausted its consecutive no-progress page budget")]
+    NoProgress(MergeSide),
 }
 
 pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> {
@@ -352,6 +378,7 @@ pub struct MergeOutcome {
 #[derive(Debug)]
 pub struct ListThroughMerger {
     max_keys: usize,
+    no_progress: Option<u8>,
     last_key: Option<String>,
     local: SideState,
     source: SideState,
@@ -371,6 +398,7 @@ impl ListThroughMerger {
         };
         Self {
             max_keys,
+            no_progress: token.and_then(|token| token.no_progress),
             last_key,
             local,
             source,
@@ -436,13 +464,18 @@ impl ListThroughMerger {
         Ok(())
     }
 
-    pub fn finish(self) -> MergeOutcome {
+    /// `issue_progress_tokens` allows a v1 chain to start carrying a budget.
+    /// An existing v2 budget is always enforced, including on reader-only nodes.
+    /// Borrowing lets a source failure re-merge the fetched local buffers.
+    pub fn finish(&self, issue_progress_tokens: bool) -> Result<MergeOutcome, ListPageError> {
         let Self {
             max_keys,
+            no_progress,
             last_key,
             local,
             source,
         } = self;
+        let max_keys = *max_keys;
 
         // A side with more pages behind it can only be trusted up to the last
         // key it handed over: past that horizon the other side's entries could
@@ -508,12 +541,44 @@ impl ListThroughMerger {
         let source_left = !source.disabled && (!source_cursor.done || consumed_source < source.entries.len());
         let is_truncated = local_left || source_left;
 
-        let last_key = consumed_key.or(last_key);
-        MergeOutcome {
+        let reached_eof = (!local.start.done && local_cursor.done) || (!source.start.done && source_cursor.done);
+        let next_no_progress = if !is_truncated || !picks.is_empty() || reached_eof {
+            None
+        } else if max_keys == 0 {
+            // A zero-sized request cannot consume entries. Preserve an existing
+            // budget without spending it or starting a new one.
+            *no_progress
+        } else if issue_progress_tokens || no_progress.is_some() {
+            let count = no_progress.unwrap_or(0).saturating_add(1);
+            if count >= MAX_LIST_NO_PROGRESS_PAGES {
+                // An empty truncated side closes the merge horizon. Local
+                // failure takes precedence; disabling the source cannot fix it.
+                let side = if local.more && local.entries.is_empty() {
+                    MergeSide::Local
+                } else if !source.disabled && source.more && source.entries.is_empty() {
+                    MergeSide::Source
+                } else {
+                    MergeSide::Local
+                };
+                return Err(ListPageError::NoProgress(side));
+            }
+            Some(count)
+        } else {
+            None
+        };
+        let last_key = consumed_key.or_else(|| last_key.clone());
+        Ok(MergeOutcome {
             picks,
             is_truncated,
-            next_token: is_truncated.then(|| ListThroughToken::new(local_cursor, source_cursor, last_key)),
-        }
+            next_token: is_truncated.then(|| {
+                let mut token = ListThroughToken::new(local_cursor, source_cursor, last_key);
+                if let Some(count) = next_no_progress {
+                    token.v = LIST_THROUGH_PROGRESS_TOKEN_VERSION;
+                    token.no_progress = Some(count);
+                }
+                token
+            }),
+        })
     }
 }
 
@@ -641,7 +706,7 @@ mod tests {
                     .push_page(fetch.side, kept, truncated, next)
                     .expect("reference provider pages must advance");
             }
-            let outcome = merger.finish();
+            let outcome = merger.finish(false).expect("valid merge outcome");
             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");
@@ -724,7 +789,7 @@ mod tests {
             .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();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(outcome.picks.len(), 1);
         assert!(!outcome.is_truncated);
         assert!(outcome.next_token.is_none());
@@ -740,6 +805,7 @@ mod tests {
             source: Some("source-1".to_string()),
             source_done: false,
             last_key: Some("a".to_string()),
+            no_progress: None,
         };
         let mut merger = ListThroughMerger::new(1, Some(&resume));
         merger.disable_source();
@@ -751,7 +817,7 @@ mod tests {
                 Some("local-2".to_string()),
             )
             .expect("local cursor advances");
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("truncated page carries a token");
         assert_eq!(token.source.as_deref(), Some("source-1"), "the source cursor must not move");
@@ -830,7 +896,7 @@ mod tests {
                 .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();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert!(outcome.picks.is_empty());
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("empty progressing page has a cursor");
@@ -840,7 +906,7 @@ mod tests {
         merger
             .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
             .expect("source EOF");
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -887,7 +953,7 @@ mod tests {
             Err(ListPageError::Repeated)
         );
         merger.disable_source();
-        let outcome = merger.finish();
+        let outcome = merger.finish(false).expect("valid merge outcome");
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -979,8 +1045,8 @@ mod tests {
         let encoded = token.encode();
         assert_eq!(decode_continuation_token(&encoded), Ok(ListThroughCursor::Merged(Box::new(token))));
 
-        let bumped = encoded.replace("\"v\":1", "\"v\":2");
-        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(2)));
+        let bumped = encoded.replace("\"v\":1", "\"v\":3");
+        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(3)));
 
         let extra = encoded.replace("{", "{\"x\":1,");
         assert_eq!(decode_continuation_token(&extra), Err(ListThroughTokenError::Malformed));
@@ -992,6 +1058,257 @@ mod tests {
         assert_eq!(decode_continuation_token(no_version), Err(ListThroughTokenError::Malformed));
     }
 
+    fn progress_token(count: Option<u8>, local_done: bool, source_done: bool) -> ListThroughToken {
+        let mut token = ListThroughToken::new(
+            SideCursor {
+                token: None,
+                done: local_done,
+            },
+            SideCursor {
+                token: Some("A".into()),
+                done: source_done,
+            },
+            Some("last-key".into()),
+        );
+        if let Some(count) = count {
+            token.v = LIST_THROUGH_PROGRESS_TOKEN_VERSION;
+            token.no_progress = Some(count);
+        }
+        token
+    }
+
+    fn push_empty_pages(merger: &mut ListThroughMerger, side: MergeSide) {
+        for _ in 0..MAX_LIST_FETCHES_PER_SIDE {
+            let fetch = merger.next_fetch().expect("empty truncated side must be fetched");
+            assert_eq!(fetch.side, side);
+            let next = format!("{}:next", fetch.token.unwrap_or_default());
+            merger
+                .push_page(side, vec![], true, Some(next))
+                .expect("opaque cursor advances");
+        }
+    }
+
+    #[test]
+    fn progress_tokens_preserve_v1_bytes_and_validate_v2_counts() {
+        let token = progress_token(None, true, false);
+        assert_eq!(
+            token.encode(),
+            r#"{"t":"odm-list","v":1,"local":null,"local_done":true,"source":"A","source_done":false,"last_key":"last-key"}"#
+        );
+        for count in 1..MAX_LIST_NO_PROGRESS_PAGES {
+            let token = progress_token(Some(count), true, false);
+            assert_eq!(decode_continuation_token(&token.encode()), Ok(ListThroughCursor::Merged(Box::new(token))));
+        }
+        for version in [1, 2] {
+            for value in ["null", "0", "16", "-1", "1.5", "256", "18446744073709551616", "\"1\""] {
+                let encoded = format!(r#"{{"t":"odm-list","v":{version},"no_progress":{value}}}"#);
+                assert_eq!(decode_continuation_token(&encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+            }
+        }
+        for encoded in [
+            r#"{"t":"odm-list","v":1,"no_progress":1}"#,
+            r#"{"t":"odm-list","v":2}"#,
+            r#"{"t":"odm-list","v":2,"no_progress":1,"extra":true}"#,
+        ] {
+            assert_eq!(decode_continuation_token(encoded), Err(ListThroughTokenError::Malformed), "{encoded}");
+        }
+    }
+
+    #[test]
+    fn reader_only_nodes_do_not_start_a_budget_but_mixed_readers_preserve_one() {
+        let mut token = progress_token(None, true, false);
+        for _ in 0..MAX_LIST_NO_PROGRESS_PAGES {
+            let mut merger = ListThroughMerger::new(2, Some(&token));
+            push_empty_pages(&mut merger, MergeSide::Source);
+            token = merger
+                .finish(false)
+                .expect("reader-only v1 behavior")
+                .next_token
+                .expect("truncated cursor");
+            assert_eq!(token.v, 1);
+            assert_eq!(token.no_progress, None);
+        }
+        for count in 1..=MAX_LIST_NO_PROGRESS_PAGES {
+            let mut merger = ListThroughMerger::new(2, Some(&token));
+            push_empty_pages(&mut merger, MergeSide::Source);
+            assert!(merger.next_fetch().is_none(), "the per-request two-fetch limit stays intact");
+            let outcome = merger.finish(count % 2 == 1);
+            if count == MAX_LIST_NO_PROGRESS_PAGES {
+                assert_eq!(outcome, Err(ListPageError::NoProgress(MergeSide::Source)));
+                break;
+            }
+            token = outcome.expect("budget not exhausted").next_token.expect("truncated cursor");
+            assert_eq!(token.no_progress, Some(count));
+            let ListThroughCursor::Merged(decoded) = decode_continuation_token(&token.encode()).expect("round-trip v2") else {
+                panic!("merged cursor expected");
+            };
+            token = *decoded;
+        }
+    }
+
+    #[test]
+    fn objects_and_common_prefixes_reset_a_budget_at_the_boundary() {
+        for entry in [ListEntryKey::object("result"), ListEntryKey::prefix("result/")] {
+            for issue_tokens in [false, true] {
+                let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+                let mut merger = ListThroughMerger::new(2, Some(&resume));
+                merger
+                    .push_page(MergeSide::Source, vec![], true, Some("B".into()))
+                    .expect("empty advancing page");
+                merger
+                    .push_page(MergeSide::Source, vec![entry.clone()], true, Some("C".into()))
+                    .expect("real progress");
+                let outcome = merger
+                    .finish(issue_tokens)
+                    .expect("real progress does not exhaust the budget");
+                assert_eq!(
+                    outcome.picks,
+                    vec![MergePick {
+                        side: MergeSide::Source,
+                        index: 0
+                    }]
+                );
+                let next = outcome.next_token.expect("source remains truncated");
+                assert_eq!(next.last_key.as_deref(), Some(entry.name.as_str()));
+                assert_eq!(next.v, 1);
+                assert_eq!(next.no_progress, None);
+                assert!(!next.encode().contains("no_progress"));
+            }
+        }
+    }
+
+    #[test]
+    fn only_a_new_eof_transition_resets_the_empty_page_budget() {
+        for finished_side in [MergeSide::Local, MergeSide::Source] {
+            let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, false);
+            let mut merger = ListThroughMerger::new(2, Some(&resume));
+            if finished_side == MergeSide::Local {
+                merger
+                    .push_page(MergeSide::Local, vec![], false, None)
+                    .expect("new local EOF");
+                push_empty_pages(&mut merger, MergeSide::Source);
+            } else {
+                push_empty_pages(&mut merger, MergeSide::Local);
+                merger
+                    .push_page(MergeSide::Source, vec![], false, None)
+                    .expect("new source EOF");
+            }
+            let next = merger
+                .finish(false)
+                .expect("new EOF is progress")
+                .next_token
+                .expect("other side truncated");
+            assert_eq!(next.no_progress, None);
+            assert_eq!(next.v, 1);
+            assert_eq!(next.local_done, finished_side == MergeSide::Local);
+            assert_eq!(next.source_done, finished_side == MergeSide::Source);
+            let mut merger = ListThroughMerger::new(2, Some(&next));
+            let remaining = if finished_side == MergeSide::Local {
+                MergeSide::Source
+            } else {
+                MergeSide::Local
+            };
+            push_empty_pages(&mut merger, remaining);
+            let next = merger
+                .finish(true)
+                .expect("a new budget starts")
+                .next_token
+                .expect("truncated");
+            assert_eq!(next.no_progress, Some(1), "an already-done side cannot reset every page");
+        }
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger.push_page(MergeSide::Source, vec![], false, None).expect("final EOF");
+        let outcome = merger.finish(false).expect("EOF succeeds at the budget boundary");
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn filtered_duplicates_cannot_reset_the_no_progress_budget() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        for next in ["B", "C"] {
+            let entries = [ListEntryKey::object("last-key"), ListEntryKey::object("earlier")]
+                .into_iter()
+                .filter(|entry| merger.accepts(&entry.name))
+                .collect::<Vec<_>>();
+            assert!(entries.is_empty(), "both provider entries were already consumed");
+            merger
+                .push_page(MergeSide::Source, entries, true, Some(next.into()))
+                .expect("advancing cursor");
+        }
+        assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Source)));
+    }
+
+    #[test]
+    fn no_progress_is_attributed_to_local_when_source_cannot_unblock_it() {
+        for source_mode in ["disabled", "done", "empty", "data"] {
+            let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, source_mode == "done");
+            let mut merger = ListThroughMerger::new(2, Some(&resume));
+            if source_mode == "disabled" {
+                merger.disable_source();
+            }
+            push_empty_pages(&mut merger, MergeSide::Local);
+            match source_mode {
+                "empty" => push_empty_pages(&mut merger, MergeSide::Source),
+                "data" => merger
+                    .push_page(MergeSide::Source, vec![ListEntryKey::object("source")], false, None)
+                    .expect("source data"),
+                _ => {}
+            }
+            assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Local)), "{source_mode}");
+        }
+    }
+
+    #[test]
+    fn source_budget_failure_remerges_local_objects_and_prefixes_without_refetching() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), false, false);
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("local")], true, Some("L1".into()))
+            .expect("local object");
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::prefix("prefix/")], true, Some("L2".into()))
+            .expect("local prefix");
+        push_empty_pages(&mut merger, MergeSide::Source);
+        assert_eq!(merger.finish(false), Err(ListPageError::NoProgress(MergeSide::Source)));
+        merger.disable_source();
+        assert!(merger.next_fetch().is_none(), "fallback does not perform another fetch");
+        let outcome = merger.finish(false).expect("local data makes progress");
+        assert_eq!(
+            outcome.picks,
+            vec![
+                MergePick {
+                    side: MergeSide::Local,
+                    index: 0
+                },
+                MergePick {
+                    side: MergeSide::Local,
+                    index: 1
+                }
+            ]
+        );
+        let token = outcome.next_token.expect("remaining local page");
+        assert_eq!(token.local.as_deref(), Some("L2"));
+        assert_eq!(token.source.as_deref(), Some("A"));
+        assert_eq!(token.last_key.as_deref(), Some("prefix/"));
+        assert_eq!(token.no_progress, None);
+        assert_eq!(token.v, 1);
+    }
+
+    #[test]
+    fn a_zero_sized_merge_preserves_an_existing_budget() {
+        let resume = progress_token(Some(MAX_LIST_NO_PROGRESS_PAGES - 1), true, false);
+        let mut merger = ListThroughMerger::new(0, Some(&resume));
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], true, Some("B".into()))
+            .expect("source page");
+        let outcome = merger.finish(false).expect("a zero-sized request cannot consume entries");
+        assert!(outcome.picks.is_empty());
+        assert_eq!(outcome.next_token.expect("unconsumed source").no_progress, resume.no_progress);
+    }
+
     #[test]
     fn a_plain_local_marker_stays_local() {
         assert_eq!(

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -40,9 +40,10 @@ pub use config::{
     SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
 };
 pub use list_through::{
-    FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
-    ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MergeOutcome, MergePick, MergeSide, SOURCE_LIST_MAX_RATE_WAIT,
-    SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter, decode_continuation_token, source_list_plan,
+    FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger,
+    ListThroughToken, ListThroughTokenError, MAX_LIST_FETCHES_PER_SIDE, MAX_LIST_NO_PROGRESS_PAGES, MergeOutcome, MergePick,
+    MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SOURCE_LIST_RATE_PER_SEC, SourceListPlan, SourceListRateLimiter,
+    decode_continuation_token, source_list_plan,
 };
 pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
 pub use pull::{

docs/operations/on-demand-migration.md

@@ -7,6 +7,16 @@ On-Demand Migration (ODM) attaches an external S3-compatible **source bucket** t
 
 The module is on by default (rustfs/backlog#2163); set `RUSTFS_ON_DEMAND_MIGRATION_ENABLED=false` on every node to turn it off (`rustfs/src/module_switches.rs`). With the switch off, the runtime never intervenes on a read and the admin `PUT` route refuses with `OnDemandMigrationDisabled`. Reads of the configuration and of the status endpoint keep working while the switch is off, so a disabled deployment can still be inspected. The switch only decides whether the module may act at all: a bucket with no `on-demand-migration.json` is never resolved by the runtime and makes no source call, so turning the module on changes nothing for buckets you have not configured.
 
+## List continuation token rollout
+
+`RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS` defaults to `false`; unset or invalid boolean values also keep it off. It controls only whether a v1 listing may first issue a v2 continuation token after an empty truncated merged page. Every node with this reader support accepts existing v2 tokens and continues their budget even with the switch off. Ordinary pages that consume an object or common prefix retain the original v1 token shape.
+
+Leave the switch off while deploying v2 reader support to every node that can receive a continuation request, including nodes behind other load-balancer routes. Then set it to `true` in each node's environment and restart those nodes to enable issuance. A v1-only binary rejects v2 with `400 InvalidArgument` before the source-error policy runs; neither `not_found` nor turning off list-through makes that old reader compatible. With issuance still off, a new v1 chain retains the existing limitation: an empty source cursor cycle spanning requests can continue indefinitely. The default rollout does not claim to fix that chain until issuance is enabled.
+
+An active v2 budget rejects the sixteenth consecutive merged page that consumes no new object/common prefix and reaches no new end-of-list state. The first fifteen empty pages can be resumed; with the existing two-fetch-per-side limit, that interval costs at most 32 fetches per side, including the failing request. A key, common prefix, or a newly exhausted side on the sixteenth request succeeds and resets the budget. A side that was already exhausted does not reset it again. This is a resource bound, not proof of a cursor cycle: an unusually long but valid empty source-page chain also reaches the limit. Tokens are unsigned base64 JSON, so this budget applies to clients that continue with the returned token unchanged; replaying or editing a token can reset it, and it is not a malicious-client defense or a global request quota. The two-fetch-per-side request limit and existing source rate limiter still apply. A source failure follows `policy.source_error`: `propagate` returns `424 SourceUnavailable` with `invalid_pagination`; `not_found` returns the fetched local listing with `x-rustfs-on-demand-migration-list: local_only`. A blocking local-side failure returns `InternalError`, without silently discarding local entries.
+
+For rollback, first turn issuance off on every node. Keep v2-capable readers available for outstanding v2 chains: switching issuance off does not erase their budgets, and tokens have no expiration that proves those chains have drained. Route those continuations to compatible readers or have clients explicitly restart their listings before restoring v1-only binaries. Restarting a listing is a new scan and can repeat entries. Do not roll back readers while assuming the issuance switch makes existing v2 tokens disappear.
+
 ## Positioning
 
 | Capability | Direction | What it moves | Where the authoritative copy is | When to use it instead |

rustfs/src/app/bucket_list_through.rs

@@ -26,8 +26,8 @@ use super::storage_api::bucket_usecase::ECStore;
 use super::storage_api::bucket_usecase::StorageObjectInfo as ObjectInfo;
 use super::storage_api::bucket_usecase::StorageObjectOptions;
 use super::storage_api::bucket_usecase::bucket::on_demand_migration::{
-    BucketOdmState, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-    OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
+    BucketOdmState, ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+    MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
     SourceListRequest, SourceObject, SourcePage, decode_continuation_token, source_list_plan,
 };
 use super::storage_api::bucket_usecase::bucket::versioning_sys::BucketVersioningSys;
@@ -51,6 +51,9 @@ type ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
 /// yet, so the only class RustFS can vouch for is the default one.
 const SOURCE_STORAGE_CLASS: &str = "STANDARD";
 
+/// Enable only after every node serving continuation requests can read v2.
+const ENV_LIST_PROGRESS_TOKENS: &str = "RUSTFS_ON_DEMAND_MIGRATION_LIST_V2_TOKENS";
+
 /// Concurrent local metadata probes when a versioned bucket has to check
 /// source-only keys for a shadowing delete marker.
 const DELETE_MARKER_PROBE_CONCURRENCY: usize = 32;
@@ -264,7 +267,18 @@ pub(crate) async fn merged_list_objects_v2(
         buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some));
     }
 
-    let outcome = merger.finish();
+    let issue_progress_tokens = rustfs_utils::get_env_bool(ENV_LIST_PROGRESS_TOKENS, false);
+    let outcome = match merger.finish(issue_progress_tokens) {
+        Ok(outcome) => outcome,
+        Err(ListPageError::NoProgress(MergeSide::Source)) => {
+            degrade_or_fail(&mut merger, &mut degraded, policy.source_error, "invalid_pagination")?;
+            merger
+                .finish(issue_progress_tokens)
+                .map_err(|error| S3Error::with_message(S3ErrorCode::InternalError, error.to_string()))?
+        }
+        Err(error) => return Err(S3Error::with_message(S3ErrorCode::InternalError, error.to_string())),
+    };
+    drop(merger);
     let mut objects = Vec::with_capacity(outcome.picks.len());
     let mut prefixes = Vec::new();
     let mut source_only_keys = Vec::new();
@@ -430,7 +444,8 @@ mod tests {
     use crate::app::bucket_usecase::DefaultBucketUsecase;
     use crate::app::gating_test_env::{run_large_stack_test, shared_gating_ecstore};
     use crate::app::storage_api::bucket_usecase::bucket::on_demand_migration::{
-        FilterConfig, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig, SourceCredentials, TlsConfig,
+        FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
+        SourceCredentials, TlsConfig,
     };
     use crate::app::storage_api::bucket_usecase::s3::{ListObjectsV2Input, ListObjectsV2Output, S3Request, S3Response};
     use crate::app::storage_api::test::StoragePutObjReader;
@@ -448,6 +463,7 @@ mod tests {
             source: Some("source-2".to_string()),
             source_done: false,
             last_key: Some("k".to_string()),
+            no_progress: None,
         }
     }
 
@@ -515,6 +531,24 @@ mod tests {
         assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
     }
 
+    #[test]
+    fn a_v2_token_keeps_the_local_cursor_when_list_through_is_turned_off() {
+        let mut resume = token(Some("local-2"), false);
+        resume.v = 2;
+        resume.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 1);
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("a v2 envelope decodes");
+        assert_eq!(decoded.as_ref(), Some(&resume));
+        assert!(matches!(
+            local_cursor(Some(&encoded), decoded.as_ref()),
+            LocalListCursor::Token(Some(local)) if local == "local-2"
+        ));
+        resume.local_done = true;
+        let encoded = resume.encode();
+        let decoded = decode_list_cursor(Some(&encoded)).expect("v2 with local EOF decodes");
+        assert!(matches!(local_cursor(Some(&encoded), decoded.as_ref()), LocalListCursor::Exhausted));
+    }
+
     #[test]
     fn a_plain_local_token_is_passed_through_and_a_tampered_one_is_rejected() {
         assert!(
@@ -551,14 +585,30 @@ mod tests {
     /// Serves exactly the scripted S3 pages and joins every connection before
     /// returning. A source retry or unexpected operation fails the test.
     async fn scripted_list_source(pages: Vec<String>) -> (String, tokio_util::task::AbortOnDropHandle<Vec<String>>) {
+        let (endpoint, server, _) = list_source(pages.into_iter()).await;
+        (endpoint, server)
+    }
+
+    async fn list_source(
+        pages: impl Iterator<Item = String> + Send + 'static,
+    ) -> (
+        String,
+        tokio_util::task::AbortOnDropHandle<Vec<String>>,
+        tokio_util::sync::CancellationToken,
+    ) {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
             .await
             .expect("bind listing source");
         let address = listener.local_addr().expect("listing source address");
+        let stop = tokio_util::sync::CancellationToken::new();
+        let server_stop = stop.clone();
         let server = tokio::spawn(async move {
             let mut requests = Vec::new();
             for body in pages {
-                let (mut stream, _) = listener.accept().await.expect("accept source listing");
+                let (mut stream, _) = tokio::select! {
+                    _ = server_stop.cancelled() => break,
+                    accepted = listener.accept() => accepted.expect("accept source listing"),
+                };
                 let mut request = Vec::new();
                 let mut chunk = [0; 4096];
                 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
@@ -588,7 +638,7 @@ mod tests {
             }
             requests
         });
-        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server))
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server), stop)
     }
 
     fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String {
@@ -616,12 +666,12 @@ mod tests {
         }
     }
 
-    async fn source_policy_request(
-        pages: Vec<String>,
+    async fn source_policy_input(
+        endpoint: String,
         policy: SourceErrorPolicy,
         resume_source: Option<&str>,
         filter_prefix: Option<&str>,
-    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+    ) -> (ListThroughTestState, ListObjectsV2Input) {
         let store = shared_gating_ecstore().await;
         crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
         let bucket = format!("odm-list-{}", uuid::Uuid::new_v4().simple());
@@ -638,9 +688,8 @@ mod tests {
             )
             .await
             .expect("seed real local listing");
-        let (endpoint, server) = scripted_list_source(pages).await;
         let sys = OnDemandMigrationSys::get();
-        let _state_guard = ListThroughTestState {
+        let state_guard = ListThroughTestState {
             bucket: bucket.clone(),
             module_enabled: sys.is_module_enabled(),
         };
@@ -685,6 +734,7 @@ mod tests {
                 source: Some(source.into()),
                 source_done: false,
                 last_key: None,
+                no_progress: None,
             };
             base64_simd::STANDARD.encode_to_string(token.encode().as_bytes())
         });
@@ -701,6 +751,10 @@ mod tests {
             request_payer: None,
             start_after: None,
         };
+        (state_guard, input)
+    }
+
+    async fn execute_source_list(input: ListObjectsV2Input) -> S3Result<S3Response<ListObjectsV2Output>> {
         let request = S3Request {
             input,
             method: http::Method::GET,
@@ -712,12 +766,23 @@ mod tests {
             service: None,
             trailing_headers: None,
         };
-        let result = tokio::time::timeout(
+        tokio::time::timeout(
             Duration::from_secs(10),
             DefaultBucketUsecase::from_global().execute_list_objects_v2(request),
         )
         .await
-        .expect("listing must complete within its bounded source budget");
+        .expect("listing must complete within its bounded source budget")
+    }
+
+    async fn source_policy_request(
+        pages: Vec<String>,
+        policy: SourceErrorPolicy,
+        resume_source: Option<&str>,
+        filter_prefix: Option<&str>,
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
+        let (endpoint, server) = scripted_list_source(pages).await;
+        let (_state_guard, input) = source_policy_input(endpoint, policy, resume_source, filter_prefix).await;
+        let result = execute_source_list(input).await;
         let requests = tokio::time::timeout(Duration::from_secs(5), server)
             .await
             .expect("source connections must finish")
@@ -842,6 +907,244 @@ mod tests {
         });
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn list_through_cross_request_empty_cursor_cycle_obeys_policy() {
+        run_large_stack_test("list-through-cross-request-cursor-cycle", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let mut seen = std::collections::HashSet::from([input
+                            .continuation_token
+                            .clone()
+                            .expect("the first request resumes source cursor A")]);
+                        let mut client_requests = 0;
+                        let mut empty_pages = 0;
+                        let terminal = tokio::time::timeout(Duration::from_secs(30), async {
+                            loop {
+                                client_requests += 1;
+                                let response = match execute_source_list(input.clone()).await {
+                                    Ok(response) => response,
+                                    Err(error) => break Err(error),
+                                };
+                                if response.headers.contains_key("x-rustfs-on-demand-migration-list") {
+                                    break Ok(response);
+                                }
+                                let output = response.output;
+                                assert!(output.contents.as_ref().is_none_or(Vec::is_empty));
+                                assert!(output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                                assert_eq!(output.key_count, Some(0));
+                                assert_eq!(output.is_truncated, Some(true));
+                                let next = output
+                                    .next_continuation_token
+                                    .expect("a truncated page must carry its cursor");
+                                assert!(
+                                    seen.insert(next.clone()),
+                                    "a cross-request source cursor cycle must not return an identical empty merged token"
+                                );
+                                empty_pages += 1;
+                                input.continuation_token = Some(next);
+                            }
+                        })
+                        .await
+                        .expect("a source cursor cycle must terminate within a bounded client pagination chain");
+                        assert_eq!(empty_pages, usize::from(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        assert_eq!(client_requests, usize::from(MAX_LIST_NO_PROGRESS_PAGES));
+                        assert_source_policy_result(terminal, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("cyclic source server must not panic");
+                        assert_eq!(requests.len(), 2 * client_requests, "the sixteenth empty page exhausts the budget");
+                        for (index, request) in requests.iter().enumerate() {
+                            let source_cursor = ["A", "B", "C"][index % 3];
+                            assert!(
+                                request.contains(&format!("continuation-token={source_cursor}")),
+                                "the real SDK must follow the returned source cursor: {request}"
+                            );
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_default_rollout_continues_v2_without_issuing_it_from_v1() {
+        run_large_stack_test("list-through-reader-first-rollout", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, None),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for policy in [SourceErrorPolicy::Propagate, SourceErrorPolicy::NotFound] {
+                        let pages = ["B", "C", "A"].map(|next| source_xml(Some(next), true, None));
+                        let (endpoint, server, stop) = list_source(pages.into_iter().cycle()).await;
+                        let (_state_guard, mut input) = source_policy_input(endpoint, policy, Some("A"), None).await;
+                        let original = input.continuation_token.clone();
+                        for _ in 0..3 {
+                            let response = execute_source_list(input.clone()).await.expect("reader-only v1 behavior");
+                            assert!(!response.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                            assert_eq!(response.output.key_count, Some(0));
+                            assert_eq!(response.output.is_truncated, Some(true));
+                            let next = response.output.next_continuation_token.expect("resumable empty page");
+                            let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                            let decoded = std::str::from_utf8(&raw).expect("JSON token");
+                            let token = decode_list_cursor(Some(decoded)).expect("v1 reader").expect("merged token");
+                            assert_eq!(token.v, 1, "the default rollout cannot begin issuing v2");
+                            assert_eq!(token.no_progress, None);
+                            assert!(!decoded.contains("no_progress"), "ordinary v1 wire shape stays unchanged");
+                            input.continuation_token = Some(next);
+                        }
+                        assert_eq!(input.continuation_token, original, "default rollout retains the known v1 limitation");
+
+                        let raw = base64_simd::STANDARD
+                            .decode_to_vec(input.continuation_token.as_ref().expect("v1 token"))
+                            .expect("base64 continuation token");
+                        let mut token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v1 reader")
+                            .expect("merged token");
+                        token.v = 2;
+                        token.no_progress = Some(MAX_LIST_NO_PROGRESS_PAGES - 2);
+                        input.continuation_token = Some(base64_simd::STANDARD.encode_to_string(token.encode().as_bytes()));
+                        let response = execute_source_list(input.clone()).await.expect("reader-only node resumes v2");
+                        assert_eq!(response.output.key_count, Some(0));
+                        assert_eq!(response.output.is_truncated, Some(true));
+                        let next = response.output.next_continuation_token.expect("last allowed empty cursor");
+                        let raw = base64_simd::STANDARD.decode_to_vec(&next).expect("base64 continuation token");
+                        let token = decode_list_cursor(Some(std::str::from_utf8(&raw).expect("JSON token")))
+                            .expect("v2 reader")
+                            .expect("merged token");
+                        assert_eq!(token.v, 2);
+                        assert_eq!(token.no_progress, Some(MAX_LIST_NO_PROGRESS_PAGES - 1));
+                        input.continuation_token = Some(next);
+                        assert_source_policy_result(execute_source_list(input).await, policy);
+                        stop.cancel();
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("cyclic source server must stop")
+                            .expect("source server must not panic");
+                        assert_eq!(requests.len(), 10, "five handler requests each fetched two source pages");
+                        for (index, request) in requests.iter().enumerate() {
+                            let cursor = ["A", "B", "C"][index % 3];
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn list_through_empty_advancing_pages_resume_across_handler_requests() {
+        run_large_stack_test("list-through-resumable-empty-pages", || async {
+            temp_env::async_with_vars(
+                [
+                    (ENV_LIST_PROGRESS_TOKENS, Some("true")),
+                    ("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", Some("true")),
+                    ("HTTP_PROXY", None),
+                    ("HTTPS_PROXY", None),
+                    ("ALL_PROXY", None),
+                    ("http_proxy", None),
+                    ("https_proxy", None),
+                    ("all_proxy", None),
+                    ("NO_PROXY", Some("*")),
+                    ("no_proxy", Some("*")),
+                ],
+                async {
+                    for filter_prefix in [None, Some("photos/2024/")] {
+                        let source_key = filter_prefix.map_or("a-source", |_| "photos/2024/a-source");
+                        let (endpoint, server) = scripted_list_source(vec![
+                            source_xml(Some("A"), true, None),
+                            source_xml(Some("B"), true, None),
+                            source_xml(Some("C"), true, None),
+                            source_xml(None, false, Some(source_key)),
+                        ])
+                        .await;
+                        let (_state_guard, mut input) =
+                            source_policy_input(endpoint, SourceErrorPolicy::Propagate, None, filter_prefix).await;
+                        let first = execute_source_list(input.clone())
+                            .await
+                            .expect("valid empty pages must remain resumable");
+                        assert!(!first.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        assert!(first.output.contents.as_ref().is_none_or(Vec::is_empty));
+                        assert!(first.output.common_prefixes.as_ref().is_none_or(Vec::is_empty));
+                        assert_eq!(first.output.key_count, Some(0));
+                        assert_eq!(first.output.is_truncated, Some(true));
+                        input.continuation_token = Some(first.output.next_continuation_token.expect("empty advancing cursor"));
+
+                        let second = execute_source_list(input)
+                            .await
+                            .expect("a progressing empty chain must reach its data");
+                        assert!(!second.headers.contains_key("x-rustfs-on-demand-migration-list"));
+                        let output = second.output;
+                        let objects = output
+                            .contents
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|object| object.key.expect("listed object key"))
+                            .collect::<Vec<_>>();
+                        let prefixes = output
+                            .common_prefixes
+                            .unwrap_or_default()
+                            .into_iter()
+                            .map(|prefix| prefix.prefix.expect("listed common prefix"))
+                            .collect::<Vec<_>>();
+                        if filter_prefix.is_some() {
+                            assert_eq!(objects, vec!["z-local"]);
+                            assert_eq!(prefixes, vec!["photos/"]);
+                        } else {
+                            assert_eq!(objects, vec!["a-source", "z-local"]);
+                            assert!(prefixes.is_empty());
+                        }
+                        assert_eq!(output.key_count, Some(2));
+                        assert_eq!(output.is_truncated, Some(false));
+                        assert!(output.next_continuation_token.is_none());
+                        let requests = tokio::time::timeout(Duration::from_secs(5), server)
+                            .await
+                            .expect("finite source connections must finish")
+                            .expect("finite source server must not panic");
+                        assert_eq!(requests.len(), 4);
+                        assert!(!requests[0].contains("continuation-token="));
+                        for (request, cursor) in requests[1..].iter().zip(["A", "B", "C"]) {
+                            assert!(request.contains(&format!("continuation-token={cursor}")), "{request}");
+                        }
+                    }
+                },
+            )
+            .await;
+        });
+    }
+
     fn assert_source_policy_result(result: S3Result<S3Response<ListObjectsV2Output>>, policy: SourceErrorPolicy) {
         match policy {
             SourceErrorPolicy::Propagate => {

rustfs/src/app/storage_api.rs

@@ -634,8 +634,8 @@ pub(crate) mod bucket {
         };
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, OnDemandMigrationConfig, PathStyle, Provider, SourceConfig,
-            SourceCredentials, TlsConfig,
+            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig,
+            PathStyle, Provider, SourceConfig, SourceCredentials, TlsConfig,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
             BucketOdmState, HeadPolicy, OdmLookup, OdmOp, OdmOutcome, OdmStateError, OnDemandMigrationSys, PolicyConfig,
@@ -643,8 +643,8 @@ pub(crate) mod bucket {
             commit_inline, idle_guarded_body,
         };
         pub(crate) use crate::storage::storage_api::ecstore_bucket::on_demand_migration::{
-            ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
-            SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
+            ListEntryKey, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
+            MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
         };
     }
 

scripts/test_security_workflow.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Exercise functional chain dispatch and security evidence without remote VMs."""
+"""Run the security workflow's evidence and result steps without remote VMs."""
 
 from __future__ import annotations
 
@@ -18,20 +18,16 @@ WORKFLOW = ROOT / ".github/workflows/rustfs-security-test.yml"
 CASE_ROW = "| IAM-101 | user CRUD lifecycle | PASS |"
 
 
-def named_steps(job: list[str]) -> dict[str, list[str]]:
-    starts = [i for i, line in enumerate(job) if line.startswith("      - name: ")]
-    return {
-        job[start].split(": ", 1)[1].strip('"'): job[start:end]
-        for start, end in zip(starts, starts[1:] + [len(job)])
-    }
-
-
 class SecurityWorkflowTests(unittest.TestCase):
     def setUp(self) -> None:
         self.source = WORKFLOW.read_text()
         self.job = yaml_block(self.source.splitlines(), "security-test", 2)
         self.assertIsNotNone(self.job)
-        self.steps = named_steps(self.job)
+        starts = [i for i, line in enumerate(self.job) if line.startswith("      - name: ")]
+        self.steps = {
+            self.job[start].split(": ", 1)[1].strip('"'): self.job[start:end]
+            for start, end in zip(starts, starts[1:] + [len(self.job)])
+        }
         self.temp = tempfile.TemporaryDirectory()
         self.addCleanup(self.temp.cleanup)
         self.directory = Path(self.temp.name)
@@ -196,110 +192,6 @@ class SecurityWorkflowTests(unittest.TestCase):
         self.assertNotIn("OLD RUN REPORT", body.read_text())
         self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
 
-    def test_all_ten_suites_hold_the_shared_lock_for_manual_and_chain_runs(self) -> None:
-        for suite in ("upgrade", "s3-compat", "kms", "tier", "storage", "heal", "pool-expand", "security", "replication", "performance"):
-            with self.subTest(suite=suite):
-                source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text().splitlines()
-                # Workflow-level concurrency covers every job, including cleanup,
-                # regardless of trigger or the runner hosting the job.
-                self.assertEqual([
-                    line.strip() for line in yaml_block(source, "concurrency", 0)
-                    if line.strip() and not line.lstrip().startswith("#")
-                ], [
-                    "group: rustfs-shared-functional-tests", "cancel-in-progress: false",
-                ])
-                self.assertIsNotNone(yaml_block(source, "workflow_dispatch", 2))
-                self.assertIsNotNone(yaml_block(source, "repository_dispatch", 2))
-                cleanup_name = "Reset test environment (after)" if suite == "performance" else "Cleanup environment (after)"
-                cleanup = named_steps(yaml_block(source, "jobs", 0))[cleanup_name]
-                self.assertTrue(any(line.startswith("        if:") and "always()" in line for line in cleanup))
-
-    def test_root_dispatches_only_upgrade_and_replication_hands_off_after_failure(self) -> None:
-        for failed_attempts, issue_exit, token in ((0, 0, "fixture"), (2, 0, "fixture"), (3, 0, "fixture"), (3, 7, "fixture"), (0, 0, "")):
-            with self.subTest(failed_attempts=failed_attempts, issue_exit=issue_exit, token=bool(token)):
-                self.setUp()
-                fake_bin = self.directory / "bin"
-                fake_bin.mkdir()
-                commands = {
-                    "gh": '''#!/usr/bin/env bash
-set -euo pipefail
-if [ "$1" = api ]; then
-  printf '%s\\n' "$*" >> "$DISPATCHES"
-  attempt=$(wc -l < "$DISPATCHES")
-  [ "$attempt" -gt "$FAILED_ATTEMPTS" ]
-elif [ "$1 $2" = 'issue create' ]; then
-  printf 'issue\\n' >> "$EXECUTED"
-  while [ "$#" -gt 0 ]; do
-    if [ "$1" = --body-file ]; then
-      cat "$2" > "$CAPTURE_BODY"
-      printf '%s\\n' "$2" > "$CAPTURE_BODY_PATH"
-    fi
-    shift
-  done
-  exit "$ISSUE_EXIT"
-else
-  exit 99
-fi
-''',
-                    "sleep": '#!/bin/sh\nprintf "sleep %s\\n" "$1" >> "$EXECUTED"\n',
-                    "ssh": '#!/bin/sh\nprintf "cleanup\\n" >> "$EXECUTED"\n',
-                }
-                for name, contents in commands.items():
-                    command = fake_bin / name
-                    command.write_text(contents)
-                    command.chmod(0o755)
-                dispatches = self.directory / "dispatches"
-                executed = self.directory / "executed"
-                body = self.directory / "issue-body.md"
-                body_path = self.directory / "issue-body-path"
-                self.env.update(
-                    PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", DISPATCHES=str(dispatches),
-                    EXECUTED=str(executed), CAPTURE_BODY=str(body), CAPTURE_BODY_PATH=str(body_path),
-                    FAILED_ATTEMPTS="0", ISSUE_EXIT=str(issue_exit),
-                    RUSTFS_NODES="fixture-node", RUSTFS_SSH_USER="fixture-user",
-                    RUSTFS_NIGHTLY_PACKAGE_URL="https://example.invalid/package.deb",
-                )
-                self.context.update({"secrets.PF_TESTING_GH_TOKEN": "fixture", "inputs.suite": "all"})
-                driver = (ROOT / ".github/workflows/rustfs-functional-chain.yml").read_text()
-                self.steps = named_steps(yaml_block(driver.splitlines(), "start-chain", 2))
-                self.assertEqual(list(self.steps), ["Dispatch first suite (upgrade)"])
-                started = self.run_step("Dispatch first suite (upgrade)")
-                self.assertEqual(started.returncode, 0, started.stderr)
-                self.assertEqual(dispatches.read_text().splitlines(), [
-                    "api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-upgrade -F client_payload[from_suite]=nightly-build",
-                ])
-                dispatches.unlink()
-
-                replication = (ROOT / ".github/workflows/rustfs-replication-test.yml").read_text()
-                job = yaml_block(replication.splitlines(), "replication-test", 2)
-                self.assertFalse(any(line.startswith("    continue-on-error:") for line in job))
-                self.steps = named_steps(job)
-                handoff = "Continue functional chain (next: Performance)"
-                self.assertIn("        if: ${{ always() && github.event_name == 'repository_dispatch' }}", self.steps[handoff])
-                self.assertFalse(any(line.strip().startswith("continue-on-error:") for line in self.steps[handoff]))
-                self.assertIn("        if: always()", self.steps["Cleanup environment (after)"])
-                self.assertLess(list(self.steps).index("Cleanup environment (after)"), list(self.steps).index(handoff))
-                suite = self.directory / "auto-testing/rustfs-replication-test.sh"
-                suite.write_text('#!/bin/sh\nprintf "suite failed\\n" >> "$EXECUTED"\nexit 17\n')
-                failed = self.run_step("Run replication suite")
-                self.assertEqual(failed.returncode, 17, failed.stderr)
-                cleaned = self.run_step("Cleanup environment (after)")
-                self.assertEqual(cleaned.returncode, 0, cleaned.stderr)
-                self.assertEqual(executed.read_text().splitlines(), ["suite failed", "cleanup"])
-                self.env["FAILED_ATTEMPTS"] = str(failed_attempts)
-                self.context["secrets.PF_TESTING_GH_TOKEN"] = token
-                forwarded = self.run_step(handoff)
-                self.assertEqual(forwarded.returncode == 0, bool(token) and failed_attempts < 3, forwarded.stderr)
-                calls = dispatches.read_text().splitlines() if dispatches.exists() else []
-                self.assertEqual(calls, [
-                    "api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-performance -F client_payload[from_suite]=replication",
-                ] * (min(failed_attempts + 1, 3) if token else 0))
-                if failed_attempts == 3:
-                    self.assertIn("could not hand off from **replication** to **Performance**", body.read_text())
-                    self.assertIn("rustfs-chain-performance", body.read_text())
-                    self.assertEqual(executed.read_text().splitlines().count("issue"), 2 if issue_exit else 1)
-                    self.assertFalse(Path(body_path.read_text().strip()).exists())
-
 
 if __name__ == "__main__":
     unittest.main()