Refuse inconsistent State Anchor acquisition checkpoints

crates/lazaret-host-coordinator/src/state_anchor.rs

@@ -159,18 +159,29 @@ impl<'a> StateAnchorCoordinator<'a> {
             &expected_lease_held,
             &expected_anchor_intent,
         )? {
-            AcquisitionCheckpoint::Absent => self.domain_store.apply(&[
-                DomainStoreMutation::PutParticipant(Participant::bare_intent(
-                    generation_id,
-                    ParticipantRole::PreparedLower,
-                )),
-                DomainStoreMutation::PutParticipant(expected_prepared_intent.clone()),
-            ])?,
+            AcquisitionCheckpoint::Absent => {
+                if self.catalog_lease_checkpoint(&lease)? != CatalogLeaseCheckpoint::Absent {
+                    return Err(StateAnchorCoordinatorError::InvalidState(
+                        "catalog lease exists before Domain intent",
+                    ));
+                }
+                self.domain_store.apply(&[
+                    DomainStoreMutation::PutParticipant(Participant::bare_intent(
+                        generation_id,
+                        ParticipantRole::PreparedLower,
+                    )),
+                    DomainStoreMutation::PutParticipant(expected_prepared_intent.clone()),
+                ])?;
+            }
             AcquisitionCheckpoint::PreparedIntent => {}
             AcquisitionCheckpoint::Complete => {
-                self.catalog
-                    .acquire_or_resume_lease_checked(artifact, &lease)?;
-                return Ok(lease);
+                return if self.catalog_lease_checkpoint(&lease)? == CatalogLeaseCheckpoint::Exact {
+                    Ok(lease)
+                } else {
+                    Err(StateAnchorCoordinatorError::InvalidState(
+                        "completed Domain acquisition lacks its catalog lease",
+                    ))
+                };
             }
         }
         self.catalog
@@ -482,6 +493,22 @@ impl<'a> StateAnchorCoordinator<'a> {
         Ok(role(&snapshot.participants, generation_id, participant_role)?.clone())
     }
 
+    fn catalog_lease_checkpoint(
+        &self,
+        expected: &ArtifactLease,
+    ) -> Result<CatalogLeaseCheckpoint, StateAnchorCoordinatorError> {
+        let snapshot = self.catalog.snapshot()?;
+        match snapshot.artifact_leases.iter().find(|lease| {
+            lease.domain_id == expected.domain_id && lease.generation_id == expected.generation_id
+        }) {
+            None => Ok(CatalogLeaseCheckpoint::Absent),
+            Some(observed) if observed == expected => Ok(CatalogLeaseCheckpoint::Exact),
+            Some(_) => Err(StateAnchorCoordinatorError::InvalidState(
+                "catalog generation lease ownership changed",
+            )),
+        }
+    }
+
     fn release_catalog_lease_if_present(
         &self,
         authority: &StateAnchorReleaseAuthority,
@@ -593,6 +620,12 @@ enum AcquisitionCheckpoint {
     Complete,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum CatalogLeaseCheckpoint {
+    Absent,
+    Exact,
+}
+
 fn acquisition_checkpoint(
     participants: &[Participant],
     generation_id: GenerationId,

crates/lazaret-host-coordinator/src/state_anchor/tests.rs

@@ -286,6 +286,67 @@ fn acquisition_resumes_after_catalog_lease_reopen() {
     );
 }
 
+#[test]
+fn completed_domain_checkpoint_refuses_absent_catalog_lease() {
+    let fixture = AcquisitionFixture::new(19);
+    let layout = StateAnchorLayout::create(&fixture.configuration).unwrap();
+    let store = fixture.open_store();
+    let coordinator = StateAnchorCoordinator::new(&store, &fixture.catalog);
+    let lease = coordinator
+        .begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
+        .unwrap();
+    fixture.catalog.release_lease(&lease).unwrap();
+    drop(store);
+
+    let store = fixture.open_store();
+    let coordinator = StateAnchorCoordinator::new(&store, &fixture.catalog);
+    assert!(
+        coordinator
+            .begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
+            .is_err()
+    );
+    assert_fixture_states(&store, fixture.generation_id);
+    assert!(
+        fixture
+            .catalog
+            .snapshot()
+            .unwrap()
+            .artifact_leases
+            .is_empty()
+    );
+}
+
+#[test]
+fn absent_domain_checkpoint_refuses_preexisting_catalog_lease() {
+    let fixture = AcquisitionFixture::new(20);
+    let layout = StateAnchorLayout::create(&fixture.configuration).unwrap();
+    let lease = fixture.lease();
+    fixture
+        .catalog
+        .acquire_lease_checked(&fixture.artifact, &lease)
+        .unwrap();
+
+    let store = fixture.open_store();
+    let coordinator = StateAnchorCoordinator::new(&store, &fixture.catalog);
+    assert!(
+        coordinator
+            .begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
+            .is_err()
+    );
+    let snapshot = store.snapshot().unwrap();
+    assert!(!snapshot.participants.iter().any(|participant| {
+        participant.generation_id() == fixture.generation_id
+            && matches!(
+                participant.role(),
+                ParticipantRole::PreparedLower | ParticipantRole::StateAnchor
+            )
+    }));
+    assert_eq!(
+        fixture.catalog.snapshot().unwrap().artifact_leases,
+        vec![lease]
+    );
+}
+
 fn assert_fixture_states(store: &HostCoordinatorDomainStore, generation_id: GenerationId) {
     let snapshot = store.snapshot().unwrap();
     assert_eq!(