Merge branch 'main' into codex/test-lifecycle-adversarial

.github/pull_request_template.md

@@ -10,16 +10,16 @@ Use N/A when there is no related issue.
 
 ## Summary of Changes
 <!--
-Describe the concrete problem and resulting behavior. For a behavior change, name the input or state that triggers it and the expected outcome. Explain any new dependency or abstraction that the change needs.
+Briefly explain what changed and why reviewers should accept it.
+Focus on behavior, compatibility, and review-relevant context.
 -->
 
 ## Verification
 <!--
-Give 1–3 concrete pieces of evidence for the changed behavior: the test or command, its observed result, and the regression it catches. For a bug fix, record a failing-before/passing-after check or explain why it was unavailable.
+List the commands or checks you ran, for example:
+- `make pre-commit`
 
-Identify the tested commit and any local changes. When testing a prebuilt binary or external service, include its source/version and artifact identity; a successful run against a different build is not evidence for this change.
-
-List relevant checks not run and the remaining risk. Use the validation tier in AGENTS.md; do not run broader checks solely to fill this section. For documentation-only changes, list the applicable documentation checks. Use N/A only when verification is not applicable.
+Use N/A only when verification is not applicable.
 -->
 
 ## Impact

crates/ecstore/src/api/mod.rs

@@ -167,10 +167,9 @@ pub mod bucket {
             idle_guarded_body,
         };
         pub use crate::bucket::on_demand_migration::{
-            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,
+            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,
         };
         pub mod backfill {
             pub use crate::bucket::on_demand_migration::backfill::{

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

@@ -25,13 +25,8 @@ use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
 use std::time::{Duration, Instant};
 
-/// The continuation-token version used by ordinary progressing pages.
+/// The only continuation-token envelope version this build reads and writes.
 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
@@ -116,10 +111,6 @@ 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 {
@@ -132,7 +123,6 @@ impl ListThroughToken {
             source: source.token,
             source_done: source.done,
             last_key,
-            no_progress: None,
         }
     }
 
@@ -180,21 +170,7 @@ 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) => {
-            // 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) if version == u64::from(LIST_THROUGH_TOKEN_VERSION) => {}
         Some(version) => return Err(ListThroughTokenError::UnsupportedVersion(version.min(u64::from(u32::MAX)) as u32)),
         None => return Err(ListThroughTokenError::Malformed),
     }
@@ -312,8 +288,6 @@ 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> {
@@ -378,7 +352,6 @@ pub struct MergeOutcome {
 #[derive(Debug)]
 pub struct ListThroughMerger {
     max_keys: usize,
-    no_progress: Option<u8>,
     last_key: Option<String>,
     local: SideState,
     source: SideState,
@@ -398,7 +371,6 @@ impl ListThroughMerger {
         };
         Self {
             max_keys,
-            no_progress: token.and_then(|token| token.no_progress),
             last_key,
             local,
             source,
@@ -464,18 +436,13 @@ impl ListThroughMerger {
         Ok(())
     }
 
-    /// `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> {
+    pub fn finish(self) -> MergeOutcome {
         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
@@ -541,44 +508,12 @@ impl ListThroughMerger {
         let source_left = !source.disabled && (!source_cursor.done || consumed_source < source.entries.len());
         let is_truncated = local_left || source_left;
 
-        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 {
+        let last_key = consumed_key.or(last_key);
+        MergeOutcome {
             picks,
             is_truncated,
-            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
-            }),
-        })
+            next_token: is_truncated.then(|| ListThroughToken::new(local_cursor, source_cursor, last_key)),
+        }
     }
 }
 
@@ -706,7 +641,7 @@ mod tests {
                     .push_page(fetch.side, kept, truncated, next)
                     .expect("reference provider pages must advance");
             }
-            let outcome = merger.finish(false).expect("valid merge outcome");
+            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");
@@ -789,7 +724,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(false).expect("valid merge outcome");
+        let outcome = merger.finish();
         assert_eq!(outcome.picks.len(), 1);
         assert!(!outcome.is_truncated);
         assert!(outcome.next_token.is_none());
@@ -805,7 +740,6 @@ 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();
@@ -817,7 +751,7 @@ mod tests {
                 Some("local-2".to_string()),
             )
             .expect("local cursor advances");
-        let outcome = merger.finish(false).expect("valid merge outcome");
+        let outcome = merger.finish();
         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");
@@ -896,7 +830,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(false).expect("valid merge outcome");
+        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");
@@ -906,7 +840,7 @@ mod tests {
         merger
             .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
             .expect("source EOF");
-        let outcome = merger.finish(false).expect("valid merge outcome");
+        let outcome = merger.finish();
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -953,7 +887,7 @@ mod tests {
             Err(ListPageError::Repeated)
         );
         merger.disable_source();
-        let outcome = merger.finish(false).expect("valid merge outcome");
+        let outcome = merger.finish();
         assert_eq!(
             outcome.picks,
             vec![MergePick {
@@ -1045,8 +979,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\":3");
-        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(3)));
+        let bumped = encoded.replace("\"v\":1", "\"v\":2");
+        assert_eq!(decode_continuation_token(&bumped), Err(ListThroughTokenError::UnsupportedVersion(2)));
 
         let extra = encoded.replace("{", "{\"x\":1,");
         assert_eq!(decode_continuation_token(&extra), Err(ListThroughTokenError::Malformed));
@@ -1058,257 +992,6 @@ 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,10 +40,9 @@ pub use config::{
     SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
 };
 pub use list_through::{
-    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,
+    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,
 };
 pub use negative_cache::{NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache};
 pub use pull::{

crates/lifecycle/src/core.rs

@@ -43,6 +43,10 @@ const ERR_LIFECYCLE_BUCKET_LOCKED: &str =
     "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an object locked bucket";
 const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration should have at most 1000 rules";
 const ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS: &str = "'Days' for Expiration action must be a positive integer";
+const ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT: &str = "Expiration cannot specify both Days and Date";
+const ERR_LIFECYCLE_MULTIPLE_TRANSITIONS: &str = "Only one Transition action per lifecycle rule is supported";
+const ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS: &str =
+    "Only one NoncurrentVersionTransition action per lifecycle rule is supported";
 const ERR_LIFECYCLE_INVALID_NONCURRENT_EXPIRATION_DAYS: &str =
     "'NoncurrentDays' for NoncurrentVersionExpiration action must be a positive integer";
 const ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS: &str =
@@ -361,6 +365,12 @@ impl Lifecycle for BucketLifecycleConfiguration {
                 {
                     return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS));
                 }
+                if expiration.days.is_some() && expiration.date.is_some() {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidInput,
+                        ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT,
+                    ));
+                }
                 if let Some(expiration_date) = &expiration.date {
                     let date = OffsetDateTime::from(expiration_date.clone());
                     if date.hour() != 0 || date.minute() != 0 || date.second() != 0 || date.nanosecond() != 0 {
@@ -394,11 +404,20 @@ impl Lifecycle for BucketLifecycleConfiguration {
                 }
             }
             if let Some(transitions) = &r.transitions {
+                if transitions.len() > 1 {
+                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, ERR_LIFECYCLE_MULTIPLE_TRANSITIONS));
+                }
                 for transition in transitions {
                     TransitionOps::validate(transition)?;
                 }
             }
             if let Some(noncurrent_transitions) = &r.noncurrent_version_transitions {
+                if noncurrent_transitions.len() > 1 {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidInput,
+                        ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS,
+                    ));
+                }
                 for transition in noncurrent_transitions {
                     NoncurrentVersionTransitionOps::validate(transition)?;
                 }
