test(ecstore): account for upstream recovery test additions

* perf(ecstore): avoid head-of-line blocking in delete recovery

Recover independent keys through a bounded rolling window while preserving per-key hint generations, admission budgets, cancellation retry, and the existing transaction decision engine.

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

* perf(ecstore): bound durable delete preparation and bootstrap recovery

Prepare independent Linux delete transactions with bounded concurrency while retaining per-key result order and serial fallback for overlapping, rollback, and legacy requests. Preserve startup discovery until bucket metadata is initialized, and add release performance and process-crash acceptance runners.

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

* test(ecstore): verify clean exits in recovery acceptance

Reject forced or OOM exits when measuring clean startup and verify original ETags across mixed-layout crash recovery and a second restart.

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

* test(ecstore): distinguish prepared and decided batch crashes

Observe real preparation and receipt states separately, retain non-vacuous acknowledged-delete sentinels, and require read-only physical convergence before S3 reads can trigger cleanup.

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

* test(ecstore): move final delete recovery options

Move the options value into the batch-ordering regression's final case
instead of cloning it. Keep the earlier clone needed for reuse.

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

* fix(ecstore): replay delete receipts after interrupted proof GC

Retire only the exact empty transaction leaf before journal activation. Preserve occupied directories and prefix children, and cover partial cross-disk proof GC without a fresh delete. Tighten crash evidence to distinguish directory-only remnants from retained content.

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

---------

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

.config/test-count-baseline.txt

@@ -20,7 +20,7 @@ rustfs-connect-symbol-catalog	6	0
 rustfs-credentials	24	0
 rustfs-crypto	67	0
 rustfs-data-usage	55	0
-rustfs-ecstore	7060	1
+rustfs-ecstore	7065	1
 rustfs-extension-schema	23	0
 rustfs-filemeta	355	0
 rustfs-heal	976	3

crates/ecstore/src/disk/local.rs

@@ -115,6 +115,31 @@ use uuid::Uuid;
 // Bound outstanding filesystem jobs and metadata buffers per disk request.
 const PART_METADATA_READ_CONCURRENCY: usize = 8;
 
+// Keep independent forward transactions bounded without reordering aliases,
+// parent/child keys, rollback, or legacy operations.
+const DELETE_VERSION_PREPARE_CONCURRENCY: usize = 8;
+
+fn delete_version_batch_concurrency(versions: &[FileInfoVersions], opts: &DeleteOptions) -> usize {
+    if !cfg!(target_os = "linux") || !opts.recoverable_delete || opts.undo_write || opts.undo_delete || versions.len() < 2 {
+        return 1;
+    }
+    let mut names = HashSet::with_capacity(versions.len());
+    for version in versions {
+        let name = version.name.as_str();
+        if name.split('/').any(|segment| matches!(segment, "" | "." | "..")) || !names.insert(name) {
+            return 1;
+        }
+    }
+    for name in &names {
+        for (separator, _) in name.match_indices('/') {
+            if names.contains(&name[..separator]) {
+                return 1;
+            }
+        }
+    }
+    DELETE_VERSION_PREPARE_CONCURRENCY
+}
+
 /// The fresh health path uses a synchronous capacity syscall. Keep the permit
 /// inside the blocking closure so cancelling its async caller cannot admit a
 /// second probe while the first syscall is still running.
