Merge branch 'main' into cxymds/fix-transition-identity-flake

.github/workflows/ci.yml

@@ -372,7 +372,7 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Build debug binary
-        run: cargo build -p rustfs --bins
+        run: cargo build -p rustfs --bins --features e2e-test-hooks
 
       - name: Upload debug binary
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
@@ -402,7 +402,7 @@ jobs:
           github-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Build debug binary with rio-v2
-        run: cargo build -p rustfs --bins --features rio-v2
+        run: cargo build -p rustfs --bins --features rio-v2,e2e-test-hooks
 
       - name: Upload debug binary
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6

crates/e2e_test/src/reliant/tiering.rs

@@ -95,6 +95,7 @@ const MANUAL_ASYNC_PARALLEL_OBJECTS: usize = 64;
 const MANUAL_ACTIVE_CANCEL_OBJECTS: usize = 512;
 const MANUAL_RESTART_CANCEL_OBJECTS: usize = 512;
 const MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT: StdDuration = StdDuration::from_secs(15);
+const MANUAL_TRANSITION_CANCEL_BARRIER_ENV: &str = "RUSTFS_E2E_MANUAL_TRANSITION_CANCEL_BARRIER";
 const MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT: StdDuration = StdDuration::from_secs(90);
 const MANUAL_RESTART_RECOVERY_TIMEOUT: StdDuration = StdDuration::from_secs(80);
 const OBJECT_KEY: &str = "tier/鲁A12345/report.bin";
@@ -1684,8 +1685,15 @@ async fn test_manual_transition_async_active_cancel_reports_terminal_cancelled()
     cold_client.create_bucket().bucket(TIER_BUCKET).send().await?;
 
     let mut hot = RustFSTestEnvironment::new().await?;
-    hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")])
-        .await?;
+    hot.start_rustfs_server_with_env(
+        vec![],
+        &[
+            ("RUSTFS_SCANNER_ENABLED", "false"),
+            ("RUSTFS_SCANNER_CYCLE", "3600"),
+            (MANUAL_TRANSITION_CANCEL_BARRIER_ENV, "1"),
+        ],
+    )
+    .await?;
     let hot_client = hot.create_s3_client();
     add_rustfs_tier(&hot, &cold).await?;
 

crates/filemeta/src/filemeta.rs

