Implement exact State Anchor mount and lease lifecycle
This commit is contained in:
parent
27bc3d835e
commit
e6d6406c64
5 changed files with 1852 additions and 0 deletions
|
|
@ -12,6 +12,8 @@ path = "src/main.rs"
|
|||
[dependencies]
|
||||
lazaret-core = { version = "=0.0.0", path = "../lazaret-core" }
|
||||
lazaret-linux = { version = "=0.0.0", path = "../lazaret-linux" }
|
||||
lazaret-state-anchor = { version = "=0.0.0", path = "../lazaret-state-anchor" }
|
||||
lazaret-storage-catalog = { version = "=0.0.0", path = "../lazaret-storage-catalog" }
|
||||
redb.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
#![doc = "Host-side coordination authority for one Lazaret Domain."]
|
||||
|
||||
pub mod domain_store;
|
||||
pub mod state_anchor;
|
||||
|
|
|
|||
|
|
@ -1,3 +1,11 @@
|
|||
fn main() {
|
||||
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
|
||||
if let Some(result) = lazaret_linux::process::dispatch_private_process_role(&arguments) {
|
||||
if let Err(error) = result {
|
||||
eprintln!("private process role failed: {error}");
|
||||
std::process::exit(1);
|
||||
}
|
||||
return;
|
||||
}
|
||||
let _ = lazaret_core::PROTOCOL_NAME;
|
||||
}
|
||||
|
|
|
|||
577
crates/lazaret-host-coordinator/src/state_anchor.rs
Normal file
577
crates/lazaret-host-coordinator/src/state_anchor.rs
Normal file
|
|
@ -0,0 +1,577 @@
|
|||
use std::fmt;
|
||||
use std::io;
|
||||
use std::os::unix::ffi::OsStrExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use lazaret_core::identity::{GenerationId, MountIdentity, PidfsProcessIdentityV1};
|
||||
use lazaret_core::records::{
|
||||
ArtifactLease, HostProcessOwnershipSlot, Participant, ParticipantOwnership, ParticipantRole,
|
||||
ParticipantState, PersistedHostProcessOwnership, PreparedArtifact, PreparedLowerOwnership,
|
||||
ProcessOwnershipCommitAck, StagedProcessOwnership, StateAnchorOwnership,
|
||||
};
|
||||
use lazaret_state_anchor::{StateAnchorError, StateAnchorLayout, require_mount_absent};
|
||||
use lazaret_storage_catalog::{Catalog, CatalogError};
|
||||
|
||||
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum StateAnchorCoordinatorError {
|
||||
DomainStore(DomainStoreError),
|
||||
Catalog(CatalogError),
|
||||
Physical(StateAnchorError),
|
||||
InvalidState(&'static str),
|
||||
}
|
||||
|
||||
impl fmt::Display for StateAnchorCoordinatorError {
|
||||
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
Self::DomainStore(error) => write!(formatter, "Domain Store failed: {error}"),
|
||||
Self::Catalog(error) => write!(formatter, "Storage Catalog failed: {error}"),
|
||||
Self::Physical(error) => write!(formatter, "State Anchor failed: {error}"),
|
||||
Self::InvalidState(detail) => write!(formatter, "State Anchor state refused: {detail}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for StateAnchorCoordinatorError {
|
||||
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
|
||||
match self {
|
||||
Self::DomainStore(error) => Some(error),
|
||||
Self::Catalog(error) => Some(error),
|
||||
Self::Physical(error) => Some(error),
|
||||
Self::InvalidState(_) => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<DomainStoreError> for StateAnchorCoordinatorError {
|
||||
fn from(error: DomainStoreError) -> Self {
|
||||
Self::DomainStore(error)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<CatalogError> for StateAnchorCoordinatorError {
|
||||
fn from(error: CatalogError) -> Self {
|
||||
Self::Catalog(error)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<StateAnchorError> for StateAnchorCoordinatorError {
|
||||
fn from(error: StateAnchorError) -> Self {
|
||||
Self::Physical(error)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub struct StateAnchorReleaseAuthority {
|
||||
generation_id: GenerationId,
|
||||
lease: ArtifactLease,
|
||||
merged: PathBuf,
|
||||
mount: MountIdentity,
|
||||
daemon: PidfsProcessIdentityV1,
|
||||
}
|
||||
|
||||
impl StateAnchorReleaseAuthority {
|
||||
#[must_use]
|
||||
pub const fn generation_id(&self) -> GenerationId {
|
||||
self.generation_id
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn merged(&self) -> &Path {
|
||||
&self.merged
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn mount(&self) -> MountIdentity {
|
||||
self.mount
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn daemon(&self) -> PidfsProcessIdentityV1 {
|
||||
self.daemon
|
||||
}
|
||||
}
|
||||
|
||||
pub struct StateAnchorCoordinator<'a> {
|
||||
domain_store: &'a HostCoordinatorDomainStore,
|
||||
catalog: &'a Catalog,
|
||||
}
|
||||
|
||||
impl<'a> StateAnchorCoordinator<'a> {
|
||||
#[must_use]
|
||||
pub const fn new(domain_store: &'a HostCoordinatorDomainStore, catalog: &'a Catalog) -> Self {
|
||||
Self {
|
||||
domain_store,
|
||||
catalog,
|
||||
}
|
||||
}
|
||||
|
||||
/// Commits prepared-lower intent, exact catalog lease, `LeaseHeld`, and State Anchor intent.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses existing participants, artifact drift, or any durable transition failure.
|
||||
pub fn begin_acquisition(
|
||||
&self,
|
||||
generation_id: GenerationId,
|
||||
artifact: &PreparedArtifact,
|
||||
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,
|
||||
generation_id,
|
||||
};
|
||||
let prepared_intent = PreparedLowerOwnership {
|
||||
prepared_artifact_id: Some(artifact.prepared_artifact_id),
|
||||
source_identity: Some(artifact.source_identity.clone()),
|
||||
lease_state: Some(ParticipantState::Intent),
|
||||
path: Some(artifact.path.clone()),
|
||||
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 lease_held = PreparedLowerOwnership {
|
||||
lease_state: Some(ParticipantState::LeaseHeld),
|
||||
..prepared_intent
|
||||
};
|
||||
self.domain_store.apply(&[
|
||||
DomainStoreMutation::PutParticipant(participant(
|
||||
generation_id,
|
||||
ParticipantRole::PreparedLower,
|
||||
ParticipantState::LeaseHeld,
|
||||
ParticipantOwnership::PreparedLower(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),
|
||||
),
|
||||
)?),
|
||||
])?;
|
||||
Ok(lease)
|
||||
}
|
||||
|
||||
/// Commits the exact Runtime Supervisor-reported staged daemon acquisition.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses the wrong Domain, generation, slot, or participant phase.
|
||||
pub fn stage_daemon(
|
||||
&self,
|
||||
staged: StagedProcessOwnership,
|
||||
) -> Result<(), StateAnchorCoordinatorError> {
|
||||
self.require_staged_identity(staged)?;
|
||||
self.update_daemon(
|
||||
staged.generation_id(),
|
||||
PersistedHostProcessOwnership::staged(staged.identity()),
|
||||
)
|
||||
}
|
||||
|
||||
/// Commits the exact daemon and returns the acknowledgment that releases its mount command.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses a staged identity mismatch or participant transition failure.
|
||||
pub fn commit_daemon(
|
||||
&self,
|
||||
staged: StagedProcessOwnership,
|
||||
) -> Result<ProcessOwnershipCommitAck, StateAnchorCoordinatorError> {
|
||||
self.require_staged_identity(staged)?;
|
||||
self.update_daemon(
|
||||
staged.generation_id(),
|
||||
PersistedHostProcessOwnership::committed(staged.identity()),
|
||||
)?;
|
||||
Ok(ProcessOwnershipCommitAck::v1(staged))
|
||||
}
|
||||
|
||||
/// Commits Prepared only after the Runtime Supervisor reports exact mount identity.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses noncommitted daemon ownership, artifact drift, or replacement of mount evidence.
|
||||
pub fn commit_prepared(
|
||||
&self,
|
||||
generation_id: GenerationId,
|
||||
prepared_artifact_id: lazaret_core::identity::PreparedArtifactId,
|
||||
mount: MountIdentity,
|
||||
) -> 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.lower_artifact_id != Some(prepared_artifact_id)
|
||||
|| !ownership.daemon.is_current_committed()
|
||||
|| ownership.mount.is_some()
|
||||
{
|
||||
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),
|
||||
)?)])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Commits `ReleaseIntent` and returns the exact authority needed by the Runtime Supervisor.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses participants that are not Prepared and `LeaseHeld` or incomplete physical evidence.
|
||||
pub fn begin_release(
|
||||
&self,
|
||||
generation_id: GenerationId,
|
||||
) -> Result<StateAnchorReleaseAuthority, StateAnchorCoordinatorError> {
|
||||
let snapshot = self.domain_store.snapshot()?;
|
||||
let anchor = role(
|
||||
&snapshot.participants,
|
||||
generation_id,
|
||||
ParticipantRole::StateAnchor,
|
||||
)?;
|
||||
let prepared = role(
|
||||
&snapshot.participants,
|
||||
generation_id,
|
||||
ParticipantRole::PreparedLower,
|
||||
)?;
|
||||
let authority = release_authority(&snapshot.domain, anchor, prepared)?;
|
||||
self.domain_store
|
||||
.apply(&[DomainStoreMutation::PutParticipant(participant(
|
||||
generation_id,
|
||||
ParticipantRole::StateAnchor,
|
||||
ParticipantState::ReleaseIntent,
|
||||
anchor
|
||||
.ownership()
|
||||
.cloned()
|
||||
.ok_or(StateAnchorCoordinatorError::InvalidState(
|
||||
"State Anchor ownership is absent",
|
||||
))?,
|
||||
)?)])?;
|
||||
Ok(authority)
|
||||
}
|
||||
|
||||
/// Completes Domain Clean, exact catalog deletion, and prepared-lower Clean after unmount.
|
||||
///
|
||||
/// # Errors
|
||||
/// Refuses a stale authority, any remaining mount, or release-evidence mismatch.
|
||||
pub fn complete_release_after_unmount(
|
||||
&self,
|
||||
authority: &StateAnchorReleaseAuthority,
|
||||
) -> Result<(), StateAnchorCoordinatorError> {
|
||||
require_mount_absent(&authority.merged)?;
|
||||
let anchor = self.participant(authority.generation_id, ParticipantRole::StateAnchor)?;
|
||||
let prepared = self.participant(authority.generation_id, ParticipantRole::PreparedLower)?;
|
||||
match (anchor.state(), prepared.state()) {
|
||||
(ParticipantState::ReleaseIntent, ParticipantState::LeaseHeld) => {
|
||||
if release_authority(&self.domain_store.snapshot()?.domain, &anchor, &prepared)?
|
||||
!= *authority
|
||||
{
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"release authority does not match durable state",
|
||||
));
|
||||
}
|
||||
self.domain_store.apply(&[
|
||||
DomainStoreMutation::CleanParticipantAfterAbsence(clean_participant(
|
||||
authority.generation_id,
|
||||
ParticipantRole::StateAnchor,
|
||||
)?),
|
||||
DomainStoreMutation::PutParticipant(prepared_release_intent(&prepared)?),
|
||||
])?;
|
||||
}
|
||||
(ParticipantState::Clean, ParticipantState::ReleaseIntent) => {}
|
||||
_ => {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"durable release checkpoints do not match the authority",
|
||||
));
|
||||
}
|
||||
}
|
||||
self.release_catalog_lease_if_present(authority)?;
|
||||
self.domain_store
|
||||
.apply(&[DomainStoreMutation::CleanParticipantAfterAbsence(
|
||||
clean_participant(authority.generation_id, ParticipantRole::PreparedLower)?,
|
||||
)])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn require_staged_identity(
|
||||
&self,
|
||||
staged: StagedProcessOwnership,
|
||||
) -> Result<(), StateAnchorCoordinatorError> {
|
||||
let snapshot = self.domain_store.snapshot()?;
|
||||
let generation_matches = snapshot.generations.iter().any(|generation| {
|
||||
generation.generation_id == staged.generation_id()
|
||||
&& generation.coordinator_id == staged.coordinator_id()
|
||||
});
|
||||
if staged.domain_id() == snapshot.domain.domain_id
|
||||
&& staged.slot() == HostProcessOwnershipSlot::StateAnchorDaemon
|
||||
&& generation_matches
|
||||
{
|
||||
Ok(())
|
||||
} else {
|
||||
Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"staged daemon context mismatch",
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn update_daemon(
|
||||
&self,
|
||||
generation_id: GenerationId,
|
||||
daemon: 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 {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"State Anchor daemon acquisition is not in Intent",
|
||||
));
|
||||
}
|
||||
let next = StateAnchorOwnership {
|
||||
daemon,
|
||||
..ownership.clone()
|
||||
};
|
||||
self.domain_store
|
||||
.apply(&[DomainStoreMutation::PutParticipant(participant(
|
||||
generation_id,
|
||||
ParticipantRole::StateAnchor,
|
||||
ParticipantState::Intent,
|
||||
ParticipantOwnership::StateAnchor(next),
|
||||
)?)])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn participant(
|
||||
&self,
|
||||
generation_id: GenerationId,
|
||||
participant_role: ParticipantRole,
|
||||
) -> Result<Participant, StateAnchorCoordinatorError> {
|
||||
let snapshot = self.domain_store.snapshot()?;
|
||||
Ok(role(&snapshot.participants, generation_id, participant_role)?.clone())
|
||||
}
|
||||
|
||||
fn release_catalog_lease_if_present(
|
||||
&self,
|
||||
authority: &StateAnchorReleaseAuthority,
|
||||
) -> Result<(), StateAnchorCoordinatorError> {
|
||||
let snapshot = self.catalog.snapshot()?;
|
||||
let generation_lease = snapshot.artifact_leases.iter().find(|lease| {
|
||||
lease.domain_id == authority.lease.domain_id
|
||||
&& lease.generation_id == authority.lease.generation_id
|
||||
});
|
||||
match generation_lease {
|
||||
None => return Ok(()),
|
||||
Some(lease) if *lease == authority.lease => {}
|
||||
Some(_) => {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"catalog generation lease does not match release authority",
|
||||
));
|
||||
}
|
||||
}
|
||||
self.catalog.release_lease_checked(&authority.lease, || {
|
||||
let snapshot = self
|
||||
.domain_store
|
||||
.snapshot()
|
||||
.map_err(|error| CatalogError::Io(io::Error::other(error)))?;
|
||||
let anchor = role(
|
||||
&snapshot.participants,
|
||||
authority.generation_id,
|
||||
ParticipantRole::StateAnchor,
|
||||
)
|
||||
.map_err(|error| CatalogError::Io(io::Error::other(error)))?;
|
||||
let prepared = role(
|
||||
&snapshot.participants,
|
||||
authority.generation_id,
|
||||
ParticipantRole::PreparedLower,
|
||||
)
|
||||
.map_err(|error| CatalogError::Io(io::Error::other(error)))?;
|
||||
if anchor.state() != ParticipantState::Clean
|
||||
|| prepared.state() != ParticipantState::ReleaseIntent
|
||||
{
|
||||
return Err(CatalogError::Io(io::Error::other(
|
||||
"Domain release evidence is not committed",
|
||||
)));
|
||||
}
|
||||
require_mount_absent(&authority.merged)
|
||||
.map_err(|error| CatalogError::Io(io::Error::other(error)))
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn release_authority(
|
||||
domain: &lazaret_core::records::Domain,
|
||||
anchor: &Participant,
|
||||
prepared: &Participant,
|
||||
) -> Result<StateAnchorReleaseAuthority, StateAnchorCoordinatorError> {
|
||||
if !matches!(
|
||||
anchor.state(),
|
||||
ParticipantState::Prepared | ParticipantState::ReleaseIntent
|
||||
) || prepared.state() != ParticipantState::LeaseHeld
|
||||
{
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"participants are not Prepared and LeaseHeld",
|
||||
));
|
||||
}
|
||||
let Some(ParticipantOwnership::StateAnchor(anchor_ownership)) = anchor.ownership() else {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"State Anchor ownership is invalid",
|
||||
));
|
||||
};
|
||||
let Some(ParticipantOwnership::PreparedLower(prepared_ownership)) = prepared.ownership() else {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"prepared-lower ownership is invalid",
|
||||
));
|
||||
};
|
||||
let artifact = prepared_ownership.prepared_artifact_id.ok_or(
|
||||
StateAnchorCoordinatorError::InvalidState("prepared artifact identity is absent"),
|
||||
)?;
|
||||
if anchor_ownership.lower_artifact_id != Some(artifact) {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"State Anchor and prepared lower differ",
|
||||
));
|
||||
}
|
||||
Ok(StateAnchorReleaseAuthority {
|
||||
generation_id: anchor.generation_id(),
|
||||
lease: ArtifactLease {
|
||||
prepared_artifact_id: artifact,
|
||||
domain_id: domain.domain_id,
|
||||
generation_id: anchor.generation_id(),
|
||||
},
|
||||
merged: path_from_record(anchor_ownership.merged_path.as_ref().ok_or(
|
||||
StateAnchorCoordinatorError::InvalidState("State Anchor merged path is absent"),
|
||||
)?)
|
||||
.to_owned(),
|
||||
mount: anchor_ownership
|
||||
.mount
|
||||
.ok_or(StateAnchorCoordinatorError::InvalidState(
|
||||
"State Anchor mount identity is absent",
|
||||
))?,
|
||||
daemon: *anchor_ownership
|
||||
.daemon
|
||||
.serviceable_pidfs()
|
||||
.map_err(|_| StateAnchorCoordinatorError::InvalidState("daemon is not committed"))?,
|
||||
})
|
||||
}
|
||||
|
||||
fn require_absent_role(
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
||||
fn role(
|
||||
participants: &[Participant],
|
||||
generation_id: GenerationId,
|
||||
participant_role: ParticipantRole,
|
||||
) -> Result<&Participant, StateAnchorCoordinatorError> {
|
||||
participants
|
||||
.iter()
|
||||
.find(|participant| {
|
||||
participant.generation_id() == generation_id && participant.role() == participant_role
|
||||
})
|
||||
.ok_or(StateAnchorCoordinatorError::InvalidState(
|
||||
"required participant is absent",
|
||||
))
|
||||
}
|
||||
|
||||
fn participant(
|
||||
generation_id: GenerationId,
|
||||
participant_role: ParticipantRole,
|
||||
state: ParticipantState,
|
||||
ownership: ParticipantOwnership,
|
||||
) -> Result<Participant, StateAnchorCoordinatorError> {
|
||||
Participant::snapshot(generation_id, participant_role, state, Some(ownership)).map_err(|_| {
|
||||
StateAnchorCoordinatorError::InvalidState("participant snapshot violates schema")
|
||||
})
|
||||
}
|
||||
|
||||
fn clean_participant(
|
||||
generation_id: GenerationId,
|
||||
participant_role: ParticipantRole,
|
||||
) -> Result<Participant, StateAnchorCoordinatorError> {
|
||||
participant(
|
||||
generation_id,
|
||||
participant_role,
|
||||
ParticipantState::Clean,
|
||||
ParticipantOwnership::empty(participant_role),
|
||||
)
|
||||
}
|
||||
|
||||
fn prepared_release_intent(
|
||||
prepared: &Participant,
|
||||
) -> Result<Participant, StateAnchorCoordinatorError> {
|
||||
let ownership = match prepared.ownership() {
|
||||
Some(ParticipantOwnership::PreparedLower(ownership)) => PreparedLowerOwnership {
|
||||
lease_state: Some(ParticipantState::ReleaseIntent),
|
||||
..ownership.clone()
|
||||
},
|
||||
_ => {
|
||||
return Err(StateAnchorCoordinatorError::InvalidState(
|
||||
"prepared-lower ownership is invalid",
|
||||
));
|
||||
}
|
||||
};
|
||||
participant(
|
||||
prepared.generation_id(),
|
||||
ParticipantRole::PreparedLower,
|
||||
ParticipantState::ReleaseIntent,
|
||||
ParticipantOwnership::PreparedLower(ownership),
|
||||
)
|
||||
}
|
||||
|
||||
fn path_from_record(path: &lazaret_core::identity::HostPath) -> &Path {
|
||||
Path::new(std::ffi::OsStr::from_bytes(path.as_bytes()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
290
crates/lazaret-host-coordinator/src/state_anchor/tests.rs
Normal file
290
crates/lazaret-host-coordinator/src/state_anchor/tests.rs
Normal file
|
|
@ -0,0 +1,290 @@
|
|||
use std::collections::BTreeMap;
|
||||
use std::fs::{self, OpenOptions};
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
|
||||
use lazaret_core::configuration::{
|
||||
ConfigurationLoadRequest, SandboxLayoutV1, load_trusted_configuration,
|
||||
};
|
||||
use lazaret_core::identity::{
|
||||
BootId, ConfigurationFingerprint, CoordinatorId, DeviceId, GenerationId, HostPath,
|
||||
HostPidLocator, IdMapDigest, InodeId, MountId, MountIdentity, NetworkSidecarId, PidfsDeviceId,
|
||||
PidfsInodeId, PidfsProcessAuthorityV1, PidfsProcessIdentityV1, PolicyId, PrepareOperationId,
|
||||
PreparedArtifactId, ProcessStartTimeTicks, SandboxId, SelectedManifestDigest, SourceIdentity,
|
||||
};
|
||||
use lazaret_core::records::{
|
||||
ArtifactState, Domain, DurableDecision, Generation, ParticipantRole, ParticipantState,
|
||||
PrepareOperation, PrepareStageState, PreparedArtifact, StagedProcessOwnership,
|
||||
};
|
||||
use lazaret_linux::filesystem::try_exclusive_flock;
|
||||
use lazaret_state_anchor::StateAnchorLayout;
|
||||
use lazaret_storage_catalog::Catalog;
|
||||
|
||||
use super::*;
|
||||
|
||||
static SEQUENCE: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
#[test]
|
||||
#[allow(clippy::too_many_lines)]
|
||||
fn durable_phases_resume_release_without_collapsing_lease_order() {
|
||||
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 lock_file = OpenOptions::new()
|
||||
.create(true)
|
||||
.read(true)
|
||||
.truncate(false)
|
||||
.write(true)
|
||||
.open(tree.path.join("domain.lock"))
|
||||
.unwrap();
|
||||
let domain_store = HostCoordinatorDomainStore::open(
|
||||
try_exclusive_flock(lock_file).unwrap(),
|
||||
&tree.path.join("domain.redb"),
|
||||
&domain,
|
||||
)
|
||||
.unwrap();
|
||||
let generation_id = GenerationId::new(7);
|
||||
let coordinator_id = CoordinatorId::new([3; 32]);
|
||||
domain_store
|
||||
.apply(&[DomainStoreMutation::PutGeneration(Generation {
|
||||
domain_id: domain.domain_id,
|
||||
generation_id,
|
||||
configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
|
||||
coordinator_id,
|
||||
network_sidecar_id: NetworkSidecarId::new([4; 32]),
|
||||
sandbox_id: SandboxId::new([5; 32]),
|
||||
durable_decision: DurableDecision::None,
|
||||
})])
|
||||
.unwrap();
|
||||
let layout = StateAnchorLayout::create(&configuration).unwrap();
|
||||
let coordinator = StateAnchorCoordinator::new(&domain_store, &catalog);
|
||||
let lease = coordinator
|
||||
.begin_acquisition(generation_id, &artifact, &layout)
|
||||
.unwrap();
|
||||
assert_eq!(catalog.snapshot().unwrap().artifact_leases, vec![lease]);
|
||||
assert_states(
|
||||
&domain_store,
|
||||
ParticipantState::LeaseHeld,
|
||||
ParticipantState::Intent,
|
||||
);
|
||||
|
||||
let staged = StagedProcessOwnership::v1(
|
||||
domain.domain_id,
|
||||
generation_id,
|
||||
coordinator_id,
|
||||
HostProcessOwnershipSlot::StateAnchorDaemon,
|
||||
process_identity(),
|
||||
);
|
||||
let wrong_coordinator = StagedProcessOwnership::v1(
|
||||
domain.domain_id,
|
||||
generation_id,
|
||||
CoordinatorId::new([99; 32]),
|
||||
HostProcessOwnershipSlot::StateAnchorDaemon,
|
||||
process_identity(),
|
||||
);
|
||||
assert!(coordinator.stage_daemon(wrong_coordinator).is_err());
|
||||
coordinator.stage_daemon(staged).unwrap();
|
||||
let acknowledgment = coordinator.commit_daemon(staged).unwrap();
|
||||
assert!(acknowledgment.acknowledges(&staged));
|
||||
coordinator
|
||||
.commit_prepared(
|
||||
generation_id,
|
||||
artifact.prepared_artifact_id,
|
||||
MountIdentity {
|
||||
mount_id: MountId::new(90),
|
||||
device: DeviceId::new(91),
|
||||
inode: InodeId::new(92),
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_states(
|
||||
&domain_store,
|
||||
ParticipantState::LeaseHeld,
|
||||
ParticipantState::Prepared,
|
||||
);
|
||||
|
||||
let authority = coordinator.begin_release(generation_id).unwrap();
|
||||
assert_states(
|
||||
&domain_store,
|
||||
ParticipantState::LeaseHeld,
|
||||
ParticipantState::ReleaseIntent,
|
||||
);
|
||||
assert_eq!(catalog.snapshot().unwrap().artifact_leases, vec![lease]);
|
||||
let snapshot = domain_store.snapshot().unwrap();
|
||||
let prepared = role(
|
||||
&snapshot.participants,
|
||||
generation_id,
|
||||
ParticipantRole::PreparedLower,
|
||||
)
|
||||
.unwrap();
|
||||
domain_store
|
||||
.apply(&[
|
||||
DomainStoreMutation::CleanParticipantAfterAbsence(
|
||||
clean_participant(generation_id, ParticipantRole::StateAnchor).unwrap(),
|
||||
),
|
||||
DomainStoreMutation::PutParticipant(prepared_release_intent(prepared).unwrap()),
|
||||
])
|
||||
.unwrap();
|
||||
coordinator
|
||||
.complete_release_after_unmount(&authority)
|
||||
.unwrap();
|
||||
assert_states(
|
||||
&domain_store,
|
||||
ParticipantState::Clean,
|
||||
ParticipantState::Clean,
|
||||
);
|
||||
assert!(catalog.snapshot().unwrap().artifact_leases.is_empty());
|
||||
}
|
||||
|
||||
fn assert_states(
|
||||
store: &HostCoordinatorDomainStore,
|
||||
prepared: ParticipantState,
|
||||
anchor: ParticipantState,
|
||||
) {
|
||||
let snapshot = store.snapshot().unwrap();
|
||||
assert_eq!(
|
||||
role(
|
||||
&snapshot.participants,
|
||||
GenerationId::new(7),
|
||||
ParticipantRole::PreparedLower,
|
||||
)
|
||||
.unwrap()
|
||||
.state(),
|
||||
prepared
|
||||
);
|
||||
assert_eq!(
|
||||
role(
|
||||
&snapshot.participants,
|
||||
GenerationId::new(7),
|
||||
ParticipantRole::StateAnchor,
|
||||
)
|
||||
.unwrap()
|
||||
.state(),
|
||||
anchor
|
||||
);
|
||||
}
|
||||
|
||||
fn artifact(path: &Path) -> PreparedArtifact {
|
||||
let metadata = path.metadata().unwrap();
|
||||
PreparedArtifact {
|
||||
prepared_artifact_id: PreparedArtifactId::new([8; 32]),
|
||||
source_identity: SourceIdentity::new(b"fixture".to_vec()),
|
||||
selected_manifest_digest: SelectedManifestDigest::new([9; 32]),
|
||||
policy_id: PolicyId::new([10; 32]),
|
||||
id_map_digest: IdMapDigest::new([11; 32]),
|
||||
path: HostPath::new(path.as_os_str().as_encoded_bytes().to_vec()),
|
||||
device: DeviceId::new(metadata.dev()),
|
||||
inode: InodeId::new(metadata.ino()),
|
||||
state: ArtifactState::Available,
|
||||
}
|
||||
}
|
||||
|
||||
fn publish(catalog: &Catalog, artifact: &PreparedArtifact) {
|
||||
let intent = PrepareOperation {
|
||||
operation_id: PrepareOperationId::new([12; 32]),
|
||||
expected_artifact_id: artifact.prepared_artifact_id,
|
||||
source_identity: artifact.source_identity.clone(),
|
||||
selected_manifest_digest: artifact.selected_manifest_digest,
|
||||
policy_id: artifact.policy_id,
|
||||
id_map_digest: artifact.id_map_digest,
|
||||
stage_path: HostPath::new(b"stages/fixture".to_vec()),
|
||||
stage_state: PrepareStageState::Intent,
|
||||
device: None,
|
||||
inode: None,
|
||||
};
|
||||
catalog.record_prepare(&intent).unwrap();
|
||||
let owned = PrepareOperation {
|
||||
stage_state: PrepareStageState::Owned,
|
||||
device: Some(artifact.device),
|
||||
inode: Some(artifact.inode),
|
||||
..intent.clone()
|
||||
};
|
||||
catalog.bind_prepare_stage(&intent, &owned).unwrap();
|
||||
catalog.publish(&owned, artifact).unwrap();
|
||||
}
|
||||
|
||||
fn process_identity() -> PidfsProcessIdentityV1 {
|
||||
PidfsProcessIdentityV1::new(
|
||||
PidfsProcessAuthorityV1 {
|
||||
boot_id: BootId::new([1; 16]),
|
||||
pidfs_st_dev: PidfsDeviceId::new(2),
|
||||
pidfs_st_ino: PidfsInodeId::new(3),
|
||||
},
|
||||
HostPidLocator::new(4),
|
||||
ProcessStartTimeTicks::new(5),
|
||||
)
|
||||
}
|
||||
|
||||
struct TestTree {
|
||||
path: PathBuf,
|
||||
}
|
||||
|
||||
impl TestTree {
|
||||
fn new() -> Self {
|
||||
let path = std::env::temp_dir().join(format!(
|
||||
"lazaret-state-anchor-durable-{}-{}",
|
||||
std::process::id(),
|
||||
SEQUENCE.fetch_add(1, Ordering::Relaxed)
|
||||
));
|
||||
fs::create_dir(&path).unwrap();
|
||||
Self { path }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestTree {
|
||||
fn drop(&mut self) {
|
||||
fs::remove_dir_all(&self.path).unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
fn configuration(tree: &TestTree) -> lazaret_core::configuration::ResolvedDomainConfiguration {
|
||||
let configuration_directory = tree.path.join("configuration");
|
||||
let workspace = tree.path.join("workspace");
|
||||
let runtime = tree.path.join("runtime");
|
||||
fs::create_dir(&configuration_directory).unwrap();
|
||||
fs::create_dir(&workspace).unwrap();
|
||||
fs::create_dir(&runtime).unwrap();
|
||||
let configuration_path = configuration_directory.join("lazaret.toml");
|
||||
fs::write(
|
||||
&configuration_path,
|
||||
format!(
|
||||
"schema = \"lazaret-trusted-host-config-v1\"\n[storage]\ndata_root = \"{}\"\n[profiles.default]\nimage = \"example.invalid/fixture:v1\"\npersistent_dirs = [\"/usr\"]\n",
|
||||
tree.path.join("data").display()
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
let layout = SandboxLayoutV1::from_production_layout([
|
||||
"/workspace".to_owned(),
|
||||
"/proc".to_owned(),
|
||||
"/sys".to_owned(),
|
||||
"/dev".to_owned(),
|
||||
"/run/lazaret".to_owned(),
|
||||
])
|
||||
.unwrap();
|
||||
load_trusted_configuration(ConfigurationLoadRequest {
|
||||
explicit_configuration_path: Some(configuration_path),
|
||||
selected_profile: "default".to_owned(),
|
||||
workspace,
|
||||
home: tree.path.join("home"),
|
||||
xdg_config_home: None,
|
||||
xdg_data_home: None,
|
||||
xdg_runtime_dir: runtime,
|
||||
environment: BTreeMap::new(),
|
||||
sandbox_layout: layout,
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
Loading…
Reference in a new issue