946 lines
35 KiB
Rust
946 lines
35 KiB
Rust
use lazaret_core::identity::{ConfigurationFingerprint, SessionId};
|
|
use lazaret_core::records::{DurableDecision, Generation, Session, SessionPhase};
|
|
|
|
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
|
|
|
|
mod contracts;
|
|
mod state;
|
|
pub use contracts::{
|
|
LifecycleAction, LifecycleAdapter, LifecycleError, SessionAdmissionIdentity,
|
|
SessionLifecycleAdapter,
|
|
};
|
|
pub use state::DomainPhase;
|
|
use state::{LifecycleAggregate, RuntimeMilestone, RuntimeProgress};
|
|
|
|
pub struct HostCoordinatorLifecycle<'store, A, S> {
|
|
store: &'store HostCoordinatorDomainStore,
|
|
adapter: A,
|
|
session_adapter: S,
|
|
state: LifecycleAggregate,
|
|
}
|
|
|
|
impl<'store, A, S> HostCoordinatorLifecycle<'store, A, S>
|
|
where
|
|
A: LifecycleAdapter,
|
|
S: SessionLifecycleAdapter<Error = A::Error>,
|
|
{
|
|
#[must_use]
|
|
pub fn new(store: &'store HostCoordinatorDomainStore, adapter: A, session_adapter: S) -> Self {
|
|
Self {
|
|
store,
|
|
adapter,
|
|
session_adapter,
|
|
state: LifecycleAggregate::absent(),
|
|
}
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn phase(&self) -> DomainPhase {
|
|
self.state.phase
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn is_published(&self) -> bool {
|
|
self.state.published
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn session_phase(&self, session_id: SessionId) -> Option<SessionPhase> {
|
|
self.state
|
|
.sessions
|
|
.get(&session_id)
|
|
.map(|session| session.session_phase)
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn registered_sessions(&self) -> usize {
|
|
self.state.registered_sessions()
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn adapter(&self) -> &A {
|
|
&self.adapter
|
|
}
|
|
|
|
#[must_use]
|
|
pub const fn adapter_mut(&mut self) -> &mut A {
|
|
&mut self.adapter
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses a non-absent Domain, invalid initial records, or adapter/store failure.
|
|
pub fn start_new_domain(
|
|
&mut self,
|
|
generation: Generation,
|
|
mut session: Session,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::StartNewDomain,
|
|
self.state.phase == DomainPhase::Absent,
|
|
"Domain is not absent",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::StartNewDomain,
|
|
generation.durable_decision == DurableDecision::None,
|
|
"new generation decision is not None",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::StartNewDomain,
|
|
session.generation_id == generation.generation_id,
|
|
"session generation differs",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::StartNewDomain,
|
|
session.session_phase == SessionPhase::New,
|
|
"session is not New",
|
|
)?;
|
|
self.require_new_generation(LifecycleAction::StartNewDomain, &generation, &session)?;
|
|
self.execute(LifecycleAction::StartNewDomain, &generation)?;
|
|
session.session_phase = SessionPhase::Waiting;
|
|
self.store.apply(&[
|
|
DomainStoreMutation::PutGeneration(generation.clone()),
|
|
DomainStoreMutation::PutSession(session.clone()),
|
|
])?;
|
|
self.transition(LifecycleAction::StartNewDomain, move |state| {
|
|
state.phase = DomainPhase::Starting;
|
|
state.bootstrap_owner = Some(session.session_id);
|
|
state.sessions.insert(session.session_id, session);
|
|
state.generation = Some(generation);
|
|
state.progress = RuntimeProgress::default();
|
|
})
|
|
}
|
|
|
|
/// Drives one cold-start leader through the complete normal publication boundary.
|
|
///
|
|
/// # Errors
|
|
/// Returns the first invalid transition, durable update, or lifecycle adapter failure.
|
|
pub fn bootstrap_new_domain(
|
|
&mut self,
|
|
generation: Generation,
|
|
session: Session,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
self.bootstrap_new_domain_while_owner_live(generation, session, || true)
|
|
.map(|_| ())
|
|
}
|
|
|
|
/// Drives cold start while checking the bootstrap owner's exact lifetime authority.
|
|
///
|
|
/// Returns `false` after applying the modeled bootstrap-owner disconnect transition.
|
|
///
|
|
/// # Errors
|
|
/// Returns the first invalid transition, durable update, or lifecycle adapter failure.
|
|
pub fn bootstrap_new_domain_while_owner_live(
|
|
&mut self,
|
|
generation: Generation,
|
|
session: Session,
|
|
owner_live: impl FnMut() -> bool,
|
|
) -> Result<bool, LifecycleError<A::Error>> {
|
|
self.bootstrap_new_domain_while_owner_live_and_service(
|
|
generation,
|
|
session,
|
|
owner_live,
|
|
|_| Ok(()),
|
|
|_| Ok(()),
|
|
)
|
|
}
|
|
|
|
/// Drives cold start while servicing bounded pre-publication waiters between actions.
|
|
///
|
|
/// # Errors
|
|
/// Returns the first invalid transition, durable update, adapter, or waiter-service failure.
|
|
pub fn bootstrap_new_domain_while_owner_live_and_service(
|
|
&mut self,
|
|
generation: Generation,
|
|
session: Session,
|
|
mut owner_live: impl FnMut() -> bool,
|
|
mut service_waiters: impl FnMut(&mut Self) -> Result<(), A::Error>,
|
|
mut withdraw_waiter_admission: impl FnMut(&mut Self) -> Result<(), A::Error>,
|
|
) -> Result<bool, LifecycleError<A::Error>> {
|
|
let session_id = session.session_id;
|
|
self.start_new_domain(generation, session)?;
|
|
macro_rules! checkpoint {
|
|
() => {
|
|
service_waiters(self).map_err(LifecycleError::Adapter)?;
|
|
if !owner_live() {
|
|
self.disconnect_session(session_id)?;
|
|
return Ok(false);
|
|
}
|
|
};
|
|
}
|
|
checkpoint!();
|
|
self.create_trust_anchor()?;
|
|
checkpoint!();
|
|
self.start_runtime_supervisor()?;
|
|
checkpoint!();
|
|
self.lease_prepared_lower()?;
|
|
checkpoint!();
|
|
self.mount_state_anchor()?;
|
|
checkpoint!();
|
|
self.start_sidecar()?;
|
|
checkpoint!();
|
|
self.start_connectivity_helper()?;
|
|
checkpoint!();
|
|
self.arm_runtime_causal_edge()?;
|
|
checkpoint!();
|
|
self.start_sandbox()?;
|
|
checkpoint!();
|
|
self.mark_network_ready()?;
|
|
checkpoint!();
|
|
withdraw_waiter_admission(self).map_err(LifecycleError::Adapter)?;
|
|
if !owner_live() {
|
|
self.disconnect_session(session_id)?;
|
|
return Ok(false);
|
|
}
|
|
self.publish_domain()?;
|
|
Ok(true)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an unpublished, stale, mismatched, duplicate, or unauthenticated session.
|
|
pub fn admit_session(
|
|
&mut self,
|
|
admission: S::Admission,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
let identity = self.session_adapter.admission_identity(&admission);
|
|
let entry_action = match self.state.sessions.get(&identity.session_id) {
|
|
None => LifecycleAction::StartReadySession,
|
|
Some(session) if session.session_phase == SessionPhase::Waiting => {
|
|
LifecycleAction::AdvanceWaitingSession
|
|
}
|
|
Some(_) => {
|
|
return Err(LifecycleError::InvalidTransition {
|
|
action: LifecycleAction::RegisterSession,
|
|
reason: "session is not New or Waiting",
|
|
});
|
|
}
|
|
};
|
|
let generation = self.current_generation(entry_action)?;
|
|
Self::require(
|
|
entry_action,
|
|
self.admission_ready()
|
|
&& identity.generation_id == generation.generation_id
|
|
&& identity.configuration_fingerprint == generation.configuration_fingerprint,
|
|
"Domain is not published for the admission generation",
|
|
)?;
|
|
let attached = self
|
|
.session_adapter
|
|
.register_session(self.store, entry_action, admission)
|
|
.map_err(LifecycleError::Adapter)?;
|
|
Self::require(
|
|
LifecycleAction::RegisterSession,
|
|
attached.generation_id == identity.generation_id
|
|
&& attached.session_id == identity.session_id
|
|
&& attached.session_phase == SessionPhase::Attached,
|
|
"session adapter returned mismatched registration evidence",
|
|
)?;
|
|
self.transition(LifecycleAction::RegisterSession, move |state| {
|
|
state.sessions.insert(attached.session_id, attached);
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses a session unless the Domain is starting or stopping.
|
|
pub fn wait_for_domain(&mut self, session: Session) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::WaitForDomain,
|
|
matches!(
|
|
self.state.phase,
|
|
DomainPhase::Starting | DomainPhase::Stopping
|
|
),
|
|
"Domain is neither starting nor stopping",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::WaitForDomain,
|
|
session.session_phase == SessionPhase::New,
|
|
"session is not New",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::WaitForDomain,
|
|
session.generation_id
|
|
== self
|
|
.current_generation(LifecycleAction::WaitForDomain)?
|
|
.generation_id
|
|
&& !self.state.sessions.contains_key(&session.session_id),
|
|
"session generation differs or Session ID is already known",
|
|
)?;
|
|
self.execute_current(LifecycleAction::WaitForDomain)?;
|
|
self.put_new_session(
|
|
LifecycleAction::WaitForDomain,
|
|
session,
|
|
SessionPhase::Waiting,
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses a restart unless the Domain is absent and the session is waiting.
|
|
pub fn restart_waiting_domain(
|
|
&mut self,
|
|
generation: Generation,
|
|
session_id: SessionId,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::RestartWaitingDomain,
|
|
self.state.phase == DomainPhase::Absent,
|
|
"Domain is not absent",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::RestartWaitingDomain,
|
|
generation.durable_decision == DurableDecision::None,
|
|
"new generation decision is not None",
|
|
)?;
|
|
let session = self
|
|
.session(session_id, LifecycleAction::RestartWaitingDomain)?
|
|
.clone();
|
|
Self::require(
|
|
LifecycleAction::RestartWaitingDomain,
|
|
session.session_phase == SessionPhase::Waiting,
|
|
"session is not Waiting",
|
|
)?;
|
|
let successor_session = Session {
|
|
generation_id: generation.generation_id,
|
|
..session.clone()
|
|
};
|
|
self.require_new_generation(
|
|
LifecycleAction::RestartWaitingDomain,
|
|
&generation,
|
|
&successor_session,
|
|
)?;
|
|
self.execute(LifecycleAction::RestartWaitingDomain, &generation)?;
|
|
let waiters = self
|
|
.state
|
|
.sessions
|
|
.values()
|
|
.filter(|candidate| candidate.session_phase == SessionPhase::Waiting)
|
|
.cloned()
|
|
.collect::<Vec<_>>();
|
|
let mut mutations = Vec::with_capacity(waiters.len() * 2 + 1);
|
|
for waiter in &waiters {
|
|
mutations.push(DomainStoreMutation::PutSession(Session {
|
|
session_phase: SessionPhase::Closed,
|
|
..waiter.clone()
|
|
}));
|
|
}
|
|
mutations.push(DomainStoreMutation::PutGeneration(generation.clone()));
|
|
for waiter in &waiters {
|
|
mutations.push(DomainStoreMutation::PutSession(Session {
|
|
generation_id: generation.generation_id,
|
|
..waiter.clone()
|
|
}));
|
|
}
|
|
self.store.apply(&mutations)?;
|
|
let generation_id = generation.generation_id;
|
|
self.transition(LifecycleAction::RestartWaitingDomain, move |state| {
|
|
state.phase = DomainPhase::Starting;
|
|
state.bootstrap_owner = Some(session_id);
|
|
state.generation = Some(generation);
|
|
state.progress = RuntimeProgress::default();
|
|
for mut waiter in waiters {
|
|
waiter.generation_id = generation_id;
|
|
state.sessions.insert(waiter.session_id, waiter);
|
|
}
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn create_trust_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::CreateTrustAnchor,
|
|
!self.state.progress.has(RuntimeMilestone::TrustAnchor),
|
|
|p| p.mark(RuntimeMilestone::TrustAnchor),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn start_runtime_supervisor(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::StartRuntimeSupervisor,
|
|
self.state.progress.has(RuntimeMilestone::TrustAnchor)
|
|
&& !self.state.progress.has(RuntimeMilestone::RuntimeSupervisor),
|
|
|p| p.mark(RuntimeMilestone::RuntimeSupervisor),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn lease_prepared_lower(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::LeasePreparedLower,
|
|
self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
|
|
&& !self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease),
|
|
|p| p.mark(RuntimeMilestone::PreparedLowerLease),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn mount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::MountStateAnchor,
|
|
self.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease)
|
|
&& !self.state.progress.has(RuntimeMilestone::StateAnchor),
|
|
|p| p.mark(RuntimeMilestone::StateAnchor),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn start_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::StartSidecar,
|
|
self.state.progress.has(RuntimeMilestone::StateAnchor)
|
|
&& !self.state.progress.has(RuntimeMilestone::Sidecar),
|
|
|p| p.mark(RuntimeMilestone::Sidecar),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn start_connectivity_helper(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::StartConnectivityHelper,
|
|
self.state.progress.has(RuntimeMilestone::Sidecar)
|
|
&& !self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::ConnectivityHelper),
|
|
|p| p.mark(RuntimeMilestone::ConnectivityHelper),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn mark_network_ready(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::MarkNetworkReady,
|
|
self.state.progress.has(RuntimeMilestone::Sandbox)
|
|
&& !self.state.progress.has(RuntimeMilestone::NetworkReady),
|
|
|p| p.mark(RuntimeMilestone::NetworkReady),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn arm_runtime_causal_edge(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.startup_action(
|
|
LifecycleAction::ArmRuntimeCausalEdge,
|
|
self.state
|
|
.progress
|
|
.has(RuntimeMilestone::ConnectivityHelper)
|
|
&& !self.state.progress.has(RuntimeMilestone::CausalEdge),
|
|
|p| p.mark(RuntimeMilestone::CausalEdge),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses launch before helper attestation and the causal edge are established.
|
|
pub fn start_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
let allowed = self.state.progress.has(RuntimeMilestone::CausalEdge)
|
|
&& self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::ConnectivityHelper)
|
|
&& !self.state.progress.has(RuntimeMilestone::Sandbox);
|
|
Self::require(
|
|
LifecycleAction::StartSandbox,
|
|
self.state.phase == DomainPhase::Starting && allowed,
|
|
"runtime prerequisites are incomplete",
|
|
)?;
|
|
let mut generation = self
|
|
.current_generation(LifecycleAction::StartSandbox)?
|
|
.clone();
|
|
generation.durable_decision = DurableDecision::Preparing;
|
|
self.store
|
|
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
|
|
self.transition(LifecycleAction::StartSandbox, {
|
|
let generation = generation.clone();
|
|
move |state| state.generation = Some(generation)
|
|
})?;
|
|
self.execute(LifecycleAction::StartSandbox, &generation)?;
|
|
self.transition(LifecycleAction::StartSandbox, |state| {
|
|
state.progress.mark(RuntimeMilestone::Sandbox);
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses publication until every ordered startup prerequisite is complete.
|
|
pub fn publish_domain(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
let allowed = self.state.phase == DomainPhase::Starting
|
|
&& self.state.progress.has(RuntimeMilestone::TrustAnchor)
|
|
&& self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
|
|
&& self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease)
|
|
&& self.state.progress.has(RuntimeMilestone::StateAnchor)
|
|
&& self.state.progress.has(RuntimeMilestone::Sidecar)
|
|
&& 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(
|
|
LifecycleAction::PublishDomain,
|
|
allowed,
|
|
"publication prerequisites are incomplete",
|
|
)?;
|
|
self.execute_current(LifecycleAction::PublishDomain)?;
|
|
let mut generation = self
|
|
.current_generation(LifecycleAction::PublishDomain)?
|
|
.clone();
|
|
generation.durable_decision = DurableDecision::Published;
|
|
self.store
|
|
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
|
|
if let Err(error) = self
|
|
.adapter
|
|
.after_commit(LifecycleAction::PublishDomain, &generation)
|
|
{
|
|
self.transition(LifecycleAction::PublishDomain, move |state| {
|
|
state.generation = Some(generation);
|
|
state.phase = DomainPhase::Failed;
|
|
state.bootstrap_owner = None;
|
|
state.published = false;
|
|
})?;
|
|
return Err(LifecycleError::Adapter(error));
|
|
}
|
|
self.transition(LifecycleAction::PublishDomain, move |state| {
|
|
state.generation = Some(generation);
|
|
state.phase = DomainPhase::Ready;
|
|
state.bootstrap_owner = None;
|
|
state.published = true;
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses unless a new or waiting session differs from the ready generation.
|
|
pub fn reject_new_configuration_mismatch(
|
|
&mut self,
|
|
mut session: Session,
|
|
fingerprint: ConfigurationFingerprint,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::RejectConfigurationMismatch,
|
|
self.state.phase == DomainPhase::Ready && session.session_phase == SessionPhase::New,
|
|
"session is not eligible for mismatch refusal",
|
|
)?;
|
|
let generation = self.current_generation(LifecycleAction::RejectConfigurationMismatch)?;
|
|
Self::require(
|
|
LifecycleAction::RejectConfigurationMismatch,
|
|
session.generation_id == generation.generation_id
|
|
&& fingerprint != generation.configuration_fingerprint,
|
|
"configuration does not differ from the current generation",
|
|
)?;
|
|
Self::require(
|
|
LifecycleAction::RejectConfigurationMismatch,
|
|
!self.state.sessions.contains_key(&session.session_id),
|
|
"Session ID is already known",
|
|
)?;
|
|
self.execute_current(LifecycleAction::RejectConfigurationMismatch)?;
|
|
let new = session.clone();
|
|
session.session_phase = SessionPhase::Failed;
|
|
self.store.apply(&[
|
|
DomainStoreMutation::PutSession(new),
|
|
DomainStoreMutation::PutSession(session.clone()),
|
|
])?;
|
|
self.transition(LifecycleAction::RejectConfigurationMismatch, move |state| {
|
|
state.sessions.insert(session.session_id, session);
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses unless a new or waiting session differs from the ready generation.
|
|
pub fn reject_configuration_mismatch(
|
|
&mut self,
|
|
session_id: SessionId,
|
|
fingerprint: ConfigurationFingerprint,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
let mut session = self
|
|
.session(session_id, LifecycleAction::RejectConfigurationMismatch)?
|
|
.clone();
|
|
Self::require(
|
|
LifecycleAction::RejectConfigurationMismatch,
|
|
self.state.phase == DomainPhase::Ready
|
|
&& matches!(
|
|
session.session_phase,
|
|
SessionPhase::New | SessionPhase::Waiting
|
|
),
|
|
"session is not eligible for mismatch refusal",
|
|
)?;
|
|
let generation = self.current_generation(LifecycleAction::RejectConfigurationMismatch)?;
|
|
Self::require(
|
|
LifecycleAction::RejectConfigurationMismatch,
|
|
fingerprint != generation.configuration_fingerprint,
|
|
"configuration matches",
|
|
)?;
|
|
self.execute_current(LifecycleAction::RejectConfigurationMismatch)?;
|
|
session.session_phase = SessionPhase::Failed;
|
|
self.store
|
|
.apply(&[DomainStoreMutation::PutSession(session.clone())])?;
|
|
self.transition(LifecycleAction::RejectConfigurationMismatch, move |state| {
|
|
state.sessions.insert(session_id, session);
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an absent or terminal session, or an adapter/store failure.
|
|
pub fn disconnect_session(
|
|
&mut self,
|
|
session_id: SessionId,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
let mut session = self
|
|
.session(session_id, LifecycleAction::DisconnectSession)?
|
|
.clone();
|
|
Self::require(
|
|
LifecycleAction::DisconnectSession,
|
|
matches!(
|
|
session.session_phase,
|
|
SessionPhase::Waiting
|
|
| SessionPhase::Registering
|
|
| SessionPhase::Attached
|
|
| SessionPhase::Prompting
|
|
| SessionPhase::HostExecuting
|
|
),
|
|
"session is not live",
|
|
)?;
|
|
let leader_lost = self.state.phase == DomainPhase::Starting
|
|
&& self.state.bootstrap_owner == Some(session_id);
|
|
self.session_adapter
|
|
.disconnect_session(self.store, &session)
|
|
.map_err(LifecycleError::Adapter)?;
|
|
session.session_phase = SessionPhase::Closed;
|
|
self.transition(LifecycleAction::DisconnectSession, move |state| {
|
|
state.sessions.insert(session_id, session);
|
|
if leader_lost {
|
|
state.phase = DomainPhase::Failed;
|
|
state.bootstrap_owner = None;
|
|
state.published = false;
|
|
}
|
|
})
|
|
}
|
|
|
|
#[must_use]
|
|
pub fn is_teardown_ready(&self) -> bool {
|
|
self.state.phase == DomainPhase::Ready && !self.state.has_session_interest()
|
|
}
|
|
|
|
/// Performs the complete ordered normal teardown when the final session has disconnected.
|
|
///
|
|
/// # Errors
|
|
/// Returns the first physical or durable teardown failure.
|
|
pub fn teardown_if_idle(&mut self) -> Result<bool, LifecycleError<A::Error>> {
|
|
if !self.is_teardown_ready() {
|
|
return Ok(false);
|
|
}
|
|
self.begin_teardown()?;
|
|
self.stop_sandbox()?;
|
|
self.stop_sidecar()?;
|
|
self.unmount_state_anchor()?;
|
|
self.release_prepared_lower_lease()?;
|
|
self.graceful_supervisor_exit()?;
|
|
self.finish_teardown()?;
|
|
Ok(true)
|
|
}
|
|
|
|
/// Transfers a published cold-start owner's lifetime edge to exact control admission.
|
|
///
|
|
/// # Errors
|
|
/// Refuses an unpublished Domain, unknown owner, or durable disconnect/teardown failure.
|
|
pub fn reconcile_published_bootstrap_owner(
|
|
&mut self,
|
|
session_id: SessionId,
|
|
bootstrap_owner_live: bool,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::DisconnectSession,
|
|
self.state.phase == DomainPhase::Ready && self.state.published,
|
|
"Domain is not published",
|
|
)?;
|
|
let phase = self
|
|
.session(session_id, LifecycleAction::DisconnectSession)?
|
|
.session_phase;
|
|
if phase == SessionPhase::Waiting && !bootstrap_owner_live {
|
|
self.disconnect_session(session_id)?;
|
|
self.teardown_if_idle()?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses while any session retains Domain interest.
|
|
pub fn begin_teardown(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
LifecycleAction::BeginTeardown,
|
|
self.is_teardown_ready(),
|
|
"sessions still have interest",
|
|
)?;
|
|
self.execute_current(LifecycleAction::BeginTeardown)?;
|
|
self.transition(LifecycleAction::BeginTeardown, |state| {
|
|
state.phase = DomainPhase::Stopping;
|
|
state.published = false;
|
|
})
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn stop_sandbox(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.teardown_action(
|
|
LifecycleAction::StopSandbox,
|
|
self.state.progress.has(RuntimeMilestone::Sandbox),
|
|
|p| {
|
|
p.clear(RuntimeMilestone::NetworkReady);
|
|
p.clear(RuntimeMilestone::Sandbox);
|
|
p.clear(RuntimeMilestone::CausalEdge);
|
|
},
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn stop_sidecar(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.teardown_action(
|
|
LifecycleAction::StopSidecar,
|
|
!self.state.progress.has(RuntimeMilestone::Sandbox)
|
|
&& self.state.progress.has(RuntimeMilestone::Sidecar),
|
|
|p| {
|
|
p.clear(RuntimeMilestone::Sidecar);
|
|
p.clear(RuntimeMilestone::ConnectivityHelper);
|
|
p.clear(RuntimeMilestone::NetworkReady);
|
|
},
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn unmount_state_anchor(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.teardown_action(
|
|
LifecycleAction::UnmountStateAnchor,
|
|
!self.state.progress.has(RuntimeMilestone::Sidecar)
|
|
&& self.state.progress.has(RuntimeMilestone::StateAnchor),
|
|
|p| p.clear(RuntimeMilestone::StateAnchor),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn release_prepared_lower_lease(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.teardown_action(
|
|
LifecycleAction::ReleasePreparedLowerLease,
|
|
!self.state.progress.has(RuntimeMilestone::StateAnchor)
|
|
&& self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease),
|
|
|p| p.clear(RuntimeMilestone::PreparedLowerLease),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses an out-of-order action or adapter failure.
|
|
pub fn graceful_supervisor_exit(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
self.teardown_action(
|
|
LifecycleAction::GracefulSupervisorExit,
|
|
!self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease)
|
|
&& self.state.progress.has(RuntimeMilestone::RuntimeSupervisor),
|
|
|p| p.clear(RuntimeMilestone::RuntimeSupervisor),
|
|
)
|
|
}
|
|
|
|
/// # Errors
|
|
/// Refuses while runtime resources remain or the final durable update fails.
|
|
pub fn finish_teardown(&mut self) -> Result<(), LifecycleError<A::Error>> {
|
|
let allowed = self.state.phase == DomainPhase::Stopping
|
|
&& !self.state.progress.has(RuntimeMilestone::RuntimeSupervisor)
|
|
&& !self
|
|
.state
|
|
.progress
|
|
.has(RuntimeMilestone::PreparedLowerLease)
|
|
&& !self.state.progress.has(RuntimeMilestone::StateAnchor)
|
|
&& !self.state.progress.has(RuntimeMilestone::Sidecar)
|
|
&& !self.state.progress.has(RuntimeMilestone::Sandbox);
|
|
Self::require(
|
|
LifecycleAction::FinishTeardown,
|
|
allowed,
|
|
"runtime resources remain",
|
|
)?;
|
|
self.execute_current(LifecycleAction::FinishTeardown)?;
|
|
let mut generation = self
|
|
.current_generation(LifecycleAction::FinishTeardown)?
|
|
.clone();
|
|
generation.durable_decision = DurableDecision::Clean;
|
|
self.store
|
|
.apply(&[DomainStoreMutation::PutGeneration(generation.clone())])?;
|
|
self.transition(LifecycleAction::FinishTeardown, move |state| {
|
|
state.generation = Some(generation);
|
|
state.phase = DomainPhase::Absent;
|
|
state.progress = RuntimeProgress::default();
|
|
state.bootstrap_owner = None;
|
|
})
|
|
}
|
|
|
|
pub fn into_adapters(self) -> (A, S) {
|
|
(self.adapter, self.session_adapter)
|
|
}
|
|
|
|
fn startup_action(
|
|
&mut self,
|
|
action: LifecycleAction,
|
|
allowed: bool,
|
|
update: impl FnOnce(&mut RuntimeProgress),
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
action,
|
|
self.state.phase == DomainPhase::Starting && allowed,
|
|
"startup ordering refused",
|
|
)?;
|
|
self.execute_current(action)?;
|
|
self.transition(action, |state| update(&mut state.progress))
|
|
}
|
|
|
|
fn teardown_action(
|
|
&mut self,
|
|
action: LifecycleAction,
|
|
allowed: bool,
|
|
update: impl FnOnce(&mut RuntimeProgress),
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
action,
|
|
self.state.phase == DomainPhase::Stopping && allowed,
|
|
"teardown ordering refused",
|
|
)?;
|
|
self.execute_current(action)?;
|
|
self.transition(action, |state| update(&mut state.progress))
|
|
}
|
|
|
|
fn put_new_session(
|
|
&mut self,
|
|
action: LifecycleAction,
|
|
mut session: Session,
|
|
phase: SessionPhase,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
Self::require(
|
|
action,
|
|
session.session_phase == SessionPhase::New,
|
|
"session is not New",
|
|
)?;
|
|
Self::require(
|
|
action,
|
|
!self.state.sessions.contains_key(&session.session_id),
|
|
"Session ID is already known",
|
|
)?;
|
|
session.session_phase = phase;
|
|
self.store
|
|
.apply(&[DomainStoreMutation::PutSession(session.clone())])?;
|
|
self.transition(action, move |state| {
|
|
state.sessions.insert(session.session_id, session);
|
|
})
|
|
}
|
|
|
|
fn require_new_generation(
|
|
&self,
|
|
action: LifecycleAction,
|
|
generation: &Generation,
|
|
session: &Session,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
let snapshot = self.store.snapshot()?;
|
|
let latest = snapshot
|
|
.generations
|
|
.iter()
|
|
.map(|candidate| candidate.generation_id)
|
|
.max();
|
|
Self::require(
|
|
action,
|
|
generation.domain_id == snapshot.domain.domain_id
|
|
&& session.generation_id == generation.generation_id
|
|
&& generation.coordinator_id.as_bytes() != generation.network_sidecar_id.as_bytes()
|
|
&& generation.coordinator_id.as_bytes() != generation.sandbox_id.as_bytes()
|
|
&& generation.network_sidecar_id.as_bytes() != generation.sandbox_id.as_bytes()
|
|
&& latest.is_none_or(|current| generation.generation_id > current),
|
|
"generation is not a fresh successor for this Domain",
|
|
)
|
|
}
|
|
|
|
fn admission_ready(&self) -> bool {
|
|
self.state.phase == DomainPhase::Ready && self.state.published
|
|
}
|
|
|
|
fn current_generation(
|
|
&self,
|
|
action: LifecycleAction,
|
|
) -> Result<&Generation, LifecycleError<A::Error>> {
|
|
self.state
|
|
.generation
|
|
.as_ref()
|
|
.ok_or(LifecycleError::InvalidTransition {
|
|
action,
|
|
reason: "generation is absent",
|
|
})
|
|
}
|
|
|
|
fn session(
|
|
&self,
|
|
session_id: SessionId,
|
|
action: LifecycleAction,
|
|
) -> Result<&Session, LifecycleError<A::Error>> {
|
|
self.state
|
|
.sessions
|
|
.get(&session_id)
|
|
.ok_or(LifecycleError::InvalidTransition {
|
|
action,
|
|
reason: "session is absent",
|
|
})
|
|
}
|
|
|
|
fn execute_current(&mut self, action: LifecycleAction) -> Result<(), LifecycleError<A::Error>> {
|
|
let generation = self.current_generation(action)?.clone();
|
|
self.execute(action, &generation)
|
|
}
|
|
|
|
fn execute(
|
|
&mut self,
|
|
action: LifecycleAction,
|
|
generation: &Generation,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
self.adapter
|
|
.execute(action, generation)
|
|
.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(
|
|
action: LifecycleAction,
|
|
condition: bool,
|
|
reason: &'static str,
|
|
) -> Result<(), LifecycleError<A::Error>> {
|
|
if condition {
|
|
Ok(())
|
|
} else {
|
|
Err(LifecycleError::InvalidTransition { action, reason })
|
|
}
|
|
}
|
|
}
|