fix(heal): retain unmatched migration manifest evidence

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/on_demand_migration/concurrency_test.rs

@@ -20,9 +20,10 @@
 //! journal (`count_requests`) carries the assertion in every one of them.
 
 use super::common::{BoxError, OdmTestEnv, RawResponse, SeedObject, start_configured_env};
-use crate::fake_s3_target::Operation;
+use crate::fake_s3_target::{FaultAction, Operation};
 use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
 use bytes::Bytes;
+use futures::{StreamExt, TryStreamExt};
 use std::time::Duration;
 
 type TestResult = Result<(), BoxError>;
@@ -145,14 +146,38 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
     .await?;
 
     let body = payload(128 * 1024);
+    let blocker = "queue/blocker.bin";
+    env.seed_source(SOURCE_BUCKET, &[SeedObject::new(blocker, body.clone())]);
+    // The one-chunk range completes immediately; its full background pull
+    // occupies the only slot while the remaining requests fill the queue.
+    env.source.inject_for_key(
+        Operation::GetObject,
+        blocker,
+        FaultAction::SlowSendBody {
+            chunk_bytes: 1024,
+            delay: Duration::from_millis(100),
+        },
+        2,
+    );
+    let response = env
+        .raw_object_request(http::Method::GET, bucket, blocker, &[("range", "bytes=0-1023")])
+        .await?;
+    assert_eq!(response.status, 206);
+    assert_eq!(response.body, body.slice(0..1024));
+    env.wait_for_status_counter(bucket, "/inflight_pulls", 1, SETTLE).await?;
+
     let keys: Vec<String> = (0..REQUESTS).map(|index| format!("queue/object-{index:03}.bin")).collect();
     let seeds: Vec<SeedObject> = keys.iter().map(|key| SeedObject::new(key.clone(), body.clone())).collect();
     env.seed_source(SOURCE_BUCKET, &seeds);
 
-    let responses: Vec<RawResponse> = futures::future::try_join_all(
+    // Bound source connections below the fixture's limit while still
+    // submitting all 100 requests to the eight-slot background queue.
+    let responses: Vec<RawResponse> = futures::stream::iter(
         keys.iter()
             .map(|key| env.raw_object_request(http::Method::GET, bucket, key, &[("range", "bytes=0-1023")])),
     )
+    .buffered(16)
+    .try_collect()
     .await?;
     for (key, response) in keys.iter().zip(&responses) {
         assert_eq!(response.status, 206, "{key}: {}", String::from_utf8_lossy(&response.body));
@@ -168,6 +193,15 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         .wait_for_status_counter(bucket, "/counters/pull_failures_total/queue_full", 1, SETTLE)
         .await?;
     assert!(queue_full > 0, "a 100-deep burst must overflow an 8-slot queue");
+    let queue_full = usize::try_from(queue_full)?;
+    assert!(queue_full <= REQUESTS);
+    env.wait_for_status_counter(
+        bucket,
+        "/counters/pulled_objects_total/background",
+        u64::try_from(REQUESTS + 1 - queue_full)?,
+        SETTLE,
+    )
+    .await?;
 
     let ranged_reads: usize = keys.iter().map(|key| source_get_count(&env, key)).sum();
     assert!(
@@ -175,9 +209,6 @@ async fn test_odm_range_burst_overflows_the_pull_queue_without_failing_clients()
         "every reader is served from the source: {ranged_reads} GETs for {REQUESTS} readers"
     );
     let dropped = keys.iter().filter(|key| source_get_count(&env, key) == 1).count();
-    assert!(
-        dropped > 0,
-        "the overflowed keys are the ones with no backfill GET, but every key got one"
-    );
+    assert_eq!(dropped, queue_full, "only overflowed keys remain without a background GET");
     Ok(())
 }

crates/e2e_test/src/on_demand_migration/list_through_test.rs

@@ -265,16 +265,13 @@ async fn list_through_rejects_a_tampered_continuation_token() -> TestResult {
     let decoded = String::from_utf8(base64_simd::STANDARD.decode_to_vec(token.as_bytes())?)?;
     assert!(decoded.contains("\"t\":\"odm-list\""), "the merged token is an envelope: {decoded}");
 
-    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":2").as_bytes());
-    let rejected = env
-        .raw_list_objects_v2(bucket, &format!("continuation-token={tampered}"))
-        .await?;
-    assert_eq!(
-        rejected.status,
-        400,
-        "a bumped token version is a client error: {}",
-        String::from_utf8_lossy(&rejected.body)
-    );
+    let tampered = base64_simd::STANDARD.encode_to_string(decoded.replace("\"v\":1", "\"v\":3").as_bytes());
+    assert_ne!(tampered, token, "the test must change the token version");
+    let query = serde_urlencoded::to_string([("continuation-token", tampered.as_str())])?;
+    let rejected = env.raw_list_objects_v2(bucket, &query).await?;
+    let error_body = String::from_utf8_lossy(&rejected.body);
+    assert_eq!(rejected.status, 400, "a bumped token version is a client error: {}", error_body);
+    assert!(error_body.contains("<Code>InvalidArgument</Code>"), "{error_body}");
     Ok(())
 }
 

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -33,6 +33,9 @@ use std::collections::HashMap;
 use tokio::io::AsyncReadExt;
 use uuid::Uuid;
 