@@ -356,6 +356,14 @@ impl FileMeta {
             .versions
             .partition_point(|existing| existing.header.sorts_before(&new_shallow.header));
         self.versions.insert(insert_pos, new_shallow);
+        // `partition_point` only returns the canonical slot when `versions` is
+        // already canonically ordered, and nothing establishes that: `FileMeta::load`
+        // replays the on-disk order verbatim, and metadata written before the
+        // canonical-order fix ordered equal-`mod_time` ties by insertion rather than
+        // by `sorts_before`. Re-sort so an insert leaves the list canonical either
+        // way, matching what `set_idx` already does on the replace path. Cheap: after
+        // a correctly placed insert the slice is a single sorted run.
+        self.sort_by_mod_time();
         Ok(())
 
         // if !ver.valid() {
@@ -1215,6 +1223,57 @@ mod test {
         assert!(fm.versions[0].header.sorts_before(&fm.versions[1].header));
     }
 
+    /// `add_version_filemata` positions an inserted version with
+    /// `partition_point(sorts_before)`, which only yields the canonical slot when
+    /// `versions` is already canonically ordered. Nothing establishes that
+    /// precondition on load: `FileMeta::unmarshal_msg` pushes versions in file
+    /// order, and metadata written before the canonical-order fix ordered
+    /// equal-`mod_time` ties by insertion instead of by `sorts_before`. An insert
+    /// into such a list must still leave it canonical, the same way `set_idx`
+    /// already re-sorts on the replace path.
+    #[test]
+    fn add_version_filemata_canonicalizes_versions_loaded_out_of_order() {
+        let mod_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid test timestamp");
+        let first = version_for_ordering(VersionType::Object, Uuid::from_u128(70), mod_time, 70);
+        let second = version_for_ordering(VersionType::Object, Uuid::from_u128(80), mod_time, 80);
+        let (early, late) = if first.header().sorts_before(&second.header()) {
+            (first, second)
+        } else {
+            (second, first)
+        };
+
+        // Persist the pair the wrong way round to emulate a pre-canonical-order
+        // xl.meta, bypassing add_version_filemata so the bad order really lands on
+        // the wire.
+        let mut legacy = FileMeta::new();
+        legacy
+            .versions
+            .push(FileMetaShallowVersion::try_from(late).expect("shallow later version"));
+        legacy
+            .versions
+            .push(FileMetaShallowVersion::try_from(early).expect("shallow earlier version"));
+
+        let mut loaded =
+            FileMeta::load(&legacy.marshal_msg().expect("serialize legacy metadata")).expect("reload legacy metadata");
+        assert!(
+            !loaded.versions[0].header.sorts_before(&loaded.versions[1].header),
+            "fixture must reach add_version_filemata non-canonically ordered"
+        );
+
+        loaded
+            .add_version_filemata(version_for_ordering(VersionType::Delete, Uuid::from_u128(90), mod_time, 90))
+            .expect("insert equal-time delete marker");
+
+        assert_eq!(loaded.versions.len(), 3);
+        assert!(
+            loaded
+                .versions
+                .windows(2)
+                .all(|pair| pair[0].header.sorts_before(&pair[1].header)),
+            "insert must leave the version list canonically ordered"
+        );
+    }
+
     /// Regression for backlog#799 B16: replication reset state must be
     /// persisted under both internal prefixes, never as a bare ARN. A bare-ARN
     /// key (produced by `ObjectInfo::replication_state`) has no internal prefix,

rustfs/Cargo.toml

@@ -47,6 +47,7 @@ io-scheduler-debug = []  # Enable debug information in I/O scheduler
 tracing-chunk-debug = []  # Enable per-chunk tracing in data plane (high noise, for debugging only)
 full = ["metrics-gpu", "ftps", "swift", "webdav", "sftp", "pyroscope"]
 manual-test-runners = []
+e2e-test-hooks = []
 rio-v2 = ["rustfs-ecstore/rio-v2"]
 pyroscope = ["rustfs-obs/pyroscope"]
 # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`.

rustfs/src/admin/handlers/ilm_transition.rs

@@ -58,6 +58,8 @@ const LOG_SUBSYSTEM_ILM_TRANSITION: &str = "ilm_transition";
 const EVENT_ADMIN_ILM_TRANSITION_STATE: &str = "admin_ilm_transition_state";
 
 static ACTIVE_MANUAL_TRANSITION_SCOPES: OnceLock<Mutex<Vec<ManualTransitionRunScope>>> = OnceLock::new();
+#[cfg(feature = "e2e-test-hooks")]
+const E2E_MANUAL_TRANSITION_CANCEL_BARRIER_ENV: &str = "RUSTFS_E2E_MANUAL_TRANSITION_CANCEL_BARRIER";
 static ACTIVE_MANUAL_TRANSITION_JOBS: OnceLock<Mutex<HashMap<Uuid, CancellationToken>>> = OnceLock::new();
 static MANUAL_TRANSITION_OWNER_ID: OnceLock<String> = OnceLock::new();
 
@@ -748,6 +750,10 @@ async fn start_manual_transition_job(
     let job_heartbeat_shutdown_token = heartbeat_shutdown_token.clone();
     spawn_manual_transition_job_heartbeat(store, job_id, scan_cancel_token, heartbeat_shutdown_token);
     tokio::spawn(async move {
+        #[cfg(feature = "e2e-test-hooks")]
+        if std::env::var_os(E2E_MANUAL_TRANSITION_CANCEL_BARRIER_ENV).is_some() {
+            job_scan_cancel_token.cancelled().await;
+        }
         let result = enqueue_transition_for_existing_objects_scoped(run_store.clone(), &bucket, run_options).await;
         if let Some(final_record) = finalize_manual_transition_job(run_store.clone(), job_id, result).await
             && final_record.is_terminal()