@@ -7125,23 +7150,29 @@ impl LocalDisk {
         opts: DeleteOptions,
         namespace_owner: Option<Arc<dyn Send + Sync>>,
     ) -> Vec<Option<Error>> {
-        let mut errs = Vec::with_capacity(versions.len());
-        for _ in 0..versions.len() {
-            errs.push(None);
+        let concurrency = delete_version_batch_concurrency(&versions, &opts);
+        let mut errors = vec![None; versions.len()];
+        let mut pending = stream::iter(
+            versions
+                .into_iter()
+                .enumerate()
+                .map(|(index, version): (usize, FileInfoVersions)| {
+                    let owner = namespace_owner.clone();
+                    let options = &opts;
+                    async move {
+                        let error = self
+                            .delete_versions_internal(volume, &version.name, &version.versions, options, owner)
+                            .await
+                            .err();
+                        (index, error)
+                    }
+                }),
+        )
+        .buffer_unordered(concurrency);
+        while let Some((index, error)) = pending.next().await {
+            errors[index] = error;
         }
-
-        for (i, ver) in versions.iter().enumerate() {
-            if let Err(e) = self
-                .delete_versions_internal(volume, ver.name.as_str(), &ver.versions, &opts, namespace_owner.clone())
-                .await
-            {
-                errs[i] = Some(e);
-            } else {
-                errs[i] = None;
-            }
-        }
-
-        errs
+        errors
     }
 
     pub(crate) async fn delete_versions_guarded_with_namespace_owner(

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -1262,6 +1262,17 @@ impl LocalDisk {
         self.delete_recovery_barrier(volume, object, transaction, "PREACTIVATION")
             .await?;
         match fs::symlink_metadata(&destination).await {
+            Ok(metadata) if metadata.is_dir() => {
+                // Proof GC may persist before its final rmdir. Only an empty
+                // leaf can be retired; nested prefixes and new evidence fail
+                // closed under the same namespace ownership as publication.
+                match os::remove_dir_with_owner(&destination, owner.clone()).await {
+                    Ok(()) => {}
+                    Err(err) if err.kind() == ErrorKind::NotFound => {}
+                    Err(err) if is_dir_not_empty_error(&err) => return Err(DiskError::OutdatedXLMeta),
+                    Err(err) => return Err(to_file_error(err).into()),
+                }
+            }
             Ok(_) => return Err(DiskError::OutdatedXLMeta),
             Err(err) if err.kind() == ErrorKind::NotFound => {}
             Err(err) => return Err(to_file_error(err).into()),
@@ -2278,6 +2289,131 @@ mod tests {
         (temp, disk, info, bytes)
     }
 
+    #[test]
+    fn delete_recovery_batch_parallelism_preserves_overlapping_key_order() {
+        let opts = DeleteOptions {
+            recoverable_delete: true,
+            ..Default::default()
+        };
+        for names in [
+            vec!["a", "a"],
+            vec!["a", "a-0", "a/child"],
+            vec!["a//b", "a/b"],
+            vec!["a/./b", "a/b"],
+            vec!["a/../b", "b"],
+        ] {
+            let groups: Vec<_> = names
+                .into_iter()
+                .map(|name| super::super::FileInfoVersions {
+                    name: name.to_owned(),
+                    ..Default::default()
+                })
+                .collect();
+            assert_eq!(
+                super::super::delete_version_batch_concurrency(&groups, &opts),
+                1,
+                "overlapping or noncanonical keys retain serial execution"
+            );
+        }
+        let groups: Vec<_> = ["a/one", "a/two"]
+            .into_iter()
+            .map(|name| super::super::FileInfoVersions {
+                name: name.to_owned(),
+                ..Default::default()
+            })
+            .collect();
+        let expected = if cfg!(target_os = "linux") {
+            super::super::DELETE_VERSION_PREPARE_CONCURRENCY
+        } else {
+            1
+        };
+        assert_eq!(super::super::delete_version_batch_concurrency(&groups, &opts), expected);
+        for options in [
+            DeleteOptions::default(),
+            DeleteOptions {
+                undo_write: true,
+                ..opts.clone()
+            },
+            DeleteOptions {
+                undo_delete: true,
+                ..opts
+            },
+        ] {
+            assert_eq!(
+                super::super::delete_version_batch_concurrency(&groups, &options),
+                1,
+                "legacy and rollback ordering is unchanged"
+            );
+        }
+    }
+
+    #[cfg(target_os = "linux")]
+    #[tokio::test]
+    async fn delete_recovery_batch_prepares_independent_keys_while_one_metadata_lease_is_held() {
+        let (temp, disk, info, _) = fixture().await;
+        let disk = Arc::new(disk);
+        let mut healthy = info.clone();
+        healthy.name = "healthy".to_owned();
+        let healthy_root = temp.path().join("recovery/healthy");
+        fs::create_dir_all(&healthy_root).await.expect("healthy object directory");
+        fs::write(healthy_root.join(STORAGE_FORMAT_FILE), test_meta(healthy))
+            .await
+            .expect("healthy source");
+        let corrupt_root = temp.path().join("recovery/corrupt");
+        fs::create_dir_all(&corrupt_root).await.expect("corrupt object directory");
+        fs::write(corrupt_root.join(STORAGE_FORMAT_FILE), b"invalid xl.meta")
+            .await
+            .expect("corrupt source");
+        let held = os::acquire_metadata_mutation_lease(&temp.path().join("recovery/object"), None).await;
+        let groups = ["object", "corrupt", "healthy"]
+            .into_iter()
+            .map(|name| super::super::FileInfoVersions {
+                name: name.to_owned(),
+                versions: vec![FileInfo {
+                    name: name.to_owned(),
+                    ..Default::default()
+                }],
+                ..Default::default()
+            })
+            .collect();
+        let worker_disk = disk.clone();
+        let worker = tokio::spawn(async move {
+            worker_disk
+                .delete_versions_with_namespace_owner(
+                    "recovery",
+                    groups,
+                    DeleteOptions {
+                        recoverable_delete: true,
+                        old_data_dir: Some(Uuid::new_v4()),
+                        ..Default::default()
+                    },
+                    None,
+                )
+                .await
+        });
+        let prepared = tokio::time::timeout(Duration::from_secs(5), async {
+            while fs::try_exists(healthy_root.join(STORAGE_FORMAT_FILE))
+                .await
+                .expect("inspect healthy metadata")
+            {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await;
+        assert!(temp.path().join("recovery/object/xl.meta").exists(), "the locked source must stay intact");
+        drop(held);
+        let results = worker.await.expect("complete every disk result");
+        assert_eq!(
+            results,
+            vec![None, Some(DiskError::FileCorrupt), None],
+            "completion order cannot change per-key error positions"
+        );
+        assert!(
+            prepared.is_ok(),
+            "one key's metadata lease cannot serialize unrelated transaction preparation"
+        );
+    }
+
     #[tokio::test]
     async fn delete_recovery_absent_publish_refreshes_prepared_witness() {
         let (_temp, disk, _info, _original) = fixture().await;
@@ -4062,6 +4198,51 @@ mod tests {
         assert!(witness.original_digest.is_none() && witness.target_digest.is_none() && witness.data_dirs.is_empty());
     }
 
+    #[tokio::test]
+    async fn delete_recovery_activation_preserves_occupied_transaction_directory() {
+        let (_source_temp, source, _, _) = fixture().await;
+        let transaction = Uuid::new_v4();
+        let applied = apply(&source, transaction).await;
+        let mut witness = source
+            .settle_delete_transaction(
+                "recovery",
+                "object",
+                transaction,
+                applied.fingerprint,
+                DeleteRecoveryAction::PublishCommit,
+            )
+            .await
+            .unwrap()
+            .journal
+            .unwrap();
+        witness.original_digest = None;
+        witness.target_digest = None;
+        witness.data_dirs.clear();
+        for nested in [false, true] {
+            let (_peer_temp, peer, _, _) = fixture().await;
+            let root = peer.io_get_object_path("recovery", "object").unwrap();
+            fs::remove_file(root.join(STORAGE_FORMAT_FILE)).await.unwrap();
+            let retained = root.join(transaction.to_string());
+            fs::create_dir(&retained).await.unwrap();
+            if nested {
+                fs::create_dir(retained.join("prefix-child")).await.unwrap();
+            } else {
+                fs::write(retained.join("unknown-file"), b"retained").await.unwrap();
+            }
+            assert!(matches!(
+                peer.activate_delete_preparation(&witness, None, None, None).await,
+                Err(DiskError::OutdatedXLMeta)
+            ));
+            assert!(!retained.join(PREPARED).exists());
+            if nested {
+                assert!(retained.join("prefix-child").is_dir());
+            } else {
+                assert_eq!(fs::read(retained.join("unknown-file")).await.unwrap(), b"retained");
+            }
+            assert!(!root.join(STORAGE_FORMAT_FILE).exists());
+        }
+    }
+
     #[tokio::test]
     async fn delete_recovery_raw_discovery_keeps_malformed_records_and_later_keys() {
         let (temp, disk, _, _) = fixture().await;

crates/ecstore/src/set_disk/ops/delete_recovery.rs

@@ -1783,6 +1783,56 @@ mod tests {
             .expect("returning old replica must not leave bucket residue");
     }
 
+    #[tokio::test]
+    async fn delete_recovery_set_replays_partial_gc_with_empty_transaction_directory() {
+        let bucket = "recover-empty-gc";
+        let (dirs, store, disks) = fixture(bucket).await;
+        put(&store, bucket, &[24; 4096], false).await;
+        let transaction = applied(&store, &disks, bucket, None, disks.len()).await;
+        for (index, disk) in disks.iter().enumerate() {
+            publish(disk, bucket, transaction).await;
+            let state = disk.inspect_delete_transaction(bucket, "object", transaction).await.unwrap();
+            let finished = disk
+                .settle_delete_transaction(bucket, "object", transaction, state.fingerprint, DeleteRecoveryAction::FinishCommit)
+                .await
+                .unwrap();
+            if index > 0 {
+                disk.settle_delete_transaction(
+                    bucket,
+                    "object",
+                    transaction,
+                    finished.fingerprint,
+                    DeleteRecoveryAction::FinalizeCommit,
+                )
+                .await
+                .unwrap();
+            }
+        }
+        // A crash can persist proof removal before the final empty rmdir.
+        let empty = dirs[3].path().join(bucket).join("object").join(transaction.to_string());
+        tokio::fs::create_dir_all(&empty).await.unwrap();
+        for _ in 0..2 {
+            store
+                .recover_delete_transactions(bucket, &HealOpts::default())
+                .await
+                .expect("the remaining receipt must settle every participant without a fresh delete");
+            for disk in &disks {
+                assert!(discover(disk, bucket, Some("object")).await.unwrap().is_empty());
+                assert!(
+                    disk.inspect_delete_transaction(bucket, "object", transaction)
+                        .await
+                        .unwrap()
+                        .current_digest
+                        .is_none()
+                );
+            }
+        }
+        store
+            .delete_bucket(bucket, &DeleteBucketOptions::default())
+            .await
+            .expect("ordinary bucket removal must succeed after replay");
+    }
+
     #[tokio::test]
     async fn delete_recovery_tracks_only_observed_removed_version() {
         let bucket = "recover-actual-removal";

crates/ecstore/src/store/delete_recovery_runtime.rs

@@ -525,6 +525,12 @@ impl RecoveryPass {
     }
 
     async fn discover(&mut self, store: &Arc<ECStore>, cancel: &CancellationToken) {
+        // ECStore starts before the bucket metadata system is published. Do
+        // not turn that bootstrap state into a deferred disk/bucket failure.
+        if store.ctx.bucket_metadata_sys().is_none() {
+            return;
+        }
+
         let hints = store.ctx.delete_recovery_hints();
         // Renew only this worker's bounded local cursors before any await.
         // Backpressure and a slow canonical rollback must not erase progress.
@@ -1032,6 +1038,26 @@ mod tests {
         });
     }
 
+    #[tokio::test]
+    async fn delete_recovery_waits_for_metadata_before_consuming_startup_discovery() {
+        let (_dirs, store) = isolated_store_over_temp_disks().await;
+        store.background_cancel_token().expect("background token").cancel();
+        tokio::task::yield_now().await;
+        assert!(store.ctx.bucket_metadata_sys().is_none(), "metadata bootstrap has not completed");
+        let hints = store.ctx.delete_recovery_hints();
+        hints.request_rescan();
+        let generation = hints.rescan().expect("startup discovery obligation");
+        let mut pass = RecoveryPass::default();
+        pass.discover(&store, &CancellationToken::new()).await;
+        assert_eq!(hints.rescan(), Some(generation), "bootstrap cannot consume the discovery obligation");
+        assert!(pass.deferred.is_empty(), "bootstrap is not a failed disk scope");
+        assert!(pass.next_inventory.is_none(), "bootstrap must not delay the initial inventory");
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        pass.discover(&store, &CancellationToken::new()).await;
+        assert!(hints.rescan().is_none(), "the first ready pass fulfills the empty-store scan immediately");
+        assert!(pass.deferred.is_empty());
+    }
+
     #[tokio::test]
     async fn delete_recovery_clean_startup_discovers_hidden_preparation() {
         let bucket = "runtime-clean-start";

docs/operations/delete-transaction-recovery.md

@@ -72,7 +72,9 @@ completed or canceled probes do not consume shared discovery cursors.
 ## Entry points
 
 Every startup, including a clean restart, starts incremental raw discovery of
-local transaction records. During normal operation, deletion registers a
+local transaction records once the instance's bucket metadata system has been
+published. Bootstrap is not classified as a failed disk scope and cannot
+consume the initial discovery obligation. During normal operation, deletion registers a
 bounded, coalesced exact-key retry hint before preparing metadata. Synchronous
 completion clears only its own hint generation; an uncertain publication or
 unfinished cleanup remains eligible for automatic replay without another
@@ -216,7 +218,21 @@ curl with `--image`, and a unique Docker namespace. The optional
 `--baseline-binary` enables the mixed-version case. The harness uses synthetic
 S3 objects, localhost ports, labeled isolated volumes, and a non-root server.
 It stops only its own containers and keeps JSON results and logs in the chosen
-artifact directory. Production builds must not enable `e2e-test-hooks`.
+artifact directory. Production builds must not enable `e2e-test-hooks`. The `release-batch-crash`
+case accepts a normal release binary: it observes persisted preparation before
+SIGKILL in one round and persisted commit receipts in another, verifies automatic
+mixed inline/external recovery without a new DELETE
+or administrator request, and checks that a second clean restart is stable.
+This case does not pin an individual instruction or transaction phase.
+
+Physical settlement permits directory-only remnants that ordinary non-force
+bucket removal safely prunes. It rejects retained payloads, backup or transaction
+records, unknown files, and symbolic links. Final cleanup deletes only surviving
+objects before requiring ordinary bucket removal. An interrupted proof GC can
+also leave an empty transaction directory beside another drive's receipt.
+Recovery retires only that exact empty leaf before republishing a replay
+witness; a nonempty directory, including an empty nested prefix, remains
+protected. No recursive removal grants transaction authority.
 
 Run the preactivation/prepare/apply/decision process-stop cases, interrupted restore,
 returning and untouched peers, both generation orderings, version/retention
@@ -224,3 +240,52 @@ and prefix protection, administrator dry-run/removal, response-loss,
 permission/corruption, and mixed-version cases. Preserve the source identity
 of both binaries with the results. A passing process-stop matrix does not
 cover hardware power cuts, Windows runtime behavior, or DELETE throughput.
+
+
+## Performance acceptance
+
+Linux forward batch preparation admits at most eight independent keys per
+local disk request. Duplicate keys, ancestor/descendant keys, noncanonical
+paths, legacy operations, and rollback keep their ordered execution. Other
+platforms retain serial preparation. The existing per-key metadata lease,
+transaction publication, namespace ownership, and strict synchronization
+barriers still apply to every admitted operation.
+
+Use `scripts/test/delete_recovery_perf.py` for resource-matched release
+comparisons. Supply separate binaries and build receipts; each receipt binds
+`binary_sha256`, `source_head`, and the build fields `profile`, `opt_level`,
+`target`, `rustc`, `features`, `default_features`, `lto`, and `codegen_units`.
+Receipts must describe matching release builds at optimization level 3 without
+`e2e-test-hooks`. Preserve any source patch and the build command with them.
+For example:
+
+```sh
+python3 scripts/test/delete_recovery_perf.py \
+  --baseline-binary ./baseline/rustfs --baseline-receipt ./baseline/build.json \
+  --candidate-binary ./candidate/rustfs --candidate-receipt ./candidate/build.json \
+  --image rust:1.99.1-trixie --artifact-dir ./delete-performance-results \
+  --cpuset-cpus 2-5 --memory 4g
+```
+
+The runner uses an A1/B1/B2/A2 sequence with at least three repetitions per
+cohort. It checks original bytes and ETags, physical inline/external layouts,
+empty listings, physical cleanup, and bucket removal. Each sample also cleanly
+restarts its isolated server and waits for the bounded raw discovery page on
+every local disk. Startup observations retain whether discovery exceeded
+thirty seconds, including for historical baselines whose bootstrap race enters
+the thirty-second retry interval; the observation deadline is ninety seconds.
+A candidate exceeding the thirty-second budget fails acceptance even when the
+longer observation eventually completes.
+This separates server readiness and local discovery service time from the
+timed DELETE. DELETE begins no earlier than thirty-five seconds after restart
+in either variant, keeping process age and early maintenance phases comparable.
+Normal scanner/heal loops are disabled for this
+controlled attribution; automatic transaction recovery remains enabled.
+
+Cohorts share the same image, CPU set, memory limit, strict durability policy,
+and workload. A baseline or candidate drift above ten percent makes the result
+inconclusive and withholds improvement percentages. The artifact retains raw
+samples and per-operation cumulative disk service times. These concurrent
+service times are not an additive wall-clock decomposition. Four logical
+volumes on one Docker host filesystem do not represent four independent
+physical devices or establish hardware power-loss durability.

scripts/test/delete_recovery_docker.py

@@ -19,7 +19,7 @@ Logical disks share the Docker host filesystem; this tests process/RPC failure
 and namespace safety, not independent hardware failure or power-loss durability.
 """
 from pathlib import Path
-import base64, hashlib, ipaddress, json, os, subprocess, tempfile, threading, time, uuid, xml.etree.ElementTree as ET
+import base64, hashlib, inspect, ipaddress, json, os, subprocess, tempfile, threading, time, uuid, xml.etree.ElementTree as ET
 from urllib.parse import quote
 ROOT = None
 PREFIX = None
@@ -36,6 +36,12 @@ def run(*args, check=True, **kwargs):
 def docker(*args, check=True):
     return run('docker', *args, check=check)
 
+def has_retained_content(root):
+    """Empty directory trees are pruned by non-force bucket removal."""
+    if root.is_symlink() or (root.exists() and not root.is_dir()):
+        return True
+    return any(path.is_symlink() or not path.is_dir() for path in root.rglob('*'))
+
 class S3:
 
     def __init__(self, port):
@@ -223,6 +229,12 @@ class Cluster:
     def kill(self, index=0):
         docker('kill', '--signal', 'KILL', self.names[index])
 
+    def stop_clean(self, index=0, timeout=60):
+        docker('stop', '--time', str(timeout), self.names[index])
+        state = json.loads(docker('inspect', '--format', '{{json .State}}', self.names[index]).stdout)
+        assert state['ExitCode'] == 0 and not state['OOMKilled'], state
+        return state
+
     def restart(self, index=None):
         for i in range(self.nodes) if index is None else [index]:
             docker('start', self.names[i])
@@ -1035,9 +1047,7 @@ def automatic_clean_restart():
         clear_barriers(cluster)
         before = residue_inventory(cluster, bucket, key)
         assert all(any(path.endswith('delete-quorum-committed') for path in drive['files']) for node in before for drive in node)
-        docker('stop', '--time', '60', cluster.names[0])
-        state = json.loads(docker('inspect', '--format', '{{json .State}}', cluster.names[0]).stdout)
-        assert state['ExitCode'] == 0 and not state['OOMKilled'], state
+        state = cluster.stop_clean()
         repair = "from pathlib import Path; import json,os; roots=list(Path('/data').glob('d*/%s/%s')); dirs=[p for root in roots for p in root.iterdir() if p.is_dir() and (p/'part.1').is_file()]; assert len(dirs)==4; [os.chmod(p,0o755) for p in dirs]; print(json.dumps({'restored_payload_dirs':len(dirs)}))" % (bucket, key)
         cluster.inspect_data(repair)
         cluster.restart()
@@ -1248,6 +1258,135 @@ def admin_smoke():
     finally:
         cluster.remove()
 
+def release_batch_crash():
+    """Observe real journals before SIGKILL without requiring test-only hooks."""
+    import random
+    rounds = []
+    for iteration, phase in enumerate(('prepared', 'receipts')):
+        cluster = Cluster(f'release-batch-crash-{iteration}', drives=4)
+        try:
+            cluster.create()
+            client = cluster.clients[0]
+            bucket = 'release-batch-crash'
+            small = random.Random(2849).randbytes(4096)
+            large = random.Random(2850).randbytes(1024 * 1024)
+            payloads = {f'inline/object-{index:04d}': small for index in range(256)}
+            payloads.update({f'external/object-{index:04d}': large for index in range(32)})
+            confirmed = {'confirmed/inline': small, 'confirmed/external': large}
+            client.put_bucket(bucket)
+            etags = {}
+            for key, payload in {**payloads, **confirmed}.items():
+                client.put(bucket, key, payload)
+                etags[key] = client.last_headers['etag']
+            for key in confirmed:
+                client.delete(bucket, key)
+                wait_inventory(cluster, bucket, key, lambda inventory: all(not drive['exists'] for node in inventory for drive in node))
+            request = {}
+
+            def delete_batch():
+                try:
+                    xml = ET.Element('Delete')
+                    for key in payloads:
+                        ET.SubElement(ET.SubElement(xml, 'Object'), 'Key').text = key
+                    body = ET.tostring(xml)
+                    request['response'] = client.request('POST', bucket + '?delete', body, {
+                        'Content-Type': 'application/xml',
+                        'Content-MD5': base64.b64encode(hashlib.md5(body).digest()).decode(),
+                    }, expected=200)
+                except Exception as error:
+                    request['exception'] = str(error)
+
+            counts = (
+                "from pathlib import Path\nimport json\n"
+                f"bucket={bucket!r}\n"
+                "print(json.dumps([{'prepared': len(list((d/bucket).rglob('delete-prepared'))), "
+                "'receipts': len(list((d/bucket).rglob('delete-quorum-committed')))} "
+                "for d in Path('/data').glob('d*')]))"
+            )
+            thread = threading.Thread(target=delete_batch, daemon=True)
+            thread.start()
+            deadline = time.monotonic() + 30
+            while True:
+                observation = cluster.inspect_data(counts)
+                if any(drive[phase] for node in observation for drive in node):
+                    break
+                if not thread.is_alive() or time.monotonic() >= deadline:
+                    raise AssertionError('the selected durable evidence must be observed in a live batch before the crash')
+                time.sleep(0.02)
+            assert thread.is_alive(), 'the observed batch must still be active before SIGKILL'
+            (ROOT / f'release-batch-crash-{iteration}-observation.json').write_text(json.dumps({
+                'observed_phase': phase, 'observed_before_kill': observation,
+                'request_active_before_kill': True,
+            }, indent=2))
+            cluster.kill(0)
+            thread.join(10)
+            assert not thread.is_alive(), 'the interrupted client must finish'
+            acknowledged = set(confirmed)
+            if 'response' in request:
+                _, body = request['response']
+                for item in ET.fromstring(body):
+                    if item.tag.rsplit('}', 1)[-1] == 'Deleted':
+                        acknowledged.update(child.text for child in item if child.tag.rsplit('}', 1)[-1] == 'Key')
+            physical_probe = (
+                "from pathlib import Path\nimport json\n" + inspect.getsource(has_retained_content) +
+                f"bucket={bucket!r};keys={list(payloads) + list(confirmed)!r}\n"
+                "record_names={'xl.meta.bkp','delete-prepared','delete-target','delete-quorum-committed','delete-commit-finished','xl.meta.delete-marker.rollback'}\n"
+                "result=[]\nfor d in Path('/data').glob('d*'):\n"
+                " root=d/bucket\n"
+                " result.append({'prepared':len(list(root.rglob('delete-prepared'))),"
+                "'receipts':len(list(root.rglob('delete-quorum-committed'))),"
+                "'records':sum(p.name in record_names or p.name.startswith(('reserve-delete-data.','recovery-delete-data.','delete-data.')) for p in root.rglob('*')),"
+                "'live':[key for key in keys if (root/key/'xl.meta').is_file()],"
+                "'residue':[key for key in keys if not (root/key/'xl.meta').is_file() and has_retained_content(root/key)]})\n"
+                "print(json.dumps(result))"
+            )
+
+            def wait_for_physical_settlement():
+                deadline = time.monotonic() + 90
+                while True:
+                    drives = [drive for node in cluster.inspect_data(physical_probe) for drive in node]
+                    if all(not drive['records'] and not drive['residue'] for drive in drives):
+                        live = set(drives[0]['live'])
+                        if all(set(drive['live']) == live for drive in drives):
+                            return live
+                    if time.monotonic() >= deadline:
+                        raise TimeoutError('automatic recovery must converge physically before any new S3 DELETE, LIST, GET, or admin request')
+                    time.sleep(0.05)
+
+            cluster.restart()
+            # Physical inspection precedes S3 listing, which can itself purge
+            # eligible residue and must not manufacture recovery evidence here.
+            present = wait_for_physical_settlement()
+            assert present <= payloads.keys()
+            assert not (present & acknowledged), 'acknowledged deletes cannot resurrect'
+            assert set(client.list(bucket)) == present
+            for key, payload in {**payloads, **confirmed}.items():
+                status, body = client.request('GET', bucket + '/' + key, expected=(200, 404))
+                assert (status == 200) == (key in present)
+                if status == 200:
+                    assert body == payload, 'every surviving key must preserve complete original bytes'
+                    assert client.last_headers.get('etag') == etags[key], 'rollback must preserve the original ETag'
+            cluster.stop_clean()
+            cluster.restart()
+            assert wait_for_physical_settlement() == present, 'a second recovery cannot change the settled key set'
+            assert set(client.list(bucket)) == present
+            for key in present:
+                assert client.get(bucket, key) == payloads[key], 'a second recovery must preserve complete bytes'
+                assert client.last_headers.get('etag') == etags[key], 'a second recovery must preserve the original ETag'
+            if present:
+                client.batch_delete(bucket, sorted(present))
+            assert client.list(bucket) == []
+            client.delete_bucket(bucket)
+            rounds.append({'observed_phase': phase, 'observed_before_kill': observation,
+                           'acknowledged': len(acknowledged), 'present': len(present),
+                           'absent': len(payloads) + len(confirmed) - len(present),
+                           'complete_bytes_and_etags_verified': len(present), 'second_restart_stable': True})
+            (ROOT / 'release-batch-crash-progress.json').write_text(json.dumps(rounds, indent=2))
+        finally:
+            cluster.remove()
+    return {'case': 'release-batch-crash', 'status': 'passed', 'rounds': rounds,
+            'fault_model': 'SIGKILL after observing durable preparation or receipt; no deterministic instruction barrier'}
+
 def large_batch_delete():
     cluster = Cluster('batch-1000', nodes=4, drives=2)
     try:
@@ -1358,7 +1497,7 @@ def main():
     global ROOT, PREFIX, IMAGE, PORT, CANDIDATE, BASELINE
     import argparse, re, sys
     parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
-    parser.add_argument('--binary', required=True, type=Path, help='Linux candidate binary built with e2e-test-hooks')
+    parser.add_argument('--binary', required=True, type=Path, help='Linux candidate; deterministic barrier cases require e2e-test-hooks')
     parser.add_argument('--baseline-binary', type=Path, help='Older Linux binary, required by mixed case')
     parser.add_argument('--image', required=True, help='Local Linux image with Python 3 and the binary runtime libraries')
     parser.add_argument('--artifact-dir', required=True, type=Path, help='New directory for logs and stage handshakes')
@@ -1385,7 +1524,7 @@ def main():
         existing = docker(resource, 'ls', '-q', '--filter', 'label=rustfs.delete-recovery-test=' + PREFIX).stdout.strip()
         if existing:
             parser.error('namespace already owns Docker resources; choose a new namespace')
-    cases = {'automatic-clean-restart': automatic_clean_restart, 'unknown-fresh': unknown_predecessor_fresh_delete, 'batch1000': large_batch_delete, 'versions3000': large_version_history, 'admin-smoke': admin_smoke, 'legacy-alias': legacy_transaction_data_dir_alias, 'missing-offline': ordinary_missing_with_offline_peer, 'inverse': retained_decision_newer_undecided, 'deep': deep_key_crash, 'deep-large-stack': lambda : deep_key_crash(True), 'admin-prefix': admin_prefix_scoped, 'response-loss': receipt_response_loss, 'admin-object': lambda : admin_legacy_dry_run(True), 'smoke': smoke, 'legacy1': lambda : legacy_retry(1), 'legacy4': lambda : legacy_retry(4), 'restore': rollback_crash, 'retention': protected_version, 'version-applied': lambda : versioned_crash('APPLIED'), 'version-decision': lambda : versioned_crash('DECISION'), 'returning': returning_drive_new_generation, 'untouched': untouched_returning_peer, 'missing-version': missing_version_returning_peer, 'mixed': mixed_versions, 'admin': admin_legacy_dry_run, 'prefix': uuid_prefix_crash, 'corruption': corruption_repair, 'permission': permission_fault, 'error-prepared': lambda : injected_error('PREPARED'), 'error-applied': lambda : injected_error('APPLIED'), 'error-decision': lambda : injected_error('DECISION')}
+    cases = {'release-batch-crash': release_batch_crash, 'automatic-clean-restart': automatic_clean_restart, 'unknown-fresh': unknown_predecessor_fresh_delete, 'batch1000': large_batch_delete, 'versions3000': large_version_history, 'admin-smoke': admin_smoke, 'legacy-alias': legacy_transaction_data_dir_alias, 'missing-offline': ordinary_missing_with_offline_peer, 'inverse': retained_decision_newer_undecided, 'deep': deep_key_crash, 'deep-large-stack': lambda : deep_key_crash(True), 'admin-prefix': admin_prefix_scoped, 'response-loss': receipt_response_loss, 'admin-object': lambda : admin_legacy_dry_run(True), 'smoke': smoke, 'legacy1': lambda : legacy_retry(1), 'legacy4': lambda : legacy_retry(4), 'restore': rollback_crash, 'retention': protected_version, 'version-applied': lambda : versioned_crash('APPLIED'), 'version-decision': lambda : versioned_crash('DECISION'), 'returning': returning_drive_new_generation, 'untouched': untouched_returning_peer, 'missing-version': missing_version_returning_peer, 'mixed': mixed_versions, 'admin': admin_legacy_dry_run, 'prefix': uuid_prefix_crash, 'corruption': corruption_repair, 'permission': permission_fault, 'error-prepared': lambda : injected_error('PREPARED'), 'error-applied': lambda : injected_error('APPLIED'), 'error-decision': lambda : injected_error('DECISION')}
     cases.update({'multipart-sidecars': completed_multipart_sidecars, 'multipart-integrity': lambda: completed_multipart_sidecars(True), 'multipart-sidecars-applied': lambda: completed_multipart_sidecars(False, True), 'multipart-integrity-applied': lambda: completed_multipart_sidecars(True, True)})
     for stage in ('PREPARED', 'APPLIED', 'DECISION'):
         cases['marker-batch-' + stage.lower()] = lambda stage=stage: marker_batch_crash(stage)

scripts/test/delete_recovery_perf.py

@@ -0,0 +1,261 @@
+#!/usr/bin/env python3
+"""Compare strict DELETE stages using isolated, resource-matched ABBA cohorts.
+
+Both binaries require a JSON receipt with binary_sha256, source_head, and an
+identical build object containing profile=release, opt_level=3, target, rustc,
+features, default_features, lto, and codegen_units. Disk operation service times
+include concurrent work; they are not an additive wall-clock decomposition.
+"""
+
+import argparse
+import hashlib
+import inspect
+import json
+from pathlib import Path
+import random
+import re
+import statistics
+import sys
+import time
+import uuid
+
+import delete_recovery_docker as harness
+
+
+def binary_digest(binary):
+    digest = hashlib.sha256()
+    with binary.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def binary_receipt(binary, receipt):
+    value = json.loads(receipt.read_text())
+    if value.get("binary_sha256") != binary_digest(binary):
+        raise ValueError(f"binary does not match its receipt: {binary}")
+    if not re.fullmatch(r"[0-9a-f]{40}", value.get("source_head", "")):
+        raise ValueError("receipt requires an exact source_head")
+    build = value["build"]
+    required = {"profile", "opt_level", "target", "rustc", "features", "default_features", "lto", "codegen_units"}
+    if not required <= build.keys() or build["profile"] != "release" or build["opt_level"] != 3:
+        raise ValueError("receipts must describe optimized release builds")
+    if not isinstance(build["features"], list) or "e2e-test-hooks" in build["features"]:
+        raise ValueError("performance receipts must exclude test-only hooks")
+    return value
+
+
+def operation_snapshot(client):
+    _, body = client.request("GET", "rustfs/admin/v3/storageinfo", expected=200)
+    found = []
+
+    def visit(value):
+        if isinstance(value, dict):
+            if "api_calls" in value and "last_minute" in value:
+                found.append(value)
+            else:
+                for item in value.values():
+                    visit(item)
+        elif isinstance(value, list):
+            for item in value:
+                visit(item)
+
+    visit(json.loads(body))
+    if len(found) != 4:
+        raise AssertionError("the stage probe requires four complete local disk metrics")
+    result = {}
+    for metric in found:
+        for operation, sample in metric["last_minute"].items():
+            item = result.setdefault(operation, {"count": 0, "ns": 0})
+            item["count"] += sample["count"]
+            item["ns"] += sample["acc_time_ns"]
+    return result, [metric["last_minute"].get("discover_delete_transactions", {}).get("count", 0) for metric in found]
+
+
+def stage_delta(before, after):
+    result = {}
+    for operation in before.keys() | after.keys():
+        old = before.get(operation, {"count": 0, "ns": 0})
+        new = after.get(operation, {"count": 0, "ns": 0})
+        count, elapsed = new["count"] - old["count"], new["ns"] - old["ns"]
+        if count < 0 or elapsed < 0:
+            raise ValueError("the rolling metric window expired; stage attribution is invalid")
+        if count:
+            result[operation] = {"calls": count, "service_ms": elapsed / 1_000_000}
+    return result
+
+
+def drift_percent(first, second):
+    if min(first, second) <= 0:
+        raise ValueError("benchmark durations must be positive")
+    return abs(first - second) / min(first, second) * 100
+
+
+def measure(cluster, case, count, size, batch, repeat):
+    client = cluster.clients[0]
+    bucket = f"delete-perf-{case}-{repeat}"
+    payload = random.Random(2849 + size).randbytes(size)
+    keys = [f"object-{index:04d}" for index in range(count)]
+    client.put_bucket(bucket)
+    for key in keys:
+        client.put(bucket, key, payload)
+        etag = client.last_headers.get("etag")
+        assert etag, "PUT must acknowledge the object's ETag"
+        assert client.get(bucket, key) == payload, "complete bytes must match before deletion"
+        assert client.last_headers.get("etag") == etag
+    assert sorted(client.list(bucket)) == keys
+    inventory = harness.residue_inventory(cluster, bucket, keys[0])
+    for node in inventory:
+        for drive in node:
+            external = any(path.rsplit("/", 1)[-1] == "part.1" for path in drive["files"])
+            assert external == (size > 128 * 1024), "the fixture must exercise its declared physical layout"
+    clean_stop = cluster.stop_clean(timeout=30)
+    restart_started = time.perf_counter()
+    cluster.restart()
+    ready_seconds = time.perf_counter() - restart_started
+    deadline = time.monotonic() + 90
+    while True:
+        startup_operations, completed = operation_snapshot(client)
+        if all(count >= 1 for count in completed):
+            break
+        if time.monotonic() >= deadline:
+            raise TimeoutError("clean restart did not complete raw discovery on all four disks")
+        time.sleep(0.02)
+    # Each fixture has one bucket and fewer than 4096 raw entries per disk,
+    # so its complete initial sweep uses one bounded page on each disk.
+    assert completed == [1] * 4, "unexpected discovery scopes invalidate startup attribution"
+    startup = {
+        "clean_exit_code": clean_stop["ExitCode"],
+        "ready_seconds": ready_seconds,
+        "initial_discovery_complete_seconds": time.perf_counter() - restart_started,
+        "discovery_within_30_seconds": time.perf_counter() - restart_started <= 30,
+        "raw_discovery_operations": startup_operations["discover_delete_transactions"],
+    }
+    assert sorted(client.list(bucket)) == keys, "the clean restart must preserve the fixture"
+    # Compare DELETE at the same process age. A faster startup must not move
+    # only the candidate into earlier periodic maintenance phases.
+    time.sleep(max(0, 35 - (time.perf_counter() - restart_started)))
+    before, _ = operation_snapshot(client)
+    latencies = []
+    started = time.perf_counter()
+    if batch:
+        client.batch_delete(bucket, keys)
+    else:
+        for key in keys:
+            tick = time.perf_counter()
+            client.delete(bucket, key)
+            latencies.append(time.perf_counter() - tick)
+    elapsed = time.perf_counter() - started
+    after, _ = operation_snapshot(client)
+    if time.perf_counter() - restart_started >= 59:
+        raise ValueError("the process-age bound no longer protects the rolling operation window")
+    assert client.list(bucket) == []
+    leftovers = cluster.inspect_data(
+        "from pathlib import Path\nimport json\n" + inspect.getsource(harness.has_retained_content) +
+        f"keys={keys!r};bucket={bucket!r}\n"
+        "print(json.dumps([str(d / bucket / key) for d in Path('/data').glob('d*') "
+        "for key in keys if has_retained_content(d / bucket / key)]))"
+    )
+    assert all(not node for node in leftovers), "requested keys must retain no payload, metadata, proof, or non-directory entries"
+    client.delete_bucket(bucket)
+    return {
+        "case": case, "repeat": repeat, "objects": count, "object_bytes": size,
+        "wall_seconds": elapsed, "operations": stage_delta(before, after), "clean_restart": startup,
+        "single_median_ms": statistics.median(latencies) * 1000 if latencies else None,
+        "complete_bytes_etag_and_physical_cleanup": True,
+    }
+
+
+def comparison_report(samples, drift_limit):
+    medians = {cohort: {case: statistics.median(item["wall_seconds"] for item in values if item["case"] == case)
+                       for case in ("inline-batch", "inline-single", "external-batch")}
+               for cohort, values in samples.items()}
+    drifts = {case: {"baseline_percent": drift_percent(medians["A1"][case], medians["A2"][case]),
+                     "candidate_percent": drift_percent(medians["B1"][case], medians["B2"][case])}
+              for case in medians["A1"]}
+    startup_ok = all(sample["clean_restart"]["discovery_within_30_seconds"]
+                     for cohort in ("B1", "B2") for sample in samples[cohort])
+    valid = startup_ok and all(value <= drift_limit for pair in drifts.values() for value in pair.values())
+    report = {"verdict": "comparable" if valid else "inconclusive", "medians_seconds": medians,
+              "drift": drifts, "candidate_discovery_within_30_seconds": startup_ok}
+    if valid:
+        report["wall_time_reduction_percent"] = {
+            case: (1 - statistics.median([medians["B1"][case], medians["B2"][case]]) /
+                   statistics.median([medians["A1"][case], medians["A2"][case]])) * 100
+            for case in medians["A1"]}
+    return report
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    for variant in ("baseline", "candidate"):
+        parser.add_argument(f"--{variant}-binary", type=Path, required=True)
+        parser.add_argument(f"--{variant}-receipt", type=Path, required=True)
+    parser.add_argument("--image", required=True, help="Existing local image with Python 3 and curl")
+    parser.add_argument("--artifact-dir", type=Path, required=True)
+    parser.add_argument("--cpuset-cpus", required=True)
+    parser.add_argument("--memory", default="4g")
+    parser.add_argument("--port", type=int, default=19640)
+    parser.add_argument("--repetitions", type=int, default=3)
+    parser.add_argument("--drift-limit-percent", type=float, default=10)
+    args = parser.parse_args()
+    if args.repetitions < 3 or not 0 < args.drift_limit_percent <= 10:
+        parser.error("at least three repetitions and a drift limit in (0, 10] are required")
+    if not 1024 <= args.port <= 65532 or not re.fullmatch(r"[0-9,-]+", args.cpuset_cpus):
+        parser.error("invalid port or CPU set")
+    baseline = binary_receipt(args.baseline_binary, args.baseline_receipt)
+    candidate = binary_receipt(args.candidate_binary, args.candidate_receipt)
+    if baseline["build"] != candidate["build"]:
+        parser.error("baseline and candidate build configurations differ")
+    args.artifact_dir.mkdir(parents=True, exist_ok=False)
+    inputs = {"baseline": baseline, "candidate": candidate, "configuration": vars(args).copy()}
+    (args.artifact_dir / "inputs.json").write_text(json.dumps(inputs, default=str, indent=2))
+    image = harness.docker("image", "inspect", "--format", "{{.Id}}", args.image).stdout.decode().strip()
+    original_docker = harness.docker
+
+    def resource_matched_docker(*command, check=True):
+        if command and command[0] == "run" and "-d" in command:
+            command = (command[0], "--cpuset-cpus", args.cpuset_cpus, "--memory", args.memory, *command[1:])
+        return original_docker(*command, check=check)
+
+    harness.docker = resource_matched_docker
+    samples = {}
+    for cohort, binary in (("A1", args.baseline_binary), ("B1", args.candidate_binary),
+                           ("B2", args.candidate_binary), ("A2", args.baseline_binary)):
+        root = args.artifact_dir / cohort
+        root.mkdir()
+        harness.ROOT, harness.PREFIX = root, f"delete-perf-{uuid.uuid4().hex[:12]}"
+        harness.IMAGE, harness.PORT = image, args.port
+        harness.CANDIDATE, harness.BASELINE = binary.resolve(), None
+        # Keep normal scanner/heal work out of stage attribution. Automatic
+        # transaction recovery is initialized by ECStore and remains enabled.
+        cluster = harness.Cluster("strict", drives=4, extra_env={
+            "RUSTFS_SCANNER_ENABLED": "false", "RUSTFS_HEAL_ENABLED": "false",
+        })
+        samples[cohort] = []
+        try:
+            cluster.create()
+            for repeat in range(args.repetitions):
+                for case, count, size, batch in (("inline-batch", 500, 4096, True),
+                                                ("inline-single", 32, 4096, False),
+                                                ("external-batch", 32, 1024 * 1024, True)):
+                    result = measure(cluster, case, count, size, batch, repeat)
+                    samples[cohort].append(result)
+                    (root / "samples.json").write_text(json.dumps(samples[cohort], indent=2))
+                    print(json.dumps({"cohort": cohort, **result}), flush=True)
+        finally:
+            cluster.remove()
+    binary_receipt(args.baseline_binary, args.baseline_receipt)
+    binary_receipt(args.candidate_binary, args.candidate_receipt)
+    report = comparison_report(samples, args.drift_limit_percent)
+    report["resource_contract"] = {
+        "cpuset_cpus": args.cpuset_cpus, "memory": args.memory, "image": image,
+        "topology": "1 process / 4 logical drives on one Docker host filesystem",
+    }
+    (args.artifact_dir / "result.json").write_text(json.dumps(report, indent=2))
+    print(json.dumps(report), flush=True)
+    return 0 if report["verdict"] == "comparable" else 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/test/test_delete_recovery_perf.py

@@ -0,0 +1,93 @@
+"""Regression checks for benchmark evidence and drift verdicts."""
+
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+from types import SimpleNamespace
+
+import delete_recovery_perf as perf
+
+
+class EvidenceTests(unittest.TestCase):
+    def test_only_empty_directory_trees_are_not_payload_residue(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "key"
+            (root / "transaction").mkdir(parents=True)
+            self.assertFalse(perf.harness.has_retained_content(root))
+            proof = root / "transaction" / "xl.meta.bkp"
+            proof.write_bytes(b"backup")
+            self.assertTrue(perf.harness.has_retained_content(root))
+            proof.unlink()
+            payload = root / "unknown-entry"
+            payload.write_bytes(b"payload")
+            self.assertTrue(perf.harness.has_retained_content(root))
+
+    def test_symlinks_cannot_be_mistaken_for_empty_directories(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / "key"
+            root.mkdir()
+            (root / "link").symlink_to(Path(directory) / "missing-target")
+            self.assertTrue(perf.harness.has_retained_content(root))
+
+    def test_clean_restart_rejects_forced_exit_or_oom(self):
+        for state in ({"ExitCode": 137, "OOMKilled": False}, {"ExitCode": 0, "OOMKilled": True}):
+            cluster = perf.harness.Cluster.__new__(perf.harness.Cluster)
+            cluster.names = ["owned-test-server"]
+            with patch.object(perf.harness, "docker", side_effect=[SimpleNamespace(), SimpleNamespace(stdout=json.dumps(state).encode())]):
+                with self.assertRaises(AssertionError):
+                    cluster.stop_clean()
+
+    def test_expired_operation_window_cannot_be_attributed_to_the_candidate(self):
+        with self.assertRaisesRegex(ValueError, "expired"):
+            perf.stage_delta({"prepare": {"count": 4, "ns": 100}}, {"prepare": {"count": 3, "ns": 90}})
+
+    def test_operation_service_times_keep_their_units(self):
+        self.assertEqual(perf.stage_delta({}, {"prepare": {"count": 4, "ns": 2_000_000}}),
+                         {"prepare": {"calls": 4, "service_ms": 2.0}})
+
+    def test_binary_identity_and_release_configuration_are_required(self):
+        with tempfile.TemporaryDirectory() as directory:
+            binary, receipt = Path(directory) / "binary", Path(directory) / "receipt.json"
+            binary.write_bytes(b"candidate")
+            value = {"binary_sha256": perf.binary_digest(binary), "source_head": "a" * 40,
+                     "build": {"profile": "debug", "opt_level": 0}}
+            receipt.write_text(json.dumps(value))
+            with self.assertRaisesRegex(ValueError, "release"):
+                perf.binary_receipt(binary, receipt)
+            binary.write_bytes(b"different executable")
+            with self.assertRaisesRegex(ValueError, "does not match"):
+                perf.binary_receipt(binary, receipt)
+
+    def test_control_or_candidate_drift_withholds_every_improvement_claim(self):
+        for unstable in ("A2", "B2"):
+            samples = self.samples()
+            for sample in samples[unstable]:
+                sample["wall_seconds"] *= 2
+            result = perf.comparison_report(samples, 10)
+            self.assertEqual(result["verdict"], "inconclusive")
+            self.assertNotIn("wall_time_reduction_percent", result)
+
+    def test_candidate_startup_timeout_withholds_improvement_claims(self):
+        samples = self.samples()
+        samples["B1"][0]["clean_restart"]["discovery_within_30_seconds"] = False
+        result = perf.comparison_report(samples, 10)
+        self.assertFalse(result["candidate_discovery_within_30_seconds"])
+        self.assertEqual(result["verdict"], "inconclusive")
+        self.assertNotIn("wall_time_reduction_percent", result)
+
+    def test_stable_controls_allow_a_quantified_result(self):
+        result = perf.comparison_report(self.samples(), 10)
+        self.assertEqual(result["verdict"], "comparable")
+        self.assertAlmostEqual(result["wall_time_reduction_percent"]["inline-batch"], 30)
+
+    @staticmethod
+    def samples():
+        return {cohort: [{"case": case, "wall_seconds": value, "clean_restart": {"discovery_within_30_seconds": True}} for case in
+                         ("inline-batch", "inline-single", "external-batch") for _ in range(3)]
+                for cohort, value in (("A1", 10), ("B1", 7), ("B2", 7), ("A2", 10))}
+
+
+if __name__ == "__main__":
+    unittest.main()