+/// Explicit pending migration; never activates the production writer or GC.
+pub mod migration;
+
 // Root-level control files avoid requiring a new directory before the first
 // atomic commit. They remain inside the storage owner's metadata volume.
 const PAYLOAD_PATHS: [&str; 2] = [".heal-mrf-snapshot.0.bin", ".heal-mrf-snapshot.1.bin"];
@@ -66,6 +69,23 @@ struct Manifest {
 }
 
 impl Manifest {
+    fn encode(owner: Uuid, sequence: u64, payload: &[u8]) -> Result<Vec<u8>, SnapshotError> {
+        let mut bytes = Vec::with_capacity(MANIFEST_LEN);
+        bytes.extend_from_slice(MAGIC);
+        bytes.push(VERSION);
+        bytes.extend_from_slice(owner.as_bytes());
+        bytes.extend_from_slice(&sequence.to_le_bytes());
+        bytes.extend_from_slice(
+            &u64::try_from(payload.len())
+                .map_err(|_| SnapshotError::TooLarge)?
+                .to_le_bytes(),
+        );
+        bytes.extend_from_slice(&Sha256::digest(payload));
+        bytes.extend_from_slice(&Sha256::digest(&bytes));
+        Self::decode(&bytes, payload.len())?;
+        Ok(bytes)
+    }
+
     fn decode(bytes: &[u8], limit: usize) -> Result<Self, SnapshotError> {
         if bytes.len() != MANIFEST_LEN || &bytes[..8] != MAGIC {
             return Err(SnapshotError::Corrupt);

docs/testing/mrf-pending-migration.md

@@ -0,0 +1,21 @@
+# Pending MRF Migration
+
+`heal::mrf_queue::snapshot::migration` exposes explicit capture, staging, and readback of pending legacy responsibility evidence. Nothing invokes it from the production MRF consumer. It does not enable the committed-snapshot writer, freeze legacy ingress, acknowledge durable admission, or authorize source garbage collection.
+
+The caller supplies every configured local disk slot, including missing slots. Missing/unformatted/duplicate disks, unavailable metadata volumes, invalid records, and aggregate byte/record/source-history overflow fail closed. A valid empty source observation can inherit earlier pending responsibilities, but staging without any current or inherited responsibility is rejected. Both legacy paths retain their original bytes, disk identity, absent-versus-empty state, and SHA-256 digest. Complete subset/superset replicas become conservative pending evidence, never a claimed newest legacy snapshot. Raw record replay preserves kind, scope and nil/absent version encodings; unknown incarnation stays unknown.
+
+Staging writes only `.heal-mrf-import-pending.{0,1}.bin`, `.heal-mrf-import-commit.{0,1}.bin`, and `.heal-mrf-import-claim.bin` under the metadata volume. It reuses the committed reader's manifest codec and the storage owner's conditional-file operation, including the configured metadata durability policy. A candidate is written before sources are revalidated, its manifest is then committed, and committed bytes plus source coverage are read back before success. Success is pending staging evidence, not a power-loss or cluster-quorum durability receipt.
+
+A successor inherits prior source bytes even if a replay consumer has already read or admitted their records. Independent byte, record, and source-history limits include inherited evidence, and source identities use a hash index with full-byte conflict checks. There is no completion-based pruning. A changed source blocks recovery of that pending generation; an explicit new capture can stage a successor that retains both the previous and current responsibilities. The inactive slot is replaced while the preceding committed slot remains intact. Any corrupt, unsupported or over-budget slot blocks selection rather than falling back to an older generation. The reader first collects bounded lineage evidence from at most two slots per configured disk. A payload without a manifest is repairable only when its length and digest match the complete retry candidate after inheritance, or any independently validated committed payload, including an older generation on another replica. Unknown payloads still block writing. Retries also repair missing replica manifests.
+
+All participating disks are claimed in disk-identity order through CAS. Normal completion conditionally releases only the current invocation's claim, attempts every acquired claim even after a release error, and reports the first release error. Cancellation, process death, or an ambiguous claim/release I/O failure may leave a claim behind. Read-only recovery remains available when the snapshot/source proof is valid, but further staging is blocked until a separate storage-fenced recovery procedure is implemented. Process liveness and the legacy ingress lease do not authorize taking over or deleting a claim.
+
+Run the focused fixtures with a nonzero test count:
+
+```sh
+cargo test -p rustfs-heal --lib heal::mrf_queue::snapshot::migration::tests
+```
+
+Fixtures use real local disks and the production CAS/readback path. They cover disk-order independence, retained raw identities, source change after candidate write, capacity rejection, interrupted commit boundaries, lost responses, torn inactive slots, and refusal to take over an interrupted claim. Boundary injection and same-process reopen are not process-kill, directory-fsync failure, disk-full, mixed-version, or power-loss tests. The actual manager Full/Accepted-to-crash pipeline remains outside this staged API.
+
+Rollback leaves all pending and legacy artifacts untouched. Activation still requires legacy-writer coordination, bounded recoverable ingress, exact object-disposition/successor receipts, and the W14/W21 process-crash and compatibility gates. The production legacy replay deletion window remains unresolved by this staging-only phase.