@@ -473,6 +492,8 @@ impl Lifecycle for BucketLifecycleConfiguration {
     }
 
     async fn eval(&self, obj: &ObjectOpts) -> Event {
+        // A single-object lookup cannot prove how many newer historical versions
+        // survive. Count-dependent actions wait for the complete-group evaluator.
         self.eval_inner(obj, OffsetDateTime::now_utc(), 0).await
     }
 
@@ -536,23 +557,8 @@ impl Lifecycle for BucketLifecycleConfiguration {
             return Event::default();
         };
 
-        if let Some(restore_expires) = obj.restore_expires
-            && restore_expires.unix_timestamp() != 0
-            && now.unix_timestamp() > restore_expires.unix_timestamp()
-        {
-            let mut action = IlmAction::DeleteRestoredAction;
-            if !obj.is_latest {
-                action = IlmAction::DeleteRestoredVersionAction;
-            }
-
-            events.push(Event {
-                action,
-                due: Some(now),
-                rule_id: "".into(),
-                noncurrent_days: 0,
-                newer_noncurrent_versions: 0,
-                storage_class: "".into(),
-            });
+        if let Some(event) = obj.restored_copy_expiry(now) {
+            events.push(event);
         }
 
         if let Some(ref lc_rules) = self.filter_rules(obj).await {
@@ -611,17 +617,12 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     continue;
                 }
 
-                if !obj.is_latest
-                    && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
-                    && let Some(retain_newer_noncurrent_versions) = noncurrent_version_expiration.newer_noncurrent_versions
-                    && newer_noncurrent_versions < usize::try_from(retain_newer_noncurrent_versions).unwrap_or(usize::MAX)
-                {
-                    continue;
-                }
-
                 if !obj.is_latest
                     && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
                     && let Some(noncurrent_days) = noncurrent_version_expiration.noncurrent_days
+                    && noncurrent_version_expiration
+                        .newer_noncurrent_versions
+                        .is_none_or(|retain| usize::try_from(retain).is_ok_and(|retain| newer_noncurrent_versions >= retain))
                 {
                     if let Some(successor_mod_time) = obj.successor_mod_time {
                         let expected_expiry = expected_expiry_time(successor_mod_time, noncurrent_days);
@@ -651,7 +652,11 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     && let Some(noncurrent_version_transition) = rule
                         .noncurrent_version_transitions
                         .as_ref()
+                        .filter(|transitions| transitions.len() == 1)
                         .and_then(|transitions| transitions.first())
+                    && noncurrent_version_transition
+                        .newer_noncurrent_versions
+                        .is_none_or(|retain| usize::try_from(retain).is_ok_and(|retain| newer_noncurrent_versions >= retain))
                     && let Some(storage_class) = noncurrent_version_transition.storage_class.as_ref()
                     && !storage_class.as_str().is_empty()
                     && !obj.delete_marker
@@ -735,7 +740,11 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     }
 
                     if obj.transition_status != TRANSITION_COMPLETE
-                        && let Some(transition) = rule.transitions.as_ref().and_then(|transitions| transitions.first())
+                        && let Some(transition) = rule
+                            .transitions
+                            .as_ref()
+                            .filter(|transitions| transitions.len() == 1)
+                            .and_then(|transitions| transitions.first())
                         && let Some(storage_class) = transition.storage_class.as_ref()
                         && !storage_class.as_str().is_empty()
                     {
@@ -758,18 +767,15 @@ impl Lifecycle for BucketLifecycleConfiguration {
         }
 
         if !events.is_empty() {
-            // Select the winning event using a strict total order (MinIO semantics):
-            // the earliest `due` wins, and ties break toward delete-type actions. A
-            // missing `due` is treated as UNIX_EPOCH. This replaces a hand-written
-            // `sort_by` comparator that was not a strict weak ordering (it could return
-            // `Ordering::Less` for both `(a, b)` and `(b, a)`), which panics on the
-            // repository toolchain and did not deterministically pick the earliest event.
+            // Eligible expiration takes precedence over transition, even when a
+            // failed transition has an earlier deadline. Within each action class,
+            // prefer the earliest deadline using a deterministic total order.
             let event = events
                 .iter()
                 .min_by_key(|event| {
                     (
-                        event.due.unwrap_or(OffsetDateTime::UNIX_EPOCH).unix_timestamp(),
                         ilm_action_priority_rank(&event.action),
+                        event.due.unwrap_or(OffsetDateTime::UNIX_EPOCH).unix_timestamp(),
                     )
                 })
                 .cloned()
@@ -1042,6 +1048,27 @@ impl ObjectOpts {
     pub fn expired_object_deletemarker(&self) -> bool {
         self.delete_marker && self.is_latest && self.num_versions == 1
     }
+
+    pub(crate) fn restored_copy_expiry(&self, now: OffsetDateTime) -> Option<Event> {
+        let restore_expires = self.restore_expires?;
+        // Restore metadata alone does not prove that a durable remote copy exists.
+        if self.transition_status != TRANSITION_COMPLETE
+            || restore_expires.unix_timestamp() == 0
+            || now.unix_timestamp() <= restore_expires.unix_timestamp()
+        {
+            return None;
+        }
+        let action = if self.is_latest {
+            IlmAction::DeleteRestoredAction
+        } else {
+            IlmAction::DeleteRestoredVersionAction
+        };
+        expiration_action_has_valid_target(action, self.version_id, self.is_latest, self.delete_marker).then(|| Event {
+            action,
+            due: Some(now),
+            ..Default::default()
+        })
+    }
 }
 
 /// Returns whether an expiry action has enough identity to target the object
@@ -1064,11 +1091,8 @@ pub fn expiration_action_has_valid_target(
     }
 }
 
-/// Total-order rank for lifecycle actions used to break `due` ties.
-///
-/// Delete-type actions rank before every other action so that, when two events
-/// share the same `due`, a delete wins (MinIO semantics). The concrete numeric
-/// values only matter relative to each other.
+/// Eligible logical expiration takes precedence over transition and restore-copy
+/// cleanup. Deadlines break ties within an action class.
 fn ilm_action_priority_rank(action: &IlmAction) -> u8 {
     match action {
         IlmAction::DeleteAllVersionsAction
@@ -4159,6 +4183,392 @@ mod tests {
         assert_eq!(event.action, IlmAction::NoneAction);
     }
 
+    mod adversarial_regressions {
+        use super::*;
+        use s3s::dto::NoncurrentVersionExpiration;
+
+        fn run(test: impl std::future::Future<Output = ()>) {
+            with_default_ilm_process_time(|| {
+                tokio::runtime::Builder::new_current_thread()
+                    .build()
+                    .expect("lifecycle regression runtime should build")
+                    .block_on(test);
+            });
+        }
+
+        fn noncurrent_object() -> ObjectOpts {
+            ObjectOpts {
+                name: "logs/object".to_string(),
+                mod_time: Some(datetime!(2020-01-01 00:00:00 UTC)),
+                successor_mod_time: Some(datetime!(2020-01-02 00:00:00 UTC)),
+                version_id: Some(Uuid::from_u128(1)),
+                size: 1024 * 1024,
+                ..Default::default()
+            }
+        }
+
+        #[test]
+        #[serial]
+        fn noncurrent_transition_retains_the_requested_newer_versions() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("retain-two-hot-versions"));
+                rule.filter = Some(LifecycleRuleFilter::default());
+                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(1),
+                    newer_noncurrent_versions: Some(2),
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]);
+                let lc = Arc::new(BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                });
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("valid noncurrent transition policy");
+                let objects = (0..4)
+                    .map(|index| ObjectOpts {
+                        mod_time: Some(datetime!(2020-01-05 00:00:00 UTC) - Duration::days(index)),
+                        successor_mod_time: (index > 0).then_some(datetime!(2020-01-06 00:00:00 UTC) - Duration::days(index)),
+                        version_id: Some(Uuid::from_u128(u128::try_from(index + 1).expect("small version index"))),
+                        is_latest: index == 0,
+                        num_versions: 4,
+                        ..noncurrent_object()
+                    })
+                    .collect::<Vec<_>>();
+                let actions = crate::Evaluator::new(lc)
+                    .eval(&objects)
+                    .await
+                    .expect("complete version chain should evaluate")
+                    .into_iter()
+                    .map(|event| event.action)
+                    .collect::<Vec<_>>();
+                assert_eq!(
+                    actions,
+                    [
+                        IlmAction::NoneAction,
+                        IlmAction::NoneAction,
+                        IlmAction::NoneAction,
+                        IlmAction::TransitionVersionAction
+                    ],
+                    "the two newest noncurrent versions must remain in their current storage class"
+                );
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn noncurrent_transition_checks_count_age_and_single_object_context() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("retain-two"));
+                rule.filter = Some(LifecycleRuleFilter::default());
+                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(3),
+                    newer_noncurrent_versions: Some(2),
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]);
+                let mut lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("valid counted transition");
+                let object = noncurrent_object();
+                let now = datetime!(2020-01-10 00:00:00 UTC);
+                for (newer, expected) in [
+                    (0, IlmAction::NoneAction),
+                    (1, IlmAction::NoneAction),
+                    (2, IlmAction::TransitionVersionAction),
+                    (3, IlmAction::TransitionVersionAction),
+                ] {
+                    assert_eq!(lc.eval_inner(&object, now, newer).await.action, expected, "newer count: {newer}");
+                }
+                assert_eq!(
+                    lc.eval_inner(&object, datetime!(2020-01-04 00:00:00 UTC), 2).await.action,
+                    IlmAction::NoneAction,
+                    "the retention count does not replace the age condition"
+                );
+                assert_eq!(
+                    lc.eval(&object).await.action,
+                    IlmAction::NoneAction,
+                    "a single-object lookup must not assume a complete version history"
+                );
+                for retain in [None, Some(0), Some(-1), Some(i32::MAX)] {
+                    lc.rules[0]
+                        .noncurrent_version_transitions
+                        .as_mut()
+                        .expect("transition exists")[0]
+                        .newer_noncurrent_versions = retain;
+                    let expected = if matches!(retain, None | Some(0)) {
+                        IlmAction::TransitionVersionAction
+                    } else {
+                        IlmAction::NoneAction
+                    };
+                    assert_eq!(lc.eval_inner(&object, now, 2).await.action, expected, "retention: {retain:?}");
+                }
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn noncurrent_expiration_and_transition_have_independent_retention_counts() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("independent-counts"));
+                rule.filter = Some(LifecycleRuleFilter::default());
+                rule.noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
+                    noncurrent_days: Some(90),
+                    newer_noncurrent_versions: Some(4),
+                });
+                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(30),
+                    newer_noncurrent_versions: Some(2),
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]);
+                let lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("valid independent retention limits");
+                let object = noncurrent_object();
+                let now = datetime!(2020-05-01 00:00:00 UTC);
+                for (newer, expected) in [
+                    (1, IlmAction::NoneAction),
+                    (2, IlmAction::TransitionVersionAction),
+                    (3, IlmAction::TransitionVersionAction),
+                    (4, IlmAction::DeleteVersionAction),
+                ] {
+                    assert_eq!(lc.eval_inner(&object, now, newer).await.action, expected, "newer count: {newer}");
+                }
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn expiration_retention_does_not_skip_an_independent_transition() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("transition-then-expire"));
+                rule.filter = Some(LifecycleRuleFilter::default());
+                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
+                    noncurrent_days: Some(1),
+                    newer_noncurrent_versions: None,
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]);
+                let mut lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                let object = noncurrent_object();
+                let now = datetime!(2020-01-10 00:00:00 UTC);
+                let transition_only = lc.eval_inner(&object, now, 0).await;
+                assert_eq!(transition_only.action, IlmAction::TransitionVersionAction);
+
+                lc.rules[0].noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
+                    noncurrent_days: Some(90),
+                    newer_noncurrent_versions: Some(2),
+                });
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("valid combined policy");
+                let combined = lc.eval_inner(&object, now, 0).await;
+                assert_eq!(combined.action, transition_only.action, "retention limits expiration, not transition");
+                assert_eq!(combined.storage_class, transition_only.storage_class);
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn current_transition_rejects_multiple_stages_in_any_order() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("two-current-transitions"));
+                rule.transitions = Some(vec![
+                    Transition {
+                        date: Some(datetime!(2020-03-01 00:00:00 UTC).into()),
+                        days: None,
+                        storage_class: Some(TransitionStorageClass::from_static("COLD")),
+                    },
+                    Transition {
+                        date: Some(datetime!(2020-01-03 00:00:00 UTC).into()),
+                        days: None,
+                        storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                    },
+                ]);
+                let mut lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                let object = ObjectOpts {
+                    is_latest: true,
+                    ..noncurrent_object()
+                };
+                let now = datetime!(2020-01-10 00:00:00 UTC);
+                for status in [ExpirationStatus::ENABLED, ExpirationStatus::DISABLED] {
+                    lc.rules[0].status = ExpirationStatus::from_static(status);
+                    for _ in 0..2 {
+                        let err = lc
+                            .validate(&ObjectLockConfiguration::default())
+                            .await
+                            .expect_err("multiple transition stages must be rejected");
+                        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+                        assert_eq!(err.to_string(), ERR_LIFECYCLE_MULTIPLE_TRANSITIONS);
+                        assert_eq!(
+                            lc.eval_inner(&object, now, 0).await.action,
+                            IlmAction::NoneAction,
+                            "legacy multi-stage configurations must not silently execute their first stage"
+                        );
+                        lc.rules[0]
+                            .transitions
+                            .as_mut()
+                            .expect("transition array is present")
+                            .reverse();
+                    }
+                }
+                lc.rules[0]
+                    .transitions
+                    .as_mut()
+                    .expect("transition array is present")
+                    .remove(0);
+                lc.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::ENABLED);
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("one stage is supported");
+                let event = lc.eval_inner(&object, now, 0).await;
+                assert_eq!(event.action, IlmAction::TransitionAction);
+                assert_eq!(event.storage_class, "WARM");
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn noncurrent_transition_rejects_multiple_stages_in_any_order() {
+            run(async {
+                let mut rule = enabled_rule(None, None, Some("two-noncurrent-transitions"));
+                rule.noncurrent_version_transitions = Some(vec![
+                    NoncurrentVersionTransition {
+                        noncurrent_days: Some(30),
+                        newer_noncurrent_versions: None,
+                        storage_class: Some(TransitionStorageClass::from_static("COLD")),
+                    },
+                    NoncurrentVersionTransition {
+                        noncurrent_days: Some(1),
+                        newer_noncurrent_versions: None,
+                        storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                    },
+                ]);
+                let mut lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                let object = noncurrent_object();
+                let now = datetime!(2020-01-10 00:00:00 UTC);
+                for status in [ExpirationStatus::ENABLED, ExpirationStatus::DISABLED] {
+                    lc.rules[0].status = ExpirationStatus::from_static(status);
+                    for _ in 0..2 {
+                        let err = lc
+                            .validate(&ObjectLockConfiguration::default())
+                            .await
+                            .expect_err("multiple noncurrent transition stages must be rejected");
+                        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+                        assert_eq!(err.to_string(), ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS);
+                        assert_eq!(
+                            lc.eval_inner(&object, now, 0).await.action,
+                            IlmAction::NoneAction,
+                            "legacy multi-stage configurations must not silently execute their first stage"
+                        );
+                        lc.rules[0]
+                            .noncurrent_version_transitions
+                            .as_mut()
+                            .expect("transition array is present")
+                            .reverse();
+                    }
+                }
+                lc.rules[0]
+                    .noncurrent_version_transitions
+                    .as_mut()
+                    .expect("transition array is present")
+                    .remove(0);
+                lc.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::ENABLED);
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("one stage is supported");
+                let event = lc.eval_inner(&object, now, 0).await;
+                assert_eq!(event.action, IlmAction::TransitionVersionAction);
+                assert_eq!(event.storage_class, "WARM");
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn expiration_rejects_simultaneous_days_and_date() {
+            run(async {
+                let mut lc = BucketLifecycleConfiguration {
+                    rules: vec![enabled_rule(
+                        Some(LifecycleExpiration {
+                            days: Some(1),
+                            ..Default::default()
+                        }),
+                        None,
+                        Some("ambiguous-expiry"),
+                    )],
+                    expiry_updated_at: None,
+                };
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("a single Days expiration is valid");
+                lc.rules[0].expiration.as_mut().expect("expiration is present").date =
+                    Some(datetime!(2099-01-01 00:00:00 UTC).into());
+                let err = lc
+                    .validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect_err("Days and Date are mutually exclusive; accepting both silently overrides Days");
+                assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+                assert_eq!(err.to_string(), ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT);
+            });
+        }
+
+        #[test]
+        #[serial]
+        fn overdue_transition_does_not_starve_permanent_expiration() {
+            run(async {
+                let mut rule = enabled_rule(
+                    Some(LifecycleExpiration {
+                        days: Some(90),
+                        ..Default::default()
+                    }),
+                    None,
+                    Some("archive-then-delete"),
+                );
+                rule.transitions = Some(vec![Transition {
+                    days: Some(30),
+                    date: None,
+                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
+                }]);
+                let lc = BucketLifecycleConfiguration {
+                    rules: vec![rule],
+                    expiry_updated_at: None,
+                };
+                lc.validate(&ObjectLockConfiguration::default())
+                    .await
+                    .expect("valid transition and expiration policy");
+                let object = ObjectOpts {
+                    is_latest: true,
+                    version_id: None,
+                    transition_status: TRANSITION_PENDING.to_string(),
+                    ..noncurrent_object()
+                };
+                let before_expiration = lc.eval_inner(&object, datetime!(2020-02-15 00:00:00 UTC), 0).await;
+                assert_eq!(before_expiration.action, IlmAction::TransitionAction);
+                let overdue = lc.eval_inner(&object, datetime!(2020-05-01 00:00:00 UTC), 0).await;
+                assert_eq!(
+                    overdue.action,
+                    IlmAction::DeleteAction,
+                    "an unavailable tier must not prevent permanent expiration indefinitely"
+                );
+            });
+        }
+    }
+
     /// Property-based tests for the rule evaluator (backlog#1148 ilm-14,
     /// follow-up to backlog#1030 / rustfs#4455).
     ///
@@ -4169,7 +4579,7 @@ mod tests {
     ///
     /// * `eval_inner` never panics and is deterministic for a fixed input;
     /// * the winning event matches an independently recomputed candidate set:
-    ///   earliest `due` wins, ties break toward delete-class actions (the
+    ///   eligible expiration wins over transition, then earliest `due` wins (the
     ///   `min_by_key` selection that replaced the rustfs#4455 comparator);
     /// * `expected_expiry_time` is monotonically non-decreasing in `days` and
     ///   always lands on the processing boundary, both at production defaults
@@ -4458,8 +4868,8 @@ mod tests {
         /// consider for a live current version under `selection`-shaped rules
         /// (expiration and first-transition only, no filters): expiration
         /// fires when `now >= due`, transition when `now > due` and the object
-        /// has not already transitioned. Selection semantics under test:
-        /// earliest due wins, ties prefer delete-class.
+        /// has not already transitioned. Eligible expiration wins over transition;
+        /// the earliest deadline wins within the selected action class.
         fn oracle_candidates(lc: &BucketLifecycleConfiguration, obj: &ObjectOpts, now: OffsetDateTime) -> Vec<Candidate> {
             let mod_time = obj.mod_time.expect("selection strategy always sets mod_time");
             let mut candidates = Vec::new();
@@ -4548,8 +4958,8 @@ mod tests {
             /// Differential test of winner selection (the rustfs#4455 fix):
             /// for a live current version under randomized expiration and
             /// transition rules, `eval_inner`'s winner must carry the
-            /// minimum `(due, rank)` of the independently recomputed
-            /// candidate set — earliest due wins, ties prefer delete-class —
+            /// earliest expiration from the independently recomputed candidate
+            /// set, or the earliest transition when no expiration is eligible,
             /// and must be `NoneAction` exactly when that set is empty.
             #[test]
             #[serial]
@@ -4578,7 +4988,13 @@ mod tests {
 
                 // Oracle and evaluator must observe the same (pinned) time env.
                 let (event, expected) = with_production_time_env(|| {
-                    let expected = oracle_candidates(&lc, &obj, now).into_iter().min();
+                    let candidates = oracle_candidates(&lc, &obj, now);
+                    let expected = candidates
+                        .iter()
+                        .filter(|(_, rank)| *rank == 0)
+                        .min()
+                        .copied()
+                        .or_else(|| candidates.into_iter().min());
                     let rt = tokio::runtime::Builder::new_current_thread()
                         .enable_all()
                         .build()

crates/lifecycle/src/evaluator.rs

@@ -116,13 +116,10 @@ impl Evaluator {
                             break 'top_loop;
                         }
                     }
-                    IlmAction::DeleteAction
-                    | IlmAction::DeleteRestoredAction
-                    | IlmAction::DeleteVersionAction
-                    | IlmAction::DeleteRestoredVersionAction
-                        if self.is_object_locked(obj) =>
-                    {
-                        event = Event::default();
+                    // Restore expiry removes only the temporary local copy; the
+                    // retained logical version and its remote data remain intact.
+                    IlmAction::DeleteAction | IlmAction::DeleteVersionAction if self.is_object_locked(obj) => {
+                        event = obj.restored_copy_expiry(now).unwrap_or_default();
                     }
                     _ => {}
                 }
@@ -206,6 +203,95 @@ mod tests {
 
     use super::*;
     use rustfs_replication::{ReplicationStatusType, VersionPurgeStatusType};
+
+    #[tokio::test]
+    async fn adversarial_restore_expiry_survives_legal_hold() {
+        let mut policy = (*latest_expiration_lifecycle()).clone();
+        policy.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::DISABLED);
+        let policy = Arc::new(policy);
+        policy
+            .validate(&lock_enabled_without_default_retention())
+            .await
+            .expect("valid disabled lifecycle rule");
+        let mut objects = [true, false].map(|is_latest| ObjectOpts {
+            is_latest,
+            num_versions: 2,
+            mod_time: Some(
+                OffsetDateTime::from_unix_timestamp(if is_latest { 1_200_000 } else { 1_000_000 })
+                    .expect("fixed version timestamp"),
+            ),
+            successor_mod_time: (!is_latest)
+                .then(|| OffsetDateTime::from_unix_timestamp(1_200_000).expect("fixed successor timestamp")),
+            transition_status: crate::TRANSITION_COMPLETE.to_string(),
+            restore_expires: Some(OffsetDateTime::from_unix_timestamp(2_000_000).expect("fixed expired restore timestamp")),
+            ..current_object_opts(ReplicationStatusType::Completed)
+        });
+        let evaluator = Evaluator::new(policy).with_lock_retention(Some(lock_enabled_without_default_retention()));
+        let expected = [IlmAction::DeleteRestoredAction, IlmAction::DeleteRestoredVersionAction];
+        let unlocked = evaluator
+            .eval(&objects)
+            .await
+            .expect("unlocked restored versions should evaluate");
+        assert_eq!(unlocked.iter().map(|event| event.action).collect::<Vec<_>>(), expected);
+
+        for object in &mut objects {
+            object
+                .user_defined
+                .insert(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), "ON".to_string());
+        }
+        let locked = evaluator
+            .eval(&objects)
+            .await
+            .expect("locked restored versions should evaluate");
+        assert_eq!(
+            locked.iter().map(|event| event.action).collect::<Vec<_>>(),
+            expected,
+            "expiring a restored local copy preserves the retained logical version and remote object"
+        );
+
+        let mut expiring_policy = (*latest_expiration_lifecycle()).clone();
+        expiring_policy.rules[0].noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
+            noncurrent_days: Some(1),
+            newer_noncurrent_versions: None,
+        });
+        let expiring_evaluator =
+            Evaluator::new(Arc::new(expiring_policy)).with_lock_retention(Some(lock_enabled_without_default_retention()));
+        let locked = expiring_evaluator
+            .eval(&objects)
+            .await
+            .expect("locked expired versions should evaluate");
+        assert_eq!(
+            locked.iter().map(|event| event.action).collect::<Vec<_>>(),
+            expected,
+            "blocked logical expiration must still allow an eligible restore-copy cleanup"
+        );
+
+        for status in [ReplicationStatusType::Pending, ReplicationStatusType::Failed] {
+            for object in &mut objects {
+                object.replication_status = status.clone();
+            }
+            for evaluator in [&evaluator, &expiring_evaluator] {
+                let events = evaluator.eval(&objects).await.expect("pending replication should evaluate");
+                assert!(events.iter().all(|event| event.action == IlmAction::NoneAction));
+            }
+        }
+        for object in &mut objects {
+            object.replication_status = ReplicationStatusType::Completed;
+        }
+        for transition_status in ["", crate::TRANSITION_PENDING, "unknown"] {
+            for object in &mut objects {
+                object.transition_status = transition_status.to_string();
+            }
+            for evaluator in [&evaluator, &expiring_evaluator] {
+                let events = evaluator.eval(&objects).await.expect("incomplete transition should evaluate");
+                assert!(
+                    events.iter().all(|event| event.action == IlmAction::NoneAction),
+                    "restore metadata cannot authorize cleanup without a completed transition"
+                );
+            }
+        }
+    }
+
     fn expired_marker_lifecycle() -> Arc<BucketLifecycleConfiguration> {
         Arc::new(BucketLifecycleConfiguration {
             expiry_updated_at: None,

docs/operations/on-demand-migration.md

@@ -7,16 +7,6 @@ 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 |

docs/operations/tier-ilm-debugging.md

@@ -22,6 +22,16 @@
 | `FileMeta` / `FileInfo` / version metadata | `crates/filemeta/src/` |
 | Dual-key internal metadata helpers (`insert_bytes` / `get_bytes`) | `crates/utils/src/http/metadata_compat.rs` |
 
+## Lifecycle rule limits and evaluation
+
+Each lifecycle rule supports at most one `Transition` and one `NoncurrentVersionTransition`. A version can make one initial transition; chaining additional tiers after it reaches `complete` is not supported. Splitting stages across overlapping rules does not enable a transition chain. `PutBucketLifecycleConfiguration` rejects multiple entries in either transition array with `InvalidArgument`, including in disabled rules. Existing stored multi-entry arrays are not executed; replace each with a single intended destination. Independent expiration actions in the rule remain eligible.
+
+`Expiration.Days` and `Expiration.Date` are mutually exclusive. A request containing both is rejected instead of silently selecting the date. When expiration and transition are both eligible, expiration takes precedence; a failed earlier transition does not keep an expired object indefinitely. Deadlines select the earliest action within the same action class.
+
+Noncurrent expiration and transition have independent `NewerNoncurrentVersions` limits. A transition with a positive limit waits for a complete version-group evaluation to establish that enough newer noncurrent versions remain. Single-object evaluation, including the current manual transition and immediate-enqueue paths, conservatively defers these counted transitions to the lifecycle scanner. An unmet expiration retention limit does not suppress a separately eligible transition.
+
+An expired restored local copy can be cleaned up under Object Lock because the retained logical version and remote data remain intact. Cleanup requires a completed transition and still waits for pending or failed replication. The storage layer revalidates the source identity and restore metadata before removing the local copy; restore headers alone do not authorize cleanup.
+
 ## Free-version recovery controls
 
 The dedicated free-version recovery loop is enabled by default and is independent of the data scanner and heal switches. Setting `RUSTFS_SCANNER_ENABLED=false` does not stop this repair loop. Set `RUSTFS_TIER_FREE_VERSION_RECOVERY_ENABLED=false` before process startup to disable only the dedicated persisted-marker walk. That setting does not disable lifecycle workers or prevent another scanner path from discovering a free version, and it can leave remote cleanup markers pending for longer, so use it as a break-glass pressure control rather than a cleanup mechanism.

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, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
-    MergeSide, OnDemandMigrationSys, SOURCE_LIST_MAX_RATE_WAIT, SourceClient, SourceError, SourceErrorPolicy, SourceListPlan,
+    BucketOdmState, ListEntryKey, 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,9 +51,6 @@ 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;
@@ -267,18 +264,7 @@ pub(crate) async fn merged_list_objects_v2(
         buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.into_iter().map(Some));
     }
 
-    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 outcome = merger.finish();
     let mut objects = Vec::with_capacity(outcome.picks.len());
     let mut prefixes = Vec::new();
     let mut source_only_keys = Vec::new();
@@ -444,8 +430,7 @@ 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, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig, PathStyle, PolicyConfig, Provider, SourceConfig,
-        SourceCredentials, TlsConfig,
+        FilterConfig, 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;
@@ -463,7 +448,6 @@ mod tests {
             source: Some("source-2".to_string()),
             source_done: false,
             last_key: Some("k".to_string()),
-            no_progress: None,
         }
     }
 
@@ -531,24 +515,6 @@ 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!(
@@ -585,30 +551,14 @@ 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, _) = tokio::select! {
-                    _ = server_stop.cancelled() => break,
-                    accepted = listener.accept() => accepted.expect("accept source listing"),
-                };
+                let (mut stream, _) = listener.accept().await.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") {
@@ -638,7 +588,7 @@ mod tests {
             }
             requests
         });
-        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server), stop)
+        (format!("http://{address}"), tokio_util::task::AbortOnDropHandle::new(server))
     }
 
     fn source_xml(next: Option<&str>, truncated: bool, key: Option<&str>) -> String {
@@ -666,12 +616,12 @@ mod tests {
         }
     }
 
