ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Make State Anchor acquisition resumable and descriptor-confined

Browse source
This commit is contained in:
ms2k 2026-07-15 17:27:39 +09:00
commit 48c10e9a7a
2 changed files with 1030 additions and 197 deletions
Download patch file Download diff file Expand all files Collapse all files

266
crates/lazaret-host-coordinator/src/state_anchor.rs
View file

@ -9,7 +9,9 @@ use lazaret_core::records::{
ParticipantState, PersistedHostProcessOwnership, PreparedArtifact, PreparedLowerOwnership,
ProcessOwnershipCommitAck, StagedProcessOwnership, StateAnchorOwnership,
};
use lazaret_state_anchor::{StateAnchorError, StateAnchorLayout, require_mount_absent};
use lazaret_state_anchor::{
StateAnchorError, StateAnchorLayout, VerifiedStateAnchorReadiness, require_mount_absent,
};
use lazaret_storage_catalog::{Catalog, CatalogError};
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
@ -66,7 +68,7 @@ impl From<StateAnchorError> for StateAnchorCoordinatorError {
pub struct StateAnchorReleaseAuthority {
generation_id: GenerationId,
lease: ArtifactLease,
merged: PathBuf,
merged_label: PathBuf,
mount: MountIdentity,
daemon: PidfsProcessIdentityV1,
}
@ -77,11 +79,6 @@ impl StateAnchorReleaseAuthority {
self.generation_id
}
#[must_use]
pub fn merged(&self) -> &Path {
&self.merged
}
#[must_use]
pub const fn mount(&self) -> MountIdentity {
self.mount
@ -110,7 +107,7 @@ impl<'a> StateAnchorCoordinator<'a> {
/// Commits prepared-lower intent, exact catalog lease, `LeaseHeld`, and State Anchor intent.
///
/// # Errors
/// Refuses existing participants, artifact drift, or any durable transition failure.
/// Refuses ownership drift or any durable transition failure.
pub fn begin_acquisition(
&self,
generation_id: GenerationId,
@ -118,16 +115,6 @@ impl<'a> StateAnchorCoordinator<'a> {
layout: &StateAnchorLayout,
) -> Result<ArtifactLease, StateAnchorCoordinatorError> {
let snapshot = self.domain_store.snapshot()?;
require_absent_role(
&snapshot.participants,
generation_id,
ParticipantRole::PreparedLower,
)?;
require_absent_role(
&snapshot.participants,
generation_id,
ParticipantRole::StateAnchor,
)?;
let lease = ArtifactLease {
prepared_artifact_id: artifact.prepared_artifact_id,
domain_id: snapshot.domain.domain_id,
@ -141,42 +128,60 @@ impl<'a> StateAnchorCoordinator<'a> {
device: Some(artifact.device),
inode: Some(artifact.inode),
};
self.domain_store.apply(&[
DomainStoreMutation::PutParticipant(Participant::bare_intent(
generation_id,
ParticipantRole::PreparedLower,
)),
DomainStoreMutation::PutParticipant(participant(
generation_id,
ParticipantRole::PreparedLower,
ParticipantState::Intent,
ParticipantOwnership::PreparedLower(prepared_intent.clone()),
)?),
])?;
self.catalog.acquire_lease_checked(artifact, &lease)?;
let expected_prepared_intent = participant(
generation_id,
ParticipantRole::PreparedLower,
ParticipantState::Intent,
ParticipantOwnership::PreparedLower(prepared_intent.clone()),
)?;
let lease_held = PreparedLowerOwnership {
lease_state: Some(ParticipantState::LeaseHeld),
..prepared_intent
};
let expected_lease_held = participant(
generation_id,
ParticipantRole::PreparedLower,
ParticipantState::LeaseHeld,
ParticipantOwnership::PreparedLower(lease_held.clone()),
)?;
let expected_anchor_intent = participant(
generation_id,
ParticipantRole::StateAnchor,
ParticipantState::Intent,
ParticipantOwnership::StateAnchor(
layout.ownership_intent(artifact.prepared_artifact_id),
),
)?;
match acquisition_checkpoint(
&snapshot.participants,
generation_id,
&expected_prepared_intent,
&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::PreparedIntent => {}
AcquisitionCheckpoint::Complete => {
self.catalog
.acquire_or_resume_lease_checked(artifact, &lease)?;
return Ok(lease);
}
}
self.catalog
.acquire_or_resume_lease_checked(artifact, &lease)?;
self.domain_store.apply(&[
DomainStoreMutation::PutParticipant(participant(
generation_id,
ParticipantRole::PreparedLower,
ParticipantState::LeaseHeld,
ParticipantOwnership::PreparedLower(lease_held),
)?),
DomainStoreMutation::PutParticipant(expected_lease_held),
DomainStoreMutation::PutParticipant(Participant::bare_intent(
generation_id,
ParticipantRole::StateAnchor,
)),
DomainStoreMutation::PutParticipant(participant(
generation_id,
ParticipantRole::StateAnchor,
ParticipantState::Intent,
ParticipantOwnership::StateAnchor(
layout.ownership_intent(artifact.prepared_artifact_id),
),
)?),
DomainStoreMutation::PutParticipant(expected_anchor_intent),
])?;
Ok(lease)
}
@ -190,28 +195,66 @@ impl<'a> StateAnchorCoordinator<'a> {
staged: StagedProcessOwnership,
) -> Result<(), StateAnchorCoordinatorError> {
self.require_staged_identity(staged)?;
self.update_daemon(
staged.generation_id(),
PersistedHostProcessOwnership::staged(staged.identity()),
)
let expected = PersistedHostProcessOwnership::staged(staged.identity());
let current = self.participant(staged.generation_id(), ParticipantRole::StateAnchor)?;
let Some(ParticipantOwnership::StateAnchor(ownership)) = current.ownership() else {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor ownership is invalid",
));
};
if current.state() != ParticipantState::Intent {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor daemon acquisition is not in Intent",
));
}
if ownership.daemon == expected && ownership.mount.is_none() {
return Ok(());
}
if ownership.daemon != PersistedHostProcessOwnership::unassigned() {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor daemon ownership changed",
));
}
self.update_daemon(staged.generation_id(), expected)
}
/// Commits the exact daemon and returns the acknowledgment that releases its mount command.
/// Acknowledges that the exact staged daemon is durable so its mount command may start.
///
/// # Errors
/// Refuses a staged identity mismatch or participant transition failure.
pub fn commit_daemon(
pub fn acknowledge_staged_daemon(
&self,
staged: StagedProcessOwnership,
) -> Result<ProcessOwnershipCommitAck, StateAnchorCoordinatorError> {
self.require_staged_identity(staged)?;
self.update_daemon(
self.require_exact_daemon(
staged.generation_id(),
PersistedHostProcessOwnership::committed(staged.identity()),
PersistedHostProcessOwnership::staged(staged.identity()),
)?;
Ok(ProcessOwnershipCommitAck::v1(staged))
}
/// Promotes the exact staged daemon only after physical target readiness was verified.
///
/// # Errors
/// Refuses a forged readiness witness, staged ownership drift, or transition failure.
pub fn commit_daemon_ready(
&self,
readiness: VerifiedStateAnchorReadiness,
) -> Result<(), StateAnchorCoordinatorError> {
let staged = readiness.staged();
self.require_staged_identity(staged)?;
self.require_exact_daemon(
staged.generation_id(),
PersistedHostProcessOwnership::staged(staged.identity()),
)?;
self.update_daemon_and_mount(
staged.generation_id(),
PersistedHostProcessOwnership::committed(staged.identity()),
Some(readiness.mount()),
)
}
/// Commits Prepared only after the Runtime Supervisor reports exact mount identity.
///
/// # Errors
@ -231,22 +274,18 @@ impl<'a> StateAnchorCoordinator<'a> {
if current.state() != ParticipantState::Intent
|| ownership.lower_artifact_id != Some(prepared_artifact_id)
|| !ownership.daemon.is_current_committed()
|| ownership.mount.is_some()
|| ownership.mount != Some(mount)
{
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor is not ready for mount commit",
));
}
let prepared = StateAnchorOwnership {
mount: Some(mount),
..ownership.clone()
};
self.domain_store
.apply(&[DomainStoreMutation::PutParticipant(participant(
generation_id,
ParticipantRole::StateAnchor,
ParticipantState::Prepared,
ParticipantOwnership::StateAnchor(prepared),
ParticipantOwnership::StateAnchor(ownership.clone()),
)?)])?;
Ok(())
}
@ -258,6 +297,7 @@ impl<'a> StateAnchorCoordinator<'a> {
pub fn begin_release(
&self,
generation_id: GenerationId,
layout: &StateAnchorLayout<'_>,
) -> Result<StateAnchorReleaseAuthority, StateAnchorCoordinatorError> {
let snapshot = self.domain_store.snapshot()?;
let anchor = role(
@ -271,6 +311,16 @@ impl<'a> StateAnchorCoordinator<'a> {
ParticipantRole::PreparedLower,
)?;
let authority = release_authority(&snapshot.domain, anchor, prepared)?;
let Some(ParticipantOwnership::StateAnchor(ownership)) = anchor.ownership() else {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor ownership is invalid",
));
};
if !layout.matches_recorded_paths(ownership) {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor layout changed",
));
}
self.domain_store
.apply(&[DomainStoreMutation::PutParticipant(participant(
generation_id,
@ -293,8 +343,18 @@ impl<'a> StateAnchorCoordinator<'a> {
pub fn complete_release_after_unmount(
&self,
authority: &StateAnchorReleaseAuthority,
layout: &StateAnchorLayout<'_>,
) -> Result<(), StateAnchorCoordinatorError> {
require_mount_absent(&authority.merged)?;
let expected_layout = layout.ownership_intent(authority.lease.prepared_artifact_id);
let expected_merged = expected_layout.merged_path.as_ref().ok_or(
StateAnchorCoordinatorError::InvalidState("State Anchor merged path is absent"),
)?;
if path_from_record(expected_merged) != authority.merged_label {
return Err(StateAnchorCoordinatorError::InvalidState(
"release authority layout changed",
));
}
require_mount_absent(layout)?;
let anchor = self.participant(authority.generation_id, ParticipantRole::StateAnchor)?;
let prepared = self.participant(authority.generation_id, ParticipantRole::PreparedLower)?;
match (anchor.state(), prepared.state()) {
@ -321,7 +381,7 @@ impl<'a> StateAnchorCoordinator<'a> {
));
}
}
self.release_catalog_lease_if_present(authority)?;
self.release_catalog_lease_if_present(authority, layout)?;
self.domain_store
.apply(&[DomainStoreMutation::CleanParticipantAfterAbsence(
clean_participant(authority.generation_id, ParticipantRole::PreparedLower)?,
@ -354,6 +414,15 @@ impl<'a> StateAnchorCoordinator<'a> {
&self,
generation_id: GenerationId,
daemon: PersistedHostProcessOwnership,
) -> Result<(), StateAnchorCoordinatorError> {
self.update_daemon_and_mount(generation_id, daemon, None)
}
fn update_daemon_and_mount(
&self,
generation_id: GenerationId,
daemon: PersistedHostProcessOwnership,
mount: Option<MountIdentity>,
) -> Result<(), StateAnchorCoordinatorError> {
let current = self.participant(generation_id, ParticipantRole::StateAnchor)?;
let Some(ParticipantOwnership::StateAnchor(ownership)) = current.ownership() else {
@ -368,6 +437,7 @@ impl<'a> StateAnchorCoordinator<'a> {
}
let next = StateAnchorOwnership {
daemon,
mount,
..ownership.clone()
};
self.domain_store
@ -380,6 +450,29 @@ impl<'a> StateAnchorCoordinator<'a> {
Ok(())
}
fn require_exact_daemon(
&self,
generation_id: GenerationId,
expected: PersistedHostProcessOwnership,
) -> Result<(), StateAnchorCoordinatorError> {
let current = self.participant(generation_id, ParticipantRole::StateAnchor)?;
let Some(ParticipantOwnership::StateAnchor(ownership)) = current.ownership() else {
return Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor ownership is invalid",
));
};
if current.state() == ParticipantState::Intent
&& ownership.daemon == expected
&& ownership.mount.is_none()
{
Ok(())
} else {
Err(StateAnchorCoordinatorError::InvalidState(
"State Anchor daemon ownership changed",
))
}
}
fn participant(
&self,
generation_id: GenerationId,
@ -392,6 +485,7 @@ impl<'a> StateAnchorCoordinator<'a> {
fn release_catalog_lease_if_present(
&self,
authority: &StateAnchorReleaseAuthority,
layout: &StateAnchorLayout<'_>,
) -> Result<(), StateAnchorCoordinatorError> {
let snapshot = self.catalog.snapshot()?;
let generation_lease = snapshot.artifact_leases.iter().find(|lease| {
@ -431,8 +525,7 @@ impl<'a> StateAnchorCoordinator<'a> {
"Domain release evidence is not committed",
)));
}
require_mount_absent(&authority.merged)
.map_err(|error| CatalogError::Io(io::Error::other(error)))
require_mount_absent(layout).map_err(|error| CatalogError::Io(io::Error::other(error)))
})?;
Ok(())
}
@ -477,7 +570,7 @@ fn release_authority(
domain_id: domain.domain_id,
generation_id: anchor.generation_id(),
},
merged: path_from_record(anchor_ownership.merged_path.as_ref().ok_or(
merged_label: path_from_record(anchor_ownership.merged_path.as_ref().ok_or(
StateAnchorCoordinatorError::InvalidState("State Anchor merged path is absent"),
)?)
.to_owned(),
@ -493,19 +586,42 @@ fn release_authority(
})
}
fn require_absent_role(
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum AcquisitionCheckpoint {
Absent,
PreparedIntent,
Complete,
}
fn acquisition_checkpoint(
participants: &[Participant],
generation_id: GenerationId,
participant_role: ParticipantRole,
) -> Result<(), StateAnchorCoordinatorError> {
if participants.iter().any(|participant| {
participant.generation_id() == generation_id && participant.role() == participant_role
}) {
Err(StateAnchorCoordinatorError::InvalidState(
"participant already exists",
))
} else {
Ok(())
expected_prepared_intent: &Participant,
expected_lease_held: &Participant,
expected_anchor_intent: &Participant,
) -> Result<AcquisitionCheckpoint, StateAnchorCoordinatorError> {
let prepared = participants.iter().find(|participant| {
participant.generation_id() == generation_id
&& participant.role() == ParticipantRole::PreparedLower
});
let anchor = participants.iter().find(|participant| {
participant.generation_id() == generation_id
&& participant.role() == ParticipantRole::StateAnchor
});
match (prepared, anchor) {
(None, None) => Ok(AcquisitionCheckpoint::Absent),
(Some(observed), None) if observed == expected_prepared_intent => {
Ok(AcquisitionCheckpoint::PreparedIntent)
}
(Some(observed_prepared), Some(observed_anchor))
if observed_prepared == expected_lease_held
&& observed_anchor == expected_anchor_intent =>
{
Ok(AcquisitionCheckpoint::Complete)
}
_ => Err(StateAnchorCoordinatorError::InvalidState(
"acquisition checkpoint ownership changed",
)),
}
}

260
crates/lazaret-host-coordinator/src/state_anchor/tests.rs
View file

@ -98,8 +98,57 @@ fn durable_phases_resume_release_without_collapsing_lease_order() {
);
assert!(coordinator.stage_daemon(wrong_coordinator).is_err());
coordinator.stage_daemon(staged).unwrap();
let acknowledgment = coordinator.commit_daemon(staged).unwrap();
let acknowledgment = coordinator.acknowledge_staged_daemon(staged).unwrap();
assert!(acknowledgment.acknowledges(&staged));
let staged_anchor = coordinator
.participant(generation_id, ParticipantRole::StateAnchor)
.unwrap();
let ParticipantOwnership::StateAnchor(staged_ownership) = staged_anchor.ownership().unwrap()
else {
panic!("wrong ownership")
};
assert_eq!(
staged_ownership.daemon,
PersistedHostProcessOwnership::staged(staged.identity())
);
assert!(
coordinator
.commit_prepared(
generation_id,
artifact.prepared_artifact_id,
MountIdentity {
mount_id: MountId::new(90),
device: DeviceId::new(91),
inode: InodeId::new(92),
},
)
.is_err()
);
let anchor = coordinator
.participant(generation_id, ParticipantRole::StateAnchor)
.unwrap();
let ParticipantOwnership::StateAnchor(ownership) = anchor.ownership().unwrap() else {
panic!("wrong ownership")
};
domain_store
.apply(&[DomainStoreMutation::PutParticipant(
participant(
generation_id,
ParticipantRole::StateAnchor,
ParticipantState::Intent,
ParticipantOwnership::StateAnchor(StateAnchorOwnership {
daemon: PersistedHostProcessOwnership::committed(staged.identity()),
mount: Some(MountIdentity {
mount_id: MountId::new(90),
device: DeviceId::new(91),
inode: InodeId::new(92),
}),
..ownership.clone()
}),
)
.unwrap(),
)])
.unwrap();
coordinator
.commit_prepared(
generation_id,
@ -117,7 +166,7 @@ fn durable_phases_resume_release_without_collapsing_lease_order() {
ParticipantState::Prepared,
);
let authority = coordinator.begin_release(generation_id).unwrap();
let authority = coordinator.begin_release(generation_id, &layout).unwrap();
assert_states(
&domain_store,
ParticipantState::LeaseHeld,
@ -140,7 +189,7 @@ fn durable_phases_resume_release_without_collapsing_lease_order() {
])
.unwrap();
coordinator
.complete_release_after_unmount(&authority)
.complete_release_after_unmount(&authority, &layout)
.unwrap();
assert_states(
&domain_store,
@ -150,6 +199,211 @@ fn durable_phases_resume_release_without_collapsing_lease_order() {
assert!(catalog.snapshot().unwrap().artifact_leases.is_empty());
}
#[test]
fn acquisition_resumes_after_prepared_intent_reopen() {
let fixture = AcquisitionFixture::new(17);
let layout = StateAnchorLayout::create(&fixture.configuration).unwrap();
let store = fixture.open_store();
store
.apply(&[
DomainStoreMutation::PutParticipant(Participant::bare_intent(
fixture.generation_id,
ParticipantRole::PreparedLower,
)),
DomainStoreMutation::PutParticipant(fixture.prepared_intent()),
])
.unwrap();
drop(store);
let store = fixture.open_store();
let coordinator = StateAnchorCoordinator::new(&store, &fixture.catalog);
let lease = coordinator
.begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
.unwrap();
assert_eq!(
fixture.catalog.snapshot().unwrap().artifact_leases,
vec![lease]
);
assert_fixture_states(&store, fixture.generation_id);
assert_eq!(
coordinator
.begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
.unwrap(),
lease
);
}
#[test]
fn acquisition_resumes_after_catalog_lease_reopen() {
let fixture = AcquisitionFixture::new(18);
let layout = StateAnchorLayout::create(&fixture.configuration).unwrap();
let store = fixture.open_store();
store
.apply(&[
DomainStoreMutation::PutParticipant(Participant::bare_intent(
fixture.generation_id,
ParticipantRole::PreparedLower,
)),
DomainStoreMutation::PutParticipant(fixture.prepared_intent()),
])
.unwrap();
let lease = fixture.lease();
fixture
.catalog
.acquire_lease_checked(&fixture.artifact, &lease)
.unwrap();
drop(store);
let store = fixture.open_store();
let coordinator = StateAnchorCoordinator::new(&store, &fixture.catalog);
assert_eq!(
coordinator
.begin_acquisition(fixture.generation_id, &fixture.artifact, &layout)
.unwrap(),
lease
);
assert_fixture_states(&store, fixture.generation_id);
let mut changed_artifact = fixture.artifact.clone();
changed_artifact.inode = InodeId::new(changed_artifact.inode.get() + 1);
assert!(
coordinator
.begin_acquisition(fixture.generation_id, &changed_artifact, &layout)
.is_err()
);
let other_tree = TestTree::new();
let other_configuration = configuration(&other_tree);
let other_layout = StateAnchorLayout::create(&other_configuration).unwrap();
assert!(
coordinator
.begin_acquisition(fixture.generation_id, &fixture.artifact, &other_layout)
.is_err()
);
assert!(
coordinator
.begin_acquisition(GenerationId::new(999), &fixture.artifact, &layout)
.is_err()
);
}
fn assert_fixture_states(store: &HostCoordinatorDomainStore, generation_id: GenerationId) {
let snapshot = store.snapshot().unwrap();
assert_eq!(
role(
&snapshot.participants,
generation_id,
ParticipantRole::PreparedLower,
)
.unwrap()
.state(),
ParticipantState::LeaseHeld
);
assert_eq!(
role(
&snapshot.participants,
generation_id,
ParticipantRole::StateAnchor,
)
.unwrap()
.state(),
ParticipantState::Intent
);
}
struct AcquisitionFixture {
tree: TestTree,
configuration: lazaret_core::configuration::ResolvedDomainConfiguration,
artifact: PreparedArtifact,
catalog: Catalog,
domain: Domain,
generation_id: GenerationId,
}
impl AcquisitionFixture {
fn new(generation: u64) -> Self {
let tree = TestTree::new();
let configuration = configuration(&tree);
let lower = tree.path.join("prepared-lower");
fs::create_dir(&lower).unwrap();
let artifact = artifact(&lower);
let catalog = Catalog::new(
tree.path.join("catalog.redb"),
tree.path.join("storage-catalog.lock"),
);
catalog.initialize().unwrap();
publish(&catalog, &artifact);
let domain = Domain {
domain_id: configuration.domain_id(),
profile_name: configuration.selected_profile().to_owned(),
canonical_workspace: configuration.workspace().canonical_label().clone(),
};
let generation_id = GenerationId::new(generation);
let fixture = Self {
tree,
configuration,
artifact,
catalog,
domain,
generation_id,
};
let store = fixture.open_store();
store
.apply(&[DomainStoreMutation::PutGeneration(Generation {
domain_id: fixture.domain.domain_id,
generation_id,
configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
coordinator_id: CoordinatorId::new([3; 32]),
network_sidecar_id: NetworkSidecarId::new([4; 32]),
sandbox_id: SandboxId::new([5; 32]),
durable_decision: DurableDecision::None,
})])
.unwrap();
drop(store);
fixture
}
fn open_store(&self) -> HostCoordinatorDomainStore {
let lock_file = OpenOptions::new()
.create(true)
.read(true)
.truncate(false)
.write(true)
.open(self.tree.path.join("domain.lock"))
.unwrap();
HostCoordinatorDomainStore::open(
try_exclusive_flock(lock_file).unwrap(),
&self.tree.path.join("domain.redb"),
&self.domain,
)
.unwrap()
}
fn lease(&self) -> ArtifactLease {
ArtifactLease {
prepared_artifact_id: self.artifact.prepared_artifact_id,
domain_id: self.domain.domain_id,
generation_id: self.generation_id,
}
}
fn prepared_intent(&self) -> Participant {
participant(
self.generation_id,
ParticipantRole::PreparedLower,
ParticipantState::Intent,
ParticipantOwnership::PreparedLower(PreparedLowerOwnership {
prepared_artifact_id: Some(self.artifact.prepared_artifact_id),
source_identity: Some(self.artifact.source_identity.clone()),
lease_state: Some(ParticipantState::Intent),
path: Some(self.artifact.path.clone()),
device: Some(self.artifact.device),
inode: Some(self.artifact.inode),
}),
)
.unwrap()
}
}
fn assert_states(
store: &HostCoordinatorDomainStore,
prepared: ParticipantState,