Make concurrency verification deterministic and exhaustive

crates/lazaret-prepared-lower/tests/prepared_lower.rs

@@ -1,11 +1,10 @@
 use std::fs::File;
 use std::io::{Cursor, Read, Write};
-use std::os::unix::ffi::OsStringExt;
+use std::os::unix::ffi::{OsStrExt, OsStringExt};
 use std::os::unix::fs::{MetadataExt, PermissionsExt};
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Barrier};
 use std::thread;
-use std::time::{Duration, Instant};
 
 use flate2::Compression;
 use flate2::write::GzEncoder;
@@ -14,6 +13,7 @@ use lazaret_core::identity::{
     SelectedManifestDigest, SourceIdentity,
 };
 use lazaret_core::records::{PrepareOperation, PrepareStageState};
+use lazaret_linux::filesystem::try_exclusive_flock;
 use lazaret_prepared_lower::{
     IdMap, IdMapRange, PrepareRecovery, PrepareRequest, PreparedDestination,
     PreparedDestinationKind, PreparedLowerError, PreparedLowerStore, ReclaimOutcome,

crates/lazaret-prepared-lower/tests/prepared_lower/publication.rs

@@ -76,112 +76,60 @@ fn concurrent_publication_reuses_one_immutable_artifact() {
 }
 
 #[test]
-fn startup_recovery_preserves_a_stage_owned_by_a_live_materializer() {
+fn startup_recovery_preserves_an_independently_locked_owned_stage() {
     let fixture = Fixture::new();
-    let first_layer = layer(&[Entry {
+    let operation_id = PrepareOperationId::new([42; 32]);
-        path: "large",
+    let stage_relative = PathBuf::from("stages").join(hex(operation_id.as_bytes()));
-        kind: EntryType::Regular,
+    let stage_path = fixture.root.join(&stage_relative);
-        data: &vec![0u8; 16 * 1024 * 1024],
+    std::fs::create_dir(&stage_path).unwrap();
-        target: None,
+    let metadata = stage_path.metadata().unwrap();
-        mode: 0o644,
+    let intent = PrepareOperation {
-    }]);
+        operation_id,
-    let second_layer = layer(&[Entry {
+        expected_artifact_id: PreparedArtifactId::new([43; 32]),
-        path: "marker",
+        source_identity: SourceIdentity::new(b"synthetic-live-owner".to_vec()),
-        kind: EntryType::Regular,
+        selected_manifest_digest: SelectedManifestDigest::new([44; 32]),
-        data: b"complete",
+        policy_id: PolicyId::new([45; 32]),
-        target: None,
+        id_map_digest: IdMapDigest::new([46; 32]),
-        mode: 0o644,
+        stage_path: HostPath::new(stage_relative.as_os_str().as_bytes().to_vec()),
-    }]);
+        stage_state: PrepareStageState::Intent,
-    let first_request = fixture.request(
+        device: None,
-        42,
+        inode: None,
-        &[(first_layer, OCI_TAR), (second_layer.clone(), OCI_TAR)],
-    );
-    let second_request = PrepareRequest {
-        operation_id: PrepareOperationId::new([43; 32]),
-        ..first_request.clone()
     };
-    let second_blob = first_request
+    let owned = PrepareOperation {
-        .layout_path
+        stage_state: PrepareStageState::Owned,
-        .join("blobs/sha256")
+        device: Some(DeviceId::new(metadata.dev())),
-        .join(hex(&Sha256::digest(&second_layer)));
+        inode: Some(InodeId::new(metadata.ino())),
-    let (_, first_store) = fixture.independent_store();
+        ..intent.clone()
-    let (second_catalog, second_store) = fixture.independent_store();
+    };
-    let first_handle = thread::spawn(move || first_store.prepare(&first_request));
+    fixture.catalog.record_prepare(&intent).unwrap();
+    fixture.catalog.bind_prepare_stage(&intent, &owned).unwrap();
+    let live_owner = try_exclusive_flock(File::open(&stage_path).unwrap()).unwrap();
+    let (_, recovery_store) = fixture.independent_store();
 
-    let deadline = Instant::now() + Duration::from_secs(5);
+    assert_eq!(
-    let operation = loop {
+        recovery_store.recover_preparations().unwrap(),
-        if let Some(operation) = second_catalog
+        vec![(operation_id, PrepareRecovery::PreservedLiveOwnedStage,)]
+    );
+    assert_eq!(
+        fixture.catalog.snapshot().unwrap().prepare_operations,
+        vec![owned]
+    );
+    assert!(stage_path.is_dir());
+
+    drop(live_owner);
+    assert_eq!(
+        recovery_store.recover_preparations().unwrap(),
+        vec![(operation_id, PrepareRecovery::RemovedOwnedStage)]
+    );
+    assert!(
+        fixture
+            .catalog
             .snapshot()
             .unwrap()
             .prepare_operations
-            .into_iter()
+            .is_empty()
-            .find(|operation| operation.stage_state == PrepareStageState::Owned)
-        {
-            break operation;
-        }
-        assert!(
-            !first_handle.is_finished(),
-            "materialization finished before synchronization"
-        );
-        assert!(
-            Instant::now() < deadline,
-            "materialization did not publish stage ownership"
-        );
-        thread::yield_now();
-    };
-
-    std::fs::remove_file(&second_blob).unwrap();
-    rustix::fs::mknodat(
-        rustix::fs::CWD,
-        &second_blob,
-        rustix::fs::FileType::Fifo,
-        rustix::fs::Mode::from_raw_mode(0o600),
-        0,
-    )
-    .unwrap();
-    thread::sleep(Duration::from_millis(25));
-    assert!(
-        !first_handle.is_finished(),
-        "materializer passed the synchronization FIFO"
-    );
-
-    assert_eq!(
-        second_store.recover_preparations().unwrap(),
-        vec![(
-            operation.operation_id,
-            PrepareRecovery::PreservedLiveOwnedStage,
-        )]
-    );
-
-    let writer_path = second_blob.clone();
-    let writer_bytes = second_layer.clone();
-    let writer = thread::spawn(move || {
-        std::fs::OpenOptions::new()
-            .write(true)
-            .open(writer_path)
-            .unwrap()
-            .write_all(&writer_bytes)
-            .unwrap();
-    });
-    let first = first_handle.join().unwrap().unwrap();
-    writer.join().unwrap();
-    std::fs::remove_file(&second_blob).unwrap();
-    std::fs::write(&second_blob, &second_layer).unwrap();
-
-    let second = second_store.prepare(&second_request).unwrap();
-    assert_eq!(first, second);
-    let snapshot = second_catalog.snapshot().unwrap();
-    assert_eq!(snapshot.prepared_artifacts, vec![first]);
-    assert!(snapshot.prepare_operations.is_empty());
-    assert!(
-        fixture
-            .root
-            .join("stages")
-            .read_dir()
-            .unwrap()
-            .next()
-            .is_none()
     );
+    assert!(!stage_path.exists());
 }
 
 #[test]
@@ -501,6 +449,74 @@ fn selected_gc_is_idempotent_only_at_the_durable_reclaim_boundary() {
     assert!(fixture.store.gc_artifacts(&[id]).is_err());
 }
 
+#[test]
+fn selected_artifact_excludes_gc_until_the_claim_is_released() {
+    let fixture = Fixture::new();
+    let artifact = fixture
+        .store
+        .prepare(&fixture.request(
+            72,
+            &[(
+                layer(&[Entry {
+                    path: "payload",
+                    kind: EntryType::Regular,
+                    data: b"selected",
+                    target: None,
+                    mode: 0o644,
+                }]),
+                OCI_TAR,
+            )],
+        ))
+        .unwrap();
+    let selected = fixture.store.select_artifact(&artifact).unwrap().unwrap();
+    let (_, gc_store) = fixture.independent_store();
+
+    let retained = gc_store
+        .gc_artifacts(&[artifact.prepared_artifact_id])
+        .unwrap();
+    assert_eq!(retained[0].outcome, ReclaimOutcome::RetainedBySelection);
+    assert_eq!(
+        fixture.catalog.snapshot().unwrap().prepared_artifacts,
+        vec![artifact.clone()]
+    );
+    assert!(fixture.artifact_root(&artifact).exists());
+
+    drop(selected);
+    let removed = gc_store
+        .gc_artifacts(&[artifact.prepared_artifact_id])
+        .unwrap();
+    assert_eq!(removed[0].outcome, ReclaimOutcome::Removed);
+}
+
+#[test]
+fn selection_rebinds_to_an_equivalent_replacement() {
+    let fixture = Fixture::new();
+    let bytes = layer(&[Entry {
+        path: "payload",
+        kind: EntryType::Regular,
+        data: b"replacement",
+        target: None,
+        mode: 0o644,
+    }]);
+    let request = fixture.request(73, &[(bytes, OCI_TAR)]);
+    let first = fixture.store.prepare(&request).unwrap();
+    assert_eq!(
+        fixture.store.reclaim(&first).unwrap(),
+        ReclaimOutcome::Removed
+    );
+    let replacement = fixture
+        .store
+        .prepare(&PrepareRequest {
+            operation_id: PrepareOperationId::new([74; 32]),
+            ..request
+        })
+        .unwrap();
+
+    let selection = fixture.store.select_artifact(&first).unwrap().unwrap();
+    assert_eq!(selection.artifact(), &replacement);
+    assert_ne!(selection.artifact().inode, first.inode);
+}
+
 #[test]
 fn destination_layout_is_materialized_and_canonical_in_artifact_identity() {
     let fixture = Fixture::new();

tests/tooling/test_verify.py

@@ -330,6 +330,7 @@ class VerificationTests(unittest.TestCase):
                 "lazaret-ignored-workspace",
                 "lazaret-launch-wrapper",
                 "lazaret-runtime-supervisor",
+                "lazaret-sandbox-init",
                 "lazaret-workspace-benchmark",
             },
         )
@@ -426,7 +427,9 @@ class VerificationTests(unittest.TestCase):
                 "--filterset",
                 (
                     "not binary(capability-drop-helper) and not binary(acquisition) "
-                    "and not binary(recovery-process) and not binary(runtime-composition) "
+                    "and not binary(state-anchor-lifecycle) and not binary(recovery-process) "
+                    "and not binary(runtime-composition) and not binary(escalation) "
+                    "and not binary(prepared-worker) "
                     "and not binary(network-runtime)"
                 ),
             ],
@@ -440,6 +443,16 @@ class VerificationTests(unittest.TestCase):
                 "acquisition",
                 "--locked",
             ],
+            [
+                "cargo",
+                "test",
+                "--offline",
+                "-p",
+                "lazaret-state-anchor",
+                "--test",
+                "state-anchor-lifecycle",
+                "--locked",
+            ],
             [
                 "cargo",
                 "test",
@@ -460,6 +473,26 @@ class VerificationTests(unittest.TestCase):
                 "capability-drop-helper",
                 "--locked",
             ],
+            [
+                "cargo",
+                "test",
+                "--offline",
+                "-p",
+                "lazaret-host-coordinator",
+                "--test",
+                "escalation",
+                "--locked",
+            ],
+            [
+                "cargo",
+                "test",
+                "--offline",
+                "-p",
+                "lazaret-host-coordinator",
+                "--test",
+                "prepared-worker",
+                "--locked",
+            ],
             [
                 "cargo",
                 "test",
@@ -505,6 +538,36 @@ class VerificationTests(unittest.TestCase):
         self.assertIn("exact uv 0.11.28", source)
         self.assertIn("Python 3.11 or newer", source)
 
+    def test_every_harnessless_fixture_is_excluded_and_run_explicitly(self) -> None:
+        harnessless = set()
+        for manifest_path in (ROOT / "crates").glob("*/Cargo.toml"):
+            manifest = VERIFY.load_toml(manifest_path)
+            package = manifest["package"]["name"]
+            for target in manifest.get("test", []):
+                if target.get("harness") is False:
+                    harnessless.add((package, target["name"]))
+
+        nextest_filter = VERIFY.NEXTEST_COMMAND[-1]
+        for _, target in harnessless:
+            self.assertIn(f"not binary({target})", nextest_filter)
+
+        commands = []
+        original_run = VERIFY.run
+        original_physical = VERIFY.run_physical
+        VERIFY.run = lambda command, cwd=VERIFY.ROOT: commands.append(command)
+        VERIFY.run_physical = lambda command, cwd=VERIFY.ROOT: commands.append(command)
+        try:
+            VERIFY.run_rust_tests()
+        finally:
+            VERIFY.run = original_run
+            VERIFY.run_physical = original_physical
+        explicit = {
+            (command[command.index("-p") + 1], command[command.index("--test") + 1])
+            for command in commands
+            if "-p" in command and "--test" in command
+        }
+        self.assertEqual(explicit, harnessless)
+
     def test_physical_test_accepts_only_pass_or_explicit_skip(self) -> None:
         with mock.patch.object(
             VERIFY.subprocess,

tools/lazaret_verify.py

@@ -33,7 +33,9 @@ NEXTEST_COMMAND = (
     "--filterset",
         (
             "not binary(capability-drop-helper) and not binary(acquisition) "
-        "and not binary(recovery-process) and not binary(runtime-composition) "
+            "and not binary(state-anchor-lifecycle) and not binary(recovery-process) "
+            "and not binary(runtime-composition) and not binary(escalation) "
+            "and not binary(prepared-worker) "
             "and not binary(network-runtime)"
         ),
 )
@@ -66,6 +68,16 @@ ACQUISITION_TEST_COMMAND = (
     "acquisition",
     "--locked",
 )
+STATE_ANCHOR_LIFECYCLE_TEST_COMMAND = (
+    "cargo",
+    "test",
+    "--offline",
+    "-p",
+    "lazaret-state-anchor",
+    "--test",
+    "state-anchor-lifecycle",
+    "--locked",
+)
 RECOVERY_PROCESS_TEST_COMMAND = (
     "cargo",
     "test",
@@ -76,6 +88,26 @@ RECOVERY_PROCESS_TEST_COMMAND = (
     "recovery-process",
     "--locked",
 )
+ESCALATION_TEST_COMMAND = (
+    "cargo",
+    "test",
+    "--offline",
+    "-p",
+    "lazaret-host-coordinator",
+    "--test",
+    "escalation",
+    "--locked",
+)
+PREPARED_WORKER_TEST_COMMAND = (
+    "cargo",
+    "test",
+    "--offline",
+    "-p",
+    "lazaret-host-coordinator",
+    "--test",
+    "prepared-worker",
+    "--locked",
+)
 RUNTIME_COMPOSITION_TEST_COMMAND = (
     "cargo",
     "test",
@@ -684,8 +716,11 @@ def run_rust_tests() -> None:
     run(list(FUSER_PATCH_TEST_COMMAND))
     run(list(NEXTEST_COMMAND))
     run(list(ACQUISITION_TEST_COMMAND))
+    run(list(STATE_ANCHOR_LIFECYCLE_TEST_COMMAND))
     run(list(RECOVERY_PROCESS_TEST_COMMAND))
     run(list(CAPABILITY_TEST_COMMAND))
+    run(list(ESCALATION_TEST_COMMAND))
+    run(list(PREPARED_WORKER_TEST_COMMAND))
     run_physical(list(RUNTIME_COMPOSITION_TEST_COMMAND))
     run_physical(list(NETWORK_RUNTIME_TEST_COMMAND))
     run(list(DOCTEST_COMMAND))