ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Unify host coordinator lifecycle state

Browse source
This commit is contained in:
ms2k 2026-07-19 04:17:25 +09:00
commit 2f8a9f9178
6 changed files with 720 additions and 337 deletions
Download patch file Download diff file Expand all files Collapse all files

511
crates/lazaret-host-coordinator/src/lifecycle.rs
View file

@ -1,180 +1,22 @@
use std::collections::{HashMap, HashSet};
use std::fmt;
use lazaret_core::identity::{ConfigurationFingerprint, SessionId}; use lazaret_core::identity::{ConfigurationFingerprint, SessionId};
use lazaret_core::records::{DurableDecision, Generation, Session, SessionPhase}; use lazaret_core::records::{DurableDecision, Generation, Session, SessionPhase};
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore}; use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
#[derive(Clone, Copy, Debug, Eq, PartialEq)] mod contracts;
pub enum LifecycleAction { mod state;
StartNewDomain, pub use contracts::{
StartReadySession, LifecycleAction, LifecycleAdapter, LifecycleError, SessionAdmissionIdentity,
WaitForDomain, SessionLifecycleAdapter,
RestartWaitingDomain, };
CreateTrustAnchor, pub use state::DomainPhase;
StartRuntimeSupervisor, use state::{LifecycleAggregate, RuntimeMilestone, RuntimeProgress};
LeasePreparedLower,
MountStateAnchor,
StartSidecar,
StartConnectivityHelper,
MarkNetworkReady,
ArmRuntimeCausalEdge,
StartSandbox,
PublishDomain,
AdvanceWaitingSession,
RejectConfigurationMismatch,
RegisterSession,
DisconnectSession,
BeginTeardown,
StopSandbox,
StopSidecar,
UnmountStateAnchor,
ReleasePreparedLowerLease,
GracefulSupervisorExit,
FinishTeardown,
}
pub trait LifecycleAdapter {
type Error;
/// # Errors
/// Returns the adapter's physical or durable action failure.
fn execute(
&mut self,
action: LifecycleAction,
generation: &Generation,
) -> Result<(), Self::Error>;
/// Completes a physical boundary that must follow its durable lifecycle commit.
///
/// # Errors
/// Returns the adapter's post-commit publication or cleanup failure.
fn after_commit(
&mut self,
_action: LifecycleAction,
_generation: &Generation,
) -> Result<(), Self::Error> {
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SessionAdmissionIdentity {
pub generation_id: lazaret_core::identity::GenerationId,
pub session_id: SessionId,
pub configuration_fingerprint: ConfigurationFingerprint,
}
pub trait SessionLifecycleAdapter {
type Error;
type Admission;
fn admission_identity(&self, admission: &Self::Admission) -> SessionAdmissionIdentity;
/// # Errors
/// Returns credential, identity, duplicate-session, or durable registration failure.
fn register_session(
&self,
store: &HostCoordinatorDomainStore,
entry_action: LifecycleAction,
admission: Self::Admission,
) -> Result<Session, Self::Error>;
/// # Errors
/// Returns exact-session or durable disconnect failure.
fn disconnect_session(
&self,
store: &HostCoordinatorDomainStore,
session: &Session,
) -> Result<(), Self::Error>;
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DomainPhase {
Absent,
Starting,
Ready,
Stopping,
Failed,
}
#[derive(Debug)]
pub enum LifecycleError<E> {
Adapter(E),
Store(DomainStoreError),
InvalidTransition {
action: LifecycleAction,
reason: &'static str,
},
}
impl<E: fmt::Display> fmt::Display for LifecycleError<E> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Adapter(error) => write!(formatter, "lifecycle adapter failed: {error}"),
Self::Store(error) => {
write!(formatter, "Domain Store lifecycle update failed: {error}")
}
Self::InvalidTransition { action, reason } => {
write!(formatter, "{action:?} refused: {reason}")
}
}
}
}
impl<E: std::error::Error + 'static> std::error::Error for LifecycleError<E> {}
impl<E> From<DomainStoreError> for LifecycleError<E> {
fn from(error: DomainStoreError) -> Self {
Self::Store(error)
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
enum RuntimeMilestone {
TrustAnchor,
RuntimeSupervisor,
PreparedLowerLease,
StateAnchor,
Sidecar,
ConnectivityHelper,
NetworkReady,
CausalEdge,
Sandbox,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct RuntimeProgress {
completed: HashSet<RuntimeMilestone>,
}
impl RuntimeProgress {
fn has(&self, milestone: RuntimeMilestone) -> bool {
self.completed.contains(&milestone)
}
fn mark(&mut self, milestone: RuntimeMilestone) {
self.completed.insert(milestone);
}
fn clear(&mut self, milestone: RuntimeMilestone) {
self.completed.remove(&milestone);
}
}
pub struct HostCoordinatorLifecycle<'store, A, S> { pub struct HostCoordinatorLifecycle<'store, A, S> {
store: &'store HostCoordinatorDomainStore, store: &'store HostCoordinatorDomainStore,
adapter: A, adapter: A,
session_adapter: S, session_adapter: S,
phase: DomainPhase, state: LifecycleAggregate,
generation: Option<Generation>,
bootstrap_owner: Option<SessionId>,
progress: RuntimeProgress,
published: bool,
sessions: HashMap<SessionId, Session>,
connected: HashSet<SessionId>,
registered: HashSet<SessionId>,
} }
impl<'store, A, S> HostCoordinatorLifecycle<'store, A, S> impl<'store, A, S> HostCoordinatorLifecycle<'store, A, S>
@ -188,37 +30,31 @@ where
store, store,
adapter, adapter,
session_adapter, session_adapter,
phase: DomainPhase::Absent, state: LifecycleAggregate::absent(),
generation: None,
bootstrap_owner: None,
progress: RuntimeProgress::default(),
published: false,
sessions: HashMap::new(),
connected: HashSet::new(),
registered: HashSet::new(),
} }
} }
#[must_use] #[must_use]
pub const fn phase(&self) -> DomainPhase { pub const fn phase(&self) -> DomainPhase {
self.phase self.state.phase
} }
#[must_use] #[must_use]
pub const fn is_published(&self) -> bool { pub const fn is_published(&self) -> bool {
self.published self.state.published
} }
#[must_use] #[must_use]
pub fn session_phase(&self, session_id: SessionId) -> Option<SessionPhase> { pub fn session_phase(&self, session_id: SessionId) -> Option<SessionPhase> {
self.sessions self.state
.sessions
.get(&session_id) .get(&session_id)
.map(|session| session.session_phase) .map(|session| session.session_phase)
} }
#[must_use] #[must_use]
pub fn registered_sessions(&self) -> usize { pub fn registered_sessions(&self) -> usize {
self.registered.len() self.state.registered_sessions()
} }
#[must_use] #[must_use]
@ -240,7 +76,7 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
LifecycleAction::StartNewDomain, LifecycleAction::StartNewDomain,
self.phase == DomainPhase::Absent, self.state.phase == DomainPhase::Absent,
"Domain is not absent", "Domain is not absent",
)?; )?;
Self::require( Self::require(
@ -265,12 +101,13 @@ where
DomainStoreMutation::PutGeneration(generation.clone()), DomainStoreMutation::PutGeneration(generation.clone()),
DomainStoreMutation::PutSession(session.clone()), DomainStoreMutation::PutSession(session.clone()),
])?; ])?;
self.phase = DomainPhase::Starting; self.transition(LifecycleAction::StartNewDomain, move |state| {
self.bootstrap_owner = Some(session.session_id); state.phase = DomainPhase::Starting;
self.sessions.insert(session.session_id, session); state.bootstrap_owner = Some(session.session_id);
self.generation = Some(generation); state.sessions.insert(session.session_id, session);
self.progress = RuntimeProgress::default(); state.generation = Some(generation);
Ok(()) state.progress = RuntimeProgress::default();
})
} }
/// Drives one cold-start leader through the complete normal publication boundary. /// Drives one cold-start leader through the complete normal publication boundary.
@ -365,7 +202,7 @@ where
admission: S::Admission, admission: S::Admission,
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
let identity = self.session_adapter.admission_identity(&admission); let identity = self.session_adapter.admission_identity(&admission);
let entry_action = match self.sessions.get(&identity.session_id) { let entry_action = match self.state.sessions.get(&identity.session_id) {
None => LifecycleAction::StartReadySession, None => LifecycleAction::StartReadySession,
Some(session) if session.session_phase == SessionPhase::Waiting => { Some(session) if session.session_phase == SessionPhase::Waiting => {
LifecycleAction::AdvanceWaitingSession LifecycleAction::AdvanceWaitingSession
@ -396,10 +233,9 @@ where
&& attached.session_phase == SessionPhase::Attached, && attached.session_phase == SessionPhase::Attached,
"session adapter returned mismatched registration evidence", "session adapter returned mismatched registration evidence",
)?; )?;
self.connected.insert(attached.session_id); self.transition(LifecycleAction::RegisterSession, move |state| {
self.registered.insert(attached.session_id); state.sessions.insert(attached.session_id, attached);
self.sessions.insert(attached.session_id, attached); })
Ok(())
} }
/// # Errors /// # Errors
@ -407,7 +243,10 @@ where
pub fn wait_for_domain(&mut self, session: Session) -> Result<(), LifecycleError<A::Error>> { pub fn wait_for_domain(&mut self, session: Session) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
LifecycleAction::WaitForDomain, LifecycleAction::WaitForDomain,
matches!(self.phase, DomainPhase::Starting | DomainPhase::Stopping), matches!(
self.state.phase,
DomainPhase::Starting | DomainPhase::Stopping
),
"Domain is neither starting nor stopping", "Domain is neither starting nor stopping",
)?; )?;
Self::require( Self::require(
@ -421,7 +260,7 @@ where
== self == self
.current_generation(LifecycleAction::WaitForDomain)? .current_generation(LifecycleAction::WaitForDomain)?
.generation_id .generation_id
&& !self.sessions.contains_key(&session.session_id), && !self.state.sessions.contains_key(&session.session_id),
"session generation differs or Session ID is already known", "session generation differs or Session ID is already known",
)?; )?;
self.execute_current(LifecycleAction::WaitForDomain)?; self.execute_current(LifecycleAction::WaitForDomain)?;
@ -429,7 +268,6 @@ where
LifecycleAction::WaitForDomain, LifecycleAction::WaitForDomain,
session, session,
SessionPhase::Waiting, SessionPhase::Waiting,
false,
) )
} }
@ -442,7 +280,7 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
LifecycleAction::RestartWaitingDomain, LifecycleAction::RestartWaitingDomain,
self.phase == DomainPhase::Absent, self.state.phase == DomainPhase::Absent,
"Domain is not absent", "Domain is not absent",
)?; )?;
Self::require( Self::require(
@ -469,6 +307,7 @@ where
)?; )?;
self.execute(LifecycleAction::RestartWaitingDomain, &generation)?; self.execute(LifecycleAction::RestartWaitingDomain, &generation)?;
let waiters = self let waiters = self
.state
.sessions .sessions
.values() .values()
.filter(|candidate| candidate.session_phase == SessionPhase::Waiting) .filter(|candidate| candidate.session_phase == SessionPhase::Waiting)
@ -489,16 +328,17 @@ where
})); }));
} }
self.store.apply(&mutations)?; self.store.apply(&mutations)?;
self.phase = DomainPhase::Starting;
self.bootstrap_owner = Some(session_id);
let generation_id = generation.generation_id; let generation_id = generation.generation_id;
self.generation = Some(generation); self.transition(LifecycleAction::RestartWaitingDomain, move |state| {
self.progress = RuntimeProgress::default(); state.phase = DomainPhase::Starting;
for mut waiter in waiters { state.bootstrap_owner = Some(session_id);
waiter.generation_id = generation_id; state.generation = Some(generation);
self.sessions.insert(waiter.session_id, waiter); state.progress = RuntimeProgress::default();
} for mut waiter in waiters {
Ok(()) waiter.generation_id = generation_id;
state.sessions.insert(waiter.session_id, waiter);
}
})
} }
/// # Errors /// # Errors
@ -506,7 +346,7 @@ where
pub fn create_trust_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn create_trust_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::CreateTrustAnchor, LifecycleAction::CreateTrustAnchor,
!self.progress.has(RuntimeMilestone::TrustAnchor), !self.state.progress.has(RuntimeMilestone::TrustAnchor),
|p| p.mark(RuntimeMilestone::TrustAnchor), |p| p.mark(RuntimeMilestone::TrustAnchor),
) )
} }
@ -516,8 +356,8 @@ where
pub fn start_runtime_supervisor(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn start_runtime_supervisor(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::StartRuntimeSupervisor, LifecycleAction::StartRuntimeSupervisor,
self.progress.has(RuntimeMilestone::TrustAnchor) self.state.progress.has(RuntimeMilestone::TrustAnchor)
&& !self.progress.has(RuntimeMilestone::RuntimeSupervisor), && !self.state.progress.has(RuntimeMilestone::RuntimeSupervisor),
|p| p.mark(RuntimeMilestone::RuntimeSupervisor), |p| p.mark(RuntimeMilestone::RuntimeSupervisor),
) )
} }
@ -527,8 +367,11 @@ where
pub fn lease_prepared_lower(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn lease_prepared_lower(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::LeasePreparedLower, LifecycleAction::LeasePreparedLower,
self.progress.has(RuntimeMilestone::RuntimeSupervisor) self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
&& !self.progress.has(RuntimeMilestone::PreparedLowerLease), && !self
.state
.progress
.has(RuntimeMilestone::PreparedLowerLease),
|p| p.mark(RuntimeMilestone::PreparedLowerLease), |p| p.mark(RuntimeMilestone::PreparedLowerLease),
) )
} }
@ -538,8 +381,10 @@ where
pub fn mount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn mount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::MountStateAnchor, LifecycleAction::MountStateAnchor,
self.progress.has(RuntimeMilestone::PreparedLowerLease) self.state
&& !self.progress.has(RuntimeMilestone::StateAnchor), .progress
.has(RuntimeMilestone::PreparedLowerLease)
&& !self.state.progress.has(RuntimeMilestone::StateAnchor),
|p| p.mark(RuntimeMilestone::StateAnchor), |p| p.mark(RuntimeMilestone::StateAnchor),
) )
} }
@ -549,8 +394,8 @@ where
pub fn start_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn start_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::StartSidecar, LifecycleAction::StartSidecar,
self.progress.has(RuntimeMilestone::StateAnchor) self.state.progress.has(RuntimeMilestone::StateAnchor)
&& !self.progress.has(RuntimeMilestone::Sidecar), && !self.state.progress.has(RuntimeMilestone::Sidecar),
|p| p.mark(RuntimeMilestone::Sidecar), |p| p.mark(RuntimeMilestone::Sidecar),
) )
} }
@ -560,8 +405,11 @@ where
pub fn start_connectivity_helper(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn start_connectivity_helper(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::StartConnectivityHelper, LifecycleAction::StartConnectivityHelper,
self.progress.has(RuntimeMilestone::Sidecar) self.state.progress.has(RuntimeMilestone::Sidecar)
&& !self.progress.has(RuntimeMilestone::ConnectivityHelper), && !self
.state
.progress
.has(RuntimeMilestone::ConnectivityHelper),
|p| p.mark(RuntimeMilestone::ConnectivityHelper), |p| p.mark(RuntimeMilestone::ConnectivityHelper),
) )
} }
@ -571,8 +419,10 @@ where
pub fn mark_network_ready(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn mark_network_ready(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::MarkNetworkReady, LifecycleAction::MarkNetworkReady,
self.progress.has(RuntimeMilestone::ConnectivityHelper) self.state
&& !self.progress.has(RuntimeMilestone::NetworkReady), .progress
.has(RuntimeMilestone::ConnectivityHelper)
&& !self.state.progress.has(RuntimeMilestone::NetworkReady),
|p| p.mark(RuntimeMilestone::NetworkReady), |p| p.mark(RuntimeMilestone::NetworkReady),
) )
} }
@ -582,8 +432,8 @@ where
pub fn arm_runtime_causal_edge(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn arm_runtime_causal_edge(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.startup_action( self.startup_action(
LifecycleAction::ArmRuntimeCausalEdge, LifecycleAction::ArmRuntimeCausalEdge,
self.progress.has(RuntimeMilestone::NetworkReady) self.state.progress.has(RuntimeMilestone::NetworkReady)
&& !self.progress.has(RuntimeMilestone::CausalEdge), && !self.state.progress.has(RuntimeMilestone::CausalEdge),
|p| p.mark(RuntimeMilestone::CausalEdge), |p| p.mark(RuntimeMilestone::CausalEdge),
) )
} }
@ -591,12 +441,12 @@ where
/// # Errors /// # Errors
/// Refuses launch before network readiness and the causal edge are established. /// Refuses launch before network readiness and the causal edge are established.
pub fn start_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn start_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> {
let allowed = self.progress.has(RuntimeMilestone::CausalEdge) let allowed = self.state.progress.has(RuntimeMilestone::CausalEdge)
&& self.progress.has(RuntimeMilestone::NetworkReady) && self.state.progress.has(RuntimeMilestone::NetworkReady)
&& !self.progress.has(RuntimeMilestone::Sandbox); && !self.state.progress.has(RuntimeMilestone::Sandbox);
Self::require( Self::require(
LifecycleAction::StartSandbox, LifecycleAction::StartSandbox,
self.phase == DomainPhase::Starting && allowed, self.state.phase == DomainPhase::Starting && allowed,
"runtime prerequisites are incomplete", "runtime prerequisites are incomplete",
)?; )?;
let mut generation = self let mut generation = self
@ -605,25 +455,35 @@ where
generation.durable_decision = DurableDecision::Preparing; generation.durable_decision = DurableDecision::Preparing;
self.store self.store
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?; .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
self.generation = Some(generation.clone()); self.transition(LifecycleAction::StartSandbox, {
let generation = generation.clone();
move |state| state.generation = Some(generation)
})?;
self.execute(LifecycleAction::StartSandbox, &generation)?; self.execute(LifecycleAction::StartSandbox, &generation)?;
self.progress.mark(RuntimeMilestone::Sandbox); self.transition(LifecycleAction::StartSandbox, |state| {
Ok(()) state.progress.mark(RuntimeMilestone::Sandbox);
})
} }
/// # Errors /// # Errors
/// Refuses publication until every ordered startup prerequisite is complete. /// Refuses publication until every ordered startup prerequisite is complete.
pub fn publish_domain(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn publish_domain(&mut self) -> Result<(), LifecycleError<A::Error>> {
let allowed = self.phase == DomainPhase::Starting let allowed = self.state.phase == DomainPhase::Starting
&& self.progress.has(RuntimeMilestone::TrustAnchor) && self.state.progress.has(RuntimeMilestone::TrustAnchor)
&& self.progress.has(RuntimeMilestone::RuntimeSupervisor) && self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
&& self.progress.has(RuntimeMilestone::PreparedLowerLease) && self
&& self.progress.has(RuntimeMilestone::StateAnchor) .state
&& self.progress.has(RuntimeMilestone::Sidecar) .progress
&& self.progress.has(RuntimeMilestone::ConnectivityHelper) .has(RuntimeMilestone::PreparedLowerLease)
&& self.progress.has(RuntimeMilestone::NetworkReady) && self.state.progress.has(RuntimeMilestone::StateAnchor)
&& self.progress.has(RuntimeMilestone::CausalEdge) && self.state.progress.has(RuntimeMilestone::Sidecar)
&& self.progress.has(RuntimeMilestone::Sandbox); && self
.state
.progress
.has(RuntimeMilestone::ConnectivityHelper)
&& self.state.progress.has(RuntimeMilestone::NetworkReady)
&& self.state.progress.has(RuntimeMilestone::CausalEdge)
&& self.state.progress.has(RuntimeMilestone::Sandbox);
Self::require( Self::require(
LifecycleAction::PublishDomain, LifecycleAction::PublishDomain,
allowed, allowed,
@ -636,20 +496,24 @@ where
generation.durable_decision = DurableDecision::Published; generation.durable_decision = DurableDecision::Published;
self.store self.store
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?; .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
self.generation = Some(generation.clone());
if let Err(error) = self if let Err(error) = self
.adapter .adapter
.after_commit(LifecycleAction::PublishDomain, &generation) .after_commit(LifecycleAction::PublishDomain, &generation)
{ {
self.phase = DomainPhase::Failed; self.transition(LifecycleAction::PublishDomain, move |state| {
self.bootstrap_owner = None; state.generation = Some(generation);
self.published = false; state.phase = DomainPhase::Failed;
state.bootstrap_owner = None;
state.published = false;
})?;
return Err(LifecycleError::Adapter(error)); return Err(LifecycleError::Adapter(error));
} }
self.phase = DomainPhase::Ready; self.transition(LifecycleAction::PublishDomain, move |state| {
self.bootstrap_owner = None; state.generation = Some(generation);
self.published = true; state.phase = DomainPhase::Ready;
Ok(()) state.bootstrap_owner = None;
state.published = true;
})
} }
/// # Errors /// # Errors
@ -661,7 +525,7 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
LifecycleAction::RejectConfigurationMismatch, LifecycleAction::RejectConfigurationMismatch,
self.phase == DomainPhase::Ready && session.session_phase == SessionPhase::New, self.state.phase == DomainPhase::Ready && session.session_phase == SessionPhase::New,
"session is not eligible for mismatch refusal", "session is not eligible for mismatch refusal",
)?; )?;
let generation = self.current_generation(LifecycleAction::RejectConfigurationMismatch)?; let generation = self.current_generation(LifecycleAction::RejectConfigurationMismatch)?;
@ -673,7 +537,7 @@ where
)?; )?;
Self::require( Self::require(
LifecycleAction::RejectConfigurationMismatch, LifecycleAction::RejectConfigurationMismatch,
!self.sessions.contains_key(&session.session_id), !self.state.sessions.contains_key(&session.session_id),
"Session ID is already known", "Session ID is already known",
)?; )?;
self.execute_current(LifecycleAction::RejectConfigurationMismatch)?; self.execute_current(LifecycleAction::RejectConfigurationMismatch)?;
@ -683,8 +547,9 @@ where
DomainStoreMutation::PutSession(new), DomainStoreMutation::PutSession(new),
DomainStoreMutation::PutSession(session.clone()), DomainStoreMutation::PutSession(session.clone()),
])?; ])?;
self.sessions.insert(session.session_id, session); self.transition(LifecycleAction::RejectConfigurationMismatch, move |state| {
Ok(()) state.sessions.insert(session.session_id, session);
})
} }
/// # Errors /// # Errors
@ -699,7 +564,7 @@ where
.clone(); .clone();
Self::require( Self::require(
LifecycleAction::RejectConfigurationMismatch, LifecycleAction::RejectConfigurationMismatch,
self.phase == DomainPhase::Ready self.state.phase == DomainPhase::Ready
&& matches!( && matches!(
session.session_phase, session.session_phase,
SessionPhase::New | SessionPhase::Waiting SessionPhase::New | SessionPhase::Waiting
@ -716,10 +581,9 @@ where
session.session_phase = SessionPhase::Failed; session.session_phase = SessionPhase::Failed;
self.store self.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])?; .apply(&[DomainStoreMutation::PutSession(session.clone())])?;
self.connected.remove(&session_id); self.transition(LifecycleAction::RejectConfigurationMismatch, move |state| {
self.registered.remove(&session_id); state.sessions.insert(session_id, session);
self.sessions.insert(session_id, session); })
Ok(())
} }
/// # Errors /// # Errors
@ -743,38 +607,25 @@ where
), ),
"session is not live", "session is not live",
)?; )?;
let leader_lost = let leader_lost = self.state.phase == DomainPhase::Starting
self.phase == DomainPhase::Starting && self.bootstrap_owner == Some(session_id); && self.state.bootstrap_owner == Some(session_id);
self.session_adapter self.session_adapter
.disconnect_session(self.store, &session) .disconnect_session(self.store, &session)
.map_err(LifecycleError::Adapter)?; .map_err(LifecycleError::Adapter)?;
session.session_phase = SessionPhase::Closed; session.session_phase = SessionPhase::Closed;
self.connected.remove(&session_id); self.transition(LifecycleAction::DisconnectSession, move |state| {
self.registered.remove(&session_id); state.sessions.insert(session_id, session);
self.sessions.insert(session_id, session); if leader_lost {
if leader_lost { state.phase = DomainPhase::Failed;
self.phase = DomainPhase::Failed; state.bootstrap_owner = None;
self.bootstrap_owner = None; state.published = false;
self.published = false; }
} })
Ok(())
} }
#[must_use] #[must_use]
pub fn is_teardown_ready(&self) -> bool { pub fn is_teardown_ready(&self) -> bool {
self.phase == DomainPhase::Ready self.state.phase == DomainPhase::Ready && !self.state.has_session_interest()
&& self.registered.is_empty()
&& self.connected.is_empty()
&& self.sessions.values().all(|session| {
!matches!(
session.session_phase,
SessionPhase::Waiting
| SessionPhase::Registering
| SessionPhase::Attached
| SessionPhase::Prompting
| SessionPhase::HostExecuting
)
})
} }
/// Performs the complete ordered normal teardown when the final session has disconnected. /// Performs the complete ordered normal teardown when the final session has disconnected.
@ -806,7 +657,7 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
LifecycleAction::DisconnectSession, LifecycleAction::DisconnectSession,
self.phase == DomainPhase::Ready && self.published, self.state.phase == DomainPhase::Ready && self.state.published,
"Domain is not published", "Domain is not published",
)?; )?;
let phase = self let phase = self
@ -828,9 +679,10 @@ where
"sessions still have interest", "sessions still have interest",
)?; )?;
self.execute_current(LifecycleAction::BeginTeardown)?; self.execute_current(LifecycleAction::BeginTeardown)?;
self.phase = DomainPhase::Stopping; self.transition(LifecycleAction::BeginTeardown, |state| {
self.published = false; state.phase = DomainPhase::Stopping;
Ok(()) state.published = false;
})
} }
/// # Errors /// # Errors
@ -838,7 +690,7 @@ where
pub fn stop_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn stop_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.teardown_action( self.teardown_action(
LifecycleAction::StopSandbox, LifecycleAction::StopSandbox,
self.progress.has(RuntimeMilestone::Sandbox), self.state.progress.has(RuntimeMilestone::Sandbox),
|p| { |p| {
p.clear(RuntimeMilestone::Sandbox); p.clear(RuntimeMilestone::Sandbox);
p.clear(RuntimeMilestone::CausalEdge); p.clear(RuntimeMilestone::CausalEdge);
@ -851,8 +703,8 @@ where
pub fn stop_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn stop_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.teardown_action( self.teardown_action(
LifecycleAction::StopSidecar, LifecycleAction::StopSidecar,
!self.progress.has(RuntimeMilestone::Sandbox) !self.state.progress.has(RuntimeMilestone::Sandbox)
&& self.progress.has(RuntimeMilestone::Sidecar), && self.state.progress.has(RuntimeMilestone::Sidecar),
|p| { |p| {
p.clear(RuntimeMilestone::Sidecar); p.clear(RuntimeMilestone::Sidecar);
p.clear(RuntimeMilestone::ConnectivityHelper); p.clear(RuntimeMilestone::ConnectivityHelper);
@ -866,8 +718,8 @@ where
pub fn unmount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn unmount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.teardown_action( self.teardown_action(
LifecycleAction::UnmountStateAnchor, LifecycleAction::UnmountStateAnchor,
!self.progress.has(RuntimeMilestone::Sidecar) !self.state.progress.has(RuntimeMilestone::Sidecar)
&& self.progress.has(RuntimeMilestone::StateAnchor), && self.state.progress.has(RuntimeMilestone::StateAnchor),
|p| p.clear(RuntimeMilestone::StateAnchor), |p| p.clear(RuntimeMilestone::StateAnchor),
) )
} }
@ -877,8 +729,11 @@ where
pub fn release_prepared_lower_lease(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn release_prepared_lower_lease(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.teardown_action( self.teardown_action(
LifecycleAction::ReleasePreparedLowerLease, LifecycleAction::ReleasePreparedLowerLease,
!self.progress.has(RuntimeMilestone::StateAnchor) !self.state.progress.has(RuntimeMilestone::StateAnchor)
&& self.progress.has(RuntimeMilestone::PreparedLowerLease), && self
.state
.progress
.has(RuntimeMilestone::PreparedLowerLease),
|p| p.clear(RuntimeMilestone::PreparedLowerLease), |p| p.clear(RuntimeMilestone::PreparedLowerLease),
) )
} }
@ -888,8 +743,11 @@ where
pub fn graceful_supervisor_exit(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn graceful_supervisor_exit(&mut self) -> Result<(), LifecycleError<A::Error>> {
self.teardown_action( self.teardown_action(
LifecycleAction::GracefulSupervisorExit, LifecycleAction::GracefulSupervisorExit,
!self.progress.has(RuntimeMilestone::PreparedLowerLease) !self
&& self.progress.has(RuntimeMilestone::RuntimeSupervisor), .state
.progress
.has(RuntimeMilestone::PreparedLowerLease)
&& self.state.progress.has(RuntimeMilestone::RuntimeSupervisor),
|p| p.clear(RuntimeMilestone::RuntimeSupervisor), |p| p.clear(RuntimeMilestone::RuntimeSupervisor),
) )
} }
@ -897,12 +755,15 @@ where
/// # Errors /// # Errors
/// Refuses while runtime resources remain or the final durable update fails. /// Refuses while runtime resources remain or the final durable update fails.
pub fn finish_teardown(&mut self) -> Result<(), LifecycleError<A::Error>> { pub fn finish_teardown(&mut self) -> Result<(), LifecycleError<A::Error>> {
let allowed = self.phase == DomainPhase::Stopping let allowed = self.state.phase == DomainPhase::Stopping
&& !self.progress.has(RuntimeMilestone::RuntimeSupervisor) && !self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
&& !self.progress.has(RuntimeMilestone::PreparedLowerLease) && !self
&& !self.progress.has(RuntimeMilestone::StateAnchor) .state
&& !self.progress.has(RuntimeMilestone::Sidecar) .progress
&& !self.progress.has(RuntimeMilestone::Sandbox); .has(RuntimeMilestone::PreparedLowerLease)
&& !self.state.progress.has(RuntimeMilestone::StateAnchor)
&& !self.state.progress.has(RuntimeMilestone::Sidecar)
&& !self.state.progress.has(RuntimeMilestone::Sandbox);
Self::require( Self::require(
LifecycleAction::FinishTeardown, LifecycleAction::FinishTeardown,
allowed, allowed,
@ -915,11 +776,12 @@ where
generation.durable_decision = DurableDecision::Clean; generation.durable_decision = DurableDecision::Clean;
self.store self.store
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?; .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
self.generation = Some(generation); self.transition(LifecycleAction::FinishTeardown, move |state| {
self.phase = DomainPhase::Absent; state.generation = Some(generation);
self.progress = RuntimeProgress::default(); state.phase = DomainPhase::Absent;
self.bootstrap_owner = None; state.progress = RuntimeProgress::default();
Ok(()) state.bootstrap_owner = None;
})
} }
pub fn into_adapters(self) -> (A, S) { pub fn into_adapters(self) -> (A, S) {
@ -934,12 +796,11 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
action, action,
self.phase == DomainPhase::Starting && allowed, self.state.phase == DomainPhase::Starting && allowed,
"startup ordering refused", "startup ordering refused",
)?; )?;
self.execute_current(action)?; self.execute_current(action)?;
update(&mut self.progress); self.transition(action, |state| update(&mut state.progress))
Ok(())
} }
fn teardown_action( fn teardown_action(
@ -950,12 +811,11 @@ where
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
action, action,
self.phase == DomainPhase::Stopping && allowed, self.state.phase == DomainPhase::Stopping && allowed,
"teardown ordering refused", "teardown ordering refused",
)?; )?;
self.execute_current(action)?; self.execute_current(action)?;
update(&mut self.progress); self.transition(action, |state| update(&mut state.progress))
Ok(())
} }
fn put_new_session( fn put_new_session(
@ -963,7 +823,6 @@ where
action: LifecycleAction, action: LifecycleAction,
mut session: Session, mut session: Session,
phase: SessionPhase, phase: SessionPhase,
connected: bool,
) -> Result<(), LifecycleError<A::Error>> { ) -> Result<(), LifecycleError<A::Error>> {
Self::require( Self::require(
action, action,
@ -972,17 +831,15 @@ where
)?; )?;
Self::require( Self::require(
action, action,
!self.sessions.contains_key(&session.session_id), !self.state.sessions.contains_key(&session.session_id),
"Session ID is already known", "Session ID is already known",
)?; )?;
session.session_phase = phase; session.session_phase = phase;
self.store self.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])?; .apply(&[DomainStoreMutation::PutSession(session.clone())])?;
if connected { self.transition(action, move |state| {
self.connected.insert(session.session_id); state.sessions.insert(session.session_id, session);
} })
self.sessions.insert(session.session_id, session);
Ok(())
} }
fn require_new_generation( fn require_new_generation(
@ -1010,14 +867,15 @@ where
} }
fn admission_ready(&self) -> bool { fn admission_ready(&self) -> bool {
self.phase == DomainPhase::Ready && self.published self.state.phase == DomainPhase::Ready && self.state.published
} }
fn current_generation( fn current_generation(
&self, &self,
action: LifecycleAction, action: LifecycleAction,
) -> Result<&Generation, LifecycleError<A::Error>> { ) -> Result<&Generation, LifecycleError<A::Error>> {
self.generation self.state
.generation
.as_ref() .as_ref()
.ok_or(LifecycleError::InvalidTransition { .ok_or(LifecycleError::InvalidTransition {
action, action,
@ -1030,7 +888,8 @@ where
session_id: SessionId, session_id: SessionId,
action: LifecycleAction, action: LifecycleAction,
) -> Result<&Session, LifecycleError<A::Error>> { ) -> Result<&Session, LifecycleError<A::Error>> {
self.sessions self.state
.sessions
.get(&session_id) .get(&session_id)
.ok_or(LifecycleError::InvalidTransition { .ok_or(LifecycleError::InvalidTransition {
action, action,
@ -1053,6 +912,22 @@ where
.map_err(LifecycleError::Adapter) .map_err(LifecycleError::Adapter)
} }
fn transition(
&mut self,
action: LifecycleAction,
update: impl FnOnce(&mut LifecycleAggregate),
) -> Result<(), LifecycleError<A::Error>> {
self.state
.validate()
.map_err(|reason| LifecycleError::InvalidAggregate { action, reason })?;
let mut next = self.state.clone();
update(&mut next);
next.validate()
.map_err(|reason| LifecycleError::InvalidAggregate { action, reason })?;
self.state = next;
Ok(())
}
fn require( fn require(
action: LifecycleAction, action: LifecycleAction,
condition: bool, condition: bool,

132
crates/lazaret-host-coordinator/src/lifecycle/contracts.rs Normal file
View file

@ -0,0 +1,132 @@
use std::fmt;
use lazaret_core::identity::{ConfigurationFingerprint, GenerationId, SessionId};
use lazaret_core::records::{Generation, Session};
use crate::domain_store::{DomainStoreError, HostCoordinatorDomainStore};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LifecycleAction {
StartNewDomain,
StartReadySession,
WaitForDomain,
RestartWaitingDomain,
CreateTrustAnchor,
StartRuntimeSupervisor,
LeasePreparedLower,
MountStateAnchor,
StartSidecar,
StartConnectivityHelper,
MarkNetworkReady,
ArmRuntimeCausalEdge,
StartSandbox,
PublishDomain,
AdvanceWaitingSession,
RejectConfigurationMismatch,
RegisterSession,
DisconnectSession,
BeginTeardown,
StopSandbox,
StopSidecar,
UnmountStateAnchor,
ReleasePreparedLowerLease,
GracefulSupervisorExit,
FinishTeardown,
}
pub trait LifecycleAdapter {
type Error;
/// # Errors
/// Returns the adapter's physical or durable action failure.
fn execute(
&mut self,
action: LifecycleAction,
generation: &Generation,
) -> Result<(), Self::Error>;
/// Completes a physical boundary that must follow its durable lifecycle commit.
///
/// # Errors
/// Returns the adapter's post-commit publication or cleanup failure.
fn after_commit(
&mut self,
_action: LifecycleAction,
_generation: &Generation,
) -> Result<(), Self::Error> {
Ok(())
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SessionAdmissionIdentity {
pub generation_id: GenerationId,
pub session_id: SessionId,
pub configuration_fingerprint: ConfigurationFingerprint,
}
pub trait SessionLifecycleAdapter {
type Error;
type Admission;
fn admission_identity(&self, admission: &Self::Admission) -> SessionAdmissionIdentity;
/// # Errors
/// Returns credential, identity, duplicate-session, or durable registration failure.
fn register_session(
&self,
store: &HostCoordinatorDomainStore,
entry_action: LifecycleAction,
admission: Self::Admission,
) -> Result<Session, Self::Error>;
/// # Errors
/// Returns exact-session or durable disconnect failure.
fn disconnect_session(
&self,
store: &HostCoordinatorDomainStore,
session: &Session,
) -> Result<(), Self::Error>;
}
#[derive(Debug)]
pub enum LifecycleError<E> {
Adapter(E),
Store(DomainStoreError),
InvalidAggregate {
action: LifecycleAction,
reason: &'static str,
},
InvalidTransition {
action: LifecycleAction,
reason: &'static str,
},
}
impl<E: fmt::Display> fmt::Display for LifecycleError<E> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Adapter(error) => write!(formatter, "lifecycle adapter failed: {error}"),
Self::Store(error) => {
write!(formatter, "Domain Store lifecycle update failed: {error}")
}
Self::InvalidAggregate { action, reason } => {
write!(
formatter,
"{action:?} produced an invalid lifecycle aggregate: {reason}"
)
}
Self::InvalidTransition { action, reason } => {
write!(formatter, "{action:?} refused: {reason}")
}
}
}
}
impl<E: std::error::Error + 'static> std::error::Error for LifecycleError<E> {}
impl<E> From<DomainStoreError> for LifecycleError<E> {
fn from(error: DomainStoreError) -> Self {
Self::Store(error)
}
}

183
crates/lazaret-host-coordinator/src/lifecycle/state.rs Normal file
View file

@ -0,0 +1,183 @@
use std::collections::{HashMap, HashSet};
use lazaret_core::identity::SessionId;
use lazaret_core::records::{DurableDecision, Generation, Session, SessionPhase};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DomainPhase {
Absent,
Starting,
Ready,
Stopping,
Failed,
}
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub(super) enum RuntimeMilestone {
TrustAnchor,
RuntimeSupervisor,
PreparedLowerLease,
StateAnchor,
Sidecar,
ConnectivityHelper,
NetworkReady,
CausalEdge,
Sandbox,
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub(super) struct RuntimeProgress {
completed: HashSet<RuntimeMilestone>,
}
impl RuntimeProgress {
pub(super) fn has(&self, milestone: RuntimeMilestone) -> bool {
self.completed.contains(&milestone)
}
pub(super) fn mark(&mut self, milestone: RuntimeMilestone) {
self.completed.insert(milestone);
}
pub(super) fn clear(&mut self, milestone: RuntimeMilestone) {
self.completed.remove(&milestone);
}
fn validate_prefix(&self) -> bool {
const ORDER: [RuntimeMilestone; 9] = [
RuntimeMilestone::TrustAnchor,
RuntimeMilestone::RuntimeSupervisor,
RuntimeMilestone::PreparedLowerLease,
RuntimeMilestone::StateAnchor,
RuntimeMilestone::Sidecar,
RuntimeMilestone::ConnectivityHelper,
RuntimeMilestone::NetworkReady,
RuntimeMilestone::CausalEdge,
RuntimeMilestone::Sandbox,
];
let completed = ORDER
.iter()
.take_while(|milestone| self.completed.contains(milestone))
.count();
self.completed.len() == completed
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub(super) struct LifecycleAggregate {
pub(super) phase: DomainPhase,
pub(super) generation: Option<Generation>,
pub(super) bootstrap_owner: Option<SessionId>,
pub(super) progress: RuntimeProgress,
pub(super) published: bool,
pub(super) sessions: HashMap<SessionId, Session>,
}
impl LifecycleAggregate {
pub(super) fn absent() -> Self {
Self {
phase: DomainPhase::Absent,
generation: None,
bootstrap_owner: None,
progress: RuntimeProgress::default(),
published: false,
sessions: HashMap::new(),
}
}
pub(super) fn validate(&self) -> Result<(), &'static str> {
if self.published != (self.phase == DomainPhase::Ready) {
return Err("publication and Domain phase disagree");
}
if !self.progress.validate_prefix() {
return Err("runtime milestones are not an ordered prefix");
}
if self.phase == DomainPhase::Ready && !self.progress.has(RuntimeMilestone::Sandbox) {
return Err("ready Domain has incomplete runtime progress");
}
if self.phase == DomainPhase::Starting {
let owner = self
.bootstrap_owner
.ok_or("starting Domain has no bootstrap owner")?;
if self
.sessions
.get(&owner)
.map(|session| session.session_phase)
!= Some(SessionPhase::Waiting)
{
return Err("bootstrap owner is not a waiting Session");
}
} else if self.bootstrap_owner.is_some() {
return Err("non-starting Domain retains a bootstrap owner");
}
match (&self.generation, self.phase) {
(None, DomainPhase::Absent) | (Some(_), DomainPhase::Failed) => {}
(Some(generation), DomainPhase::Absent)
if generation.durable_decision == DurableDecision::Clean
&& self.progress.completed.is_empty() => {}
(Some(generation), DomainPhase::Starting)
if matches!(
generation.durable_decision,
DurableDecision::None | DurableDecision::Preparing
) => {}
(Some(generation), DomainPhase::Ready | DomainPhase::Stopping)
if generation.durable_decision == DurableDecision::Published => {}
(None, _) => return Err("active Domain has no generation"),
_ => return Err("generation decision and Domain phase disagree"),
}
for (session_id, session) in &self.sessions {
if *session_id != session.session_id {
return Err("Session map key and record identity disagree");
}
let live = matches!(
session.session_phase,
SessionPhase::Waiting
| SessionPhase::Registering
| SessionPhase::Attached
| SessionPhase::Prompting
| SessionPhase::HostExecuting
);
if live
&& matches!(
self.phase,
DomainPhase::Starting | DomainPhase::Ready | DomainPhase::Stopping
)
&& self
.generation
.as_ref()
.is_none_or(|generation| session.generation_id != generation.generation_id)
{
return Err("live Session generation differs from the active generation");
}
}
Ok(())
}
pub(super) fn registered_sessions(&self) -> usize {
self.sessions
.values()
.filter(|session| {
matches!(
session.session_phase,
SessionPhase::Attached | SessionPhase::Prompting | SessionPhase::HostExecuting
)
})
.count()
}
pub(super) fn has_session_interest(&self) -> bool {
self.sessions.values().any(|session| {
matches!(
session.session_phase,
SessionPhase::Waiting
| SessionPhase::Registering
| SessionPhase::Attached
| SessionPhase::Prompting
| SessionPhase::HostExecuting
)
})
}
}
#[cfg(test)]
mod tests;

124
crates/lazaret-host-coordinator/src/lifecycle/state/tests.rs Normal file
View file

@ -0,0 +1,124 @@
use lazaret_core::identity::{
ClientId, ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, NetworkSidecarId,
SandboxId, SessionAnchorId,
};
use super::*;
fn bytes(value: u8) -> [u8; 32] {
[value; 32]
}
fn generation(decision: DurableDecision) -> Generation {
Generation {
domain_id: DomainId::new(bytes(1)),
generation_id: GenerationId::new(1),
configuration_fingerprint: ConfigurationFingerprint::new(bytes(2)),
coordinator_id: CoordinatorId::new(bytes(3)),
network_sidecar_id: NetworkSidecarId::new(bytes(4)),
sandbox_id: SandboxId::new(bytes(5)),
durable_decision: decision,
}
}
fn session(id: u8, phase: SessionPhase) -> Session {
Session {
generation_id: GenerationId::new(1),
session_id: SessionId::new(bytes(id)),
cli_id: ClientId::new(bytes(id.wrapping_add(32))),
session_anchor_id: SessionAnchorId::new(bytes(id.wrapping_add(64))),
session_phase: phase,
}
}
fn complete_progress() -> RuntimeProgress {
let mut progress = RuntimeProgress::default();
for milestone in [
RuntimeMilestone::TrustAnchor,
RuntimeMilestone::RuntimeSupervisor,
RuntimeMilestone::PreparedLowerLease,
RuntimeMilestone::StateAnchor,
RuntimeMilestone::Sidecar,
RuntimeMilestone::ConnectivityHelper,
RuntimeMilestone::NetworkReady,
RuntimeMilestone::CausalEdge,
RuntimeMilestone::Sandbox,
] {
progress.mark(milestone);
}
progress
}
#[test]
fn aggregate_refuses_independently_inconsistent_facts() {
let mut state = LifecycleAggregate::absent();
assert!(state.validate().is_ok());
state.published = true;
assert_eq!(
state.validate(),
Err("publication and Domain phase disagree")
);
state.published = false;
state.phase = DomainPhase::Starting;
state.generation = Some(generation(DurableDecision::None));
assert_eq!(
state.validate(),
Err("starting Domain has no bootstrap owner")
);
let owner = session(7, SessionPhase::Closed);
state.bootstrap_owner = Some(owner.session_id);
state.sessions.insert(owner.session_id, owner.clone());
assert_eq!(
state.validate(),
Err("bootstrap owner is not a waiting Session")
);
state.sessions.insert(
owner.session_id,
Session {
session_phase: SessionPhase::Waiting,
..owner
},
);
state.progress.mark(RuntimeMilestone::RuntimeSupervisor);
assert_eq!(
state.validate(),
Err("runtime milestones are not an ordered prefix")
);
}
#[test]
fn session_interest_and_registration_are_derived_from_session_records() {
let mut state = LifecycleAggregate {
phase: DomainPhase::Ready,
generation: Some(generation(DurableDecision::Published)),
bootstrap_owner: None,
progress: complete_progress(),
published: true,
sessions: HashMap::new(),
};
for (id, phase) in [
(1, SessionPhase::New),
(2, SessionPhase::Waiting),
(3, SessionPhase::Registering),
(4, SessionPhase::Attached),
(5, SessionPhase::Prompting),
(6, SessionPhase::HostExecuting),
(7, SessionPhase::Closed),
(8, SessionPhase::Failed),
] {
let record = session(id, phase);
state.sessions.insert(record.session_id, record);
}
assert!(state.validate().is_ok());
assert_eq!(state.registered_sessions(), 3);
assert!(state.has_session_interest());
for session in state.sessions.values_mut() {
session.session_phase = SessionPhase::Closed;
}
assert_eq!(state.registered_sessions(), 0);
assert!(!state.has_session_interest());
}

101
crates/lazaret-host-coordinator/tests/lifecycle.rs
View file

@ -1,5 +1,7 @@
use std::cell::Cell; use std::cell::Cell;
use std::fs::OpenOptions; use std::fs::OpenOptions;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use lazaret_core::identity::{ use lazaret_core::identity::{
@ -16,6 +18,31 @@ use lazaret_host_coordinator::lifecycle::{
use lazaret_linux::filesystem::try_exclusive_flock; use lazaret_linux::filesystem::try_exclusive_flock;
use rustix::pipe::{PipeFlags, pipe_with}; use rustix::pipe::{PipeFlags, pipe_with};
static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(0);
struct TestDirectory(PathBuf);
impl TestDirectory {
fn new() -> Self {
let sequence = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed);
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../target/test-artifacts/host-lifecycle")
.join(format!("{}-{sequence}", std::process::id()));
std::fs::create_dir_all(&path).unwrap();
Self(path)
}
fn path(&self) -> &Path {
&self.0
}
}
impl Drop for TestDirectory {
fn drop(&mut self) {
std::fs::remove_dir_all(&self.0).unwrap();
}
}
#[derive(Clone, Default)] #[derive(Clone, Default)]
struct RecordingAdapter { struct RecordingAdapter {
actions: Arc<Mutex<Vec<LifecycleAction>>>, actions: Arc<Mutex<Vec<LifecycleAction>>>,
@ -200,7 +227,7 @@ fn session(id: u8, generation_id: u64) -> Session {
} }
} }
fn open_store(directory: &tempfile::TempDir) -> HostCoordinatorDomainStore { fn open_store(directory: &TestDirectory) -> HostCoordinatorDomainStore {
let lock_file = OpenOptions::new() let lock_file = OpenOptions::new()
.read(true) .read(true)
.write(true) .write(true)
@ -228,9 +255,55 @@ fn complete_startup(
lifecycle.publish_domain().unwrap(); lifecycle.publish_domain().unwrap();
} }
#[test]
fn absent_aggregate_refuses_every_non_entry_action_before_effects() {
let directory = TestDirectory::new();
let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters();
let actions = Arc::clone(&domain_adapter.actions);
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
macro_rules! refused {
($operation:expr) => {
assert!(matches!(
$operation,
Err(LifecycleError::InvalidTransition { .. })
));
};
}
refused!(lifecycle.admit_session(admission(session(10, 1))));
refused!(lifecycle.wait_for_domain(session(10, 1)));
refused!(lifecycle.restart_waiting_domain(generation(1), SessionId::new(bytes(10))));
refused!(lifecycle.create_trust_anchor());
refused!(lifecycle.start_runtime_supervisor());
refused!(lifecycle.lease_prepared_lower());
refused!(lifecycle.mount_state_anchor());
refused!(lifecycle.start_sidecar());
refused!(lifecycle.start_connectivity_helper());
refused!(lifecycle.mark_network_ready());
refused!(lifecycle.arm_runtime_causal_edge());
refused!(lifecycle.start_sandbox());
refused!(lifecycle.publish_domain());
refused!(lifecycle.reject_new_configuration_mismatch(session(10, 1), fingerprint(9)));
refused!(lifecycle.reject_configuration_mismatch(SessionId::new(bytes(10)), fingerprint(9)));
refused!(lifecycle.disconnect_session(SessionId::new(bytes(10))));
refused!(lifecycle.reconcile_published_bootstrap_owner(SessionId::new(bytes(10)), false));
refused!(lifecycle.begin_teardown());
refused!(lifecycle.stop_sandbox());
refused!(lifecycle.stop_sidecar());
refused!(lifecycle.unmount_state_anchor());
refused!(lifecycle.release_prepared_lower_lease());
refused!(lifecycle.graceful_supervisor_exit());
refused!(lifecycle.finish_teardown());
assert!(actions.lock().unwrap().is_empty());
assert!(store.snapshot().unwrap().generations.is_empty());
}
#[test] #[test]
fn bootstrap_owner_guard_failure_uses_exact_modeled_disconnect_transition() { fn bootstrap_owner_guard_failure_uses_exact_modeled_disconnect_transition() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let actions = Arc::clone(&domain_adapter.actions); let actions = Arc::clone(&domain_adapter.actions);
@ -260,7 +333,7 @@ fn bootstrap_owner_guard_failure_uses_exact_modeled_disconnect_transition() {
#[test] #[test]
fn published_bootstrap_owner_eof_disconnects_waiting_session_and_tears_down() { fn published_bootstrap_owner_eof_disconnects_waiting_session_and_tears_down() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -287,7 +360,7 @@ fn published_bootstrap_owner_eof_disconnects_waiting_session_and_tears_down() {
#[test] #[test]
fn admitted_bootstrap_owner_ignores_later_bootstrap_pipe_eof() { fn admitted_bootstrap_owner_ignores_later_bootstrap_pipe_eof() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -315,7 +388,7 @@ fn admitted_bootstrap_owner_ignores_later_bootstrap_pipe_eof() {
#[test] #[test]
fn exact_start_join_disconnect_and_teardown_actions_are_ordered() { fn exact_start_join_disconnect_and_teardown_actions_are_ordered() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -383,7 +456,7 @@ fn exact_start_join_disconnect_and_teardown_actions_are_ordered() {
#[test] #[test]
fn cold_start_entry_reaches_publication_in_exact_modeled_order() { fn cold_start_entry_reaches_publication_in_exact_modeled_order() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -414,7 +487,7 @@ fn cold_start_entry_reaches_publication_in_exact_modeled_order() {
#[test] #[test]
fn publication_exposure_failure_retains_recoverable_durable_decision() { fn publication_exposure_failure_retains_recoverable_durable_decision() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (_, session_adapter) = adapters(); let (_, session_adapter) = adapters();
let mut lifecycle = let mut lifecycle =
@ -435,7 +508,7 @@ fn publication_exposure_failure_retains_recoverable_durable_decision() {
#[test] #[test]
fn publication_and_invalid_transitions_fail_before_adapter_execution() { fn publication_and_invalid_transitions_fail_before_adapter_execution() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -476,7 +549,7 @@ fn publication_and_invalid_transitions_fail_before_adapter_execution() {
#[test] #[test]
fn aliased_generation_role_identities_are_refused_before_adapter_or_store() { fn aliased_generation_role_identities_are_refused_before_adapter_or_store() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -504,7 +577,7 @@ fn aliased_generation_role_identities_are_refused_before_adapter_or_store() {
#[test] #[test]
fn configuration_mismatch_is_refused_without_registration() { fn configuration_mismatch_is_refused_without_registration() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -537,7 +610,7 @@ fn configuration_mismatch_is_refused_without_registration() {
#[test] #[test]
fn waiting_session_restarts_after_normal_teardown() { fn waiting_session_restarts_after_normal_teardown() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -605,7 +678,7 @@ fn waiting_session_restarts_after_normal_teardown() {
#[test] #[test]
fn admission_adapter_refuses_duplicate_cli_and_anchor_identities() { fn admission_adapter_refuses_duplicate_cli_and_anchor_identities() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -640,7 +713,7 @@ fn admission_adapter_refuses_duplicate_cli_and_anchor_identities() {
#[test] #[test]
fn serialized_concurrent_joins_register_distinct_sessions() { fn serialized_concurrent_joins_register_distinct_sessions() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);
@ -678,7 +751,7 @@ fn serialized_concurrent_joins_register_distinct_sessions() {
#[test] #[test]
fn bootstrap_owner_disconnect_enters_failed_without_teardown_or_recovery() { fn bootstrap_owner_disconnect_enters_failed_without_teardown_or_recovery() {
let directory = tempfile::tempdir().unwrap(); let directory = TestDirectory::new();
let store = open_store(&directory); let store = open_store(&directory);
let (domain_adapter, session_adapter) = adapters(); let (domain_adapter, session_adapter) = adapters();
let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter); let mut lifecycle = HostCoordinatorLifecycle::new(&store, domain_adapter, session_adapter);

6
tools/lazaret_verify.py
View file

@ -226,11 +226,7 @@ PROCESS_LINT_ALLOWANCES = {
}, },
} }
RUST_FILE_LINE_LIMIT = 1000 RUST_FILE_LINE_LIMIT = 1000
RUST_FILE_SIZE_EXCEPTIONS = { RUST_FILE_SIZE_EXCEPTIONS = {}
"crates/lazaret-host-coordinator/src/lifecycle.rs": (
"host coordinator lifecycle transition state machine and invariant-preserving adapter"
),
}
FORBIDDEN_PROCESS_SYSCALLS = tuple( FORBIDDEN_PROCESS_SYSCALLS = tuple(
re.compile(pattern) re.compile(pattern)
for pattern in ( for pattern in (