-    async fn source_policy_input(
-        endpoint: String,
+    async fn source_policy_request(
+        pages: Vec<String>,
         policy: SourceErrorPolicy,
         resume_source: Option<&str>,
         filter_prefix: Option<&str>,
-    ) -> (ListThroughTestState, ListObjectsV2Input) {
+    ) -> (S3Result<S3Response<ListObjectsV2Output>>, Vec<String>) {
         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());
@@ -688,8 +638,9 @@ 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(),
         };
@@ -734,7 +685,6 @@ 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())
         });
@@ -751,10 +701,6 @@ 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,
@@ -766,23 +712,12 @@ mod tests {
             service: None,
             trailing_headers: None,
         };
-        tokio::time::timeout(
+        let result = 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")
-    }
-
-    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;
+        .expect("listing must complete within its bounded source budget");
         let requests = tokio::time::timeout(Duration::from_secs(5), server)
             .await
             .expect("source connections must finish")
@@ -907,244 +842,6 @@ 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, MAX_LIST_NO_PROGRESS_PAGES, OnDemandMigrationConfig,
-            PathStyle, Provider, SourceConfig, SourceCredentials, TlsConfig,
+            BREAKER_FAILURE_THRESHOLD, BreakerState, FilterConfig, 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, ListPageError, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError,
-            MergeSide, SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
+            ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken, ListThroughTokenError, MergeSide,
+            SOURCE_LIST_MAX_RATE_WAIT, SourceListPlan, decode_continuation_token, source_list_plan,
         };
     }