ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Route lifecycle admission through the control registry

Browse source
This commit is contained in:
ms2k 2026-07-16 16:08:12 +09:00
commit 510dfccf5a
2 changed files with 431 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

98
crates/lazaret-host-coordinator/src/control.rs
View file

@ -39,6 +39,9 @@ use sha2::{Digest, Sha256};
use time::{Duration, OffsetDateTime};
use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::lifecycle::{
HostCoordinatorLifecycle, LifecycleAction, SessionAdmissionIdentity, SessionLifecycleAdapter,
};
pub const CONTROL_ALPN: &[u8] = b"lazaret-control/1";
pub const CONTROL_SERVER_NAME: &str = "coordinator.lazaret.invalid";
@ -387,6 +390,15 @@ pub struct RegisteredSession {
pub session_anchor_id: SessionAnchorId,
}
#[derive(Clone)]
pub struct ControlSessionAdmission {
pub request: SessionRegistrationRequest,
pub credential: TlsCredential,
}
pub type ControlHostCoordinatorLifecycle<'store, A> =
HostCoordinatorLifecycle<'store, A, Arc<SessionRegistry>>;
#[derive(Default)]
struct RegistryState {
enrolled: BTreeMap<CertificateDigest, TlsCredential>,
@ -478,11 +490,26 @@ impl SessionRegistry {
|| registered.binding.cli_id == binding.cli_id
|| registered.session_anchor_id == session_anchor_id
});
let resumable_waiter = snapshot.sessions.iter().any(|session| {
session.generation_id == binding.generation_identity.generation_id
&& session.session_id == binding.session_id
&& session.cli_id == binding.cli_id
&& session.session_anchor_id == session_anchor_id
&& session.session_phase == SessionPhase::Waiting
});
let duplicate_session = snapshot.sessions.iter().any(|session| {
session.session_id == binding.session_id
|| (is_live_session(session.session_phase)
&& (session.cli_id == binding.cli_id
|| session.session_anchor_id == session_anchor_id))
let same_session_allowed = resumable_waiter
&& session.session_id == binding.session_id
&& (session.session_phase == SessionPhase::Closed
|| (session.generation_id == binding.generation_identity.generation_id
&& session.cli_id == binding.cli_id
&& session.session_anchor_id == session_anchor_id
&& session.session_phase == SessionPhase::Waiting));
!same_session_allowed
&& (session.session_id == binding.session_id
|| (is_live_session(session.session_phase)
&& (session.cli_id == binding.cli_id
|| session.session_anchor_id == session_anchor_id)))
});
if duplicate_in_memory || duplicate_session {
return Err(ControlError::DuplicateIdentity);
@ -580,6 +607,69 @@ impl SessionRegistry {
}
}
impl SessionLifecycleAdapter for Arc<SessionRegistry> {
type Error = ControlError;
type Admission = ControlSessionAdmission;
fn admission_identity(&self, admission: &Self::Admission) -> SessionAdmissionIdentity {
let binding = &admission.request.session_binding;
SessionAdmissionIdentity {
generation_id: binding.generation_identity.generation_id,
session_id: binding.session_id,
configuration_fingerprint: binding.generation_identity.configuration_fingerprint,
}
}
fn register_session(
&self,
_store: &HostCoordinatorDomainStore,
entry_action: LifecycleAction,
admission: Self::Admission,
) -> Result<Session, Self::Error> {
if !matches!(
entry_action,
LifecycleAction::StartReadySession | LifecycleAction::AdvanceWaitingSession
) {
return Err(ControlError::InvalidRequest);
}
let registered = self.register(admission.request, &admission.credential)?;
Ok(Session {
generation_id: registered.binding.generation_identity.generation_id,
session_id: registered.binding.session_id,
cli_id: registered.binding.cli_id,
session_anchor_id: registered.session_anchor_id,
session_phase: SessionPhase::Attached,
})
}
fn disconnect_session(
&self,
_store: &HostCoordinatorDomainStore,
session: &Session,
) -> Result<(), Self::Error> {
let mut state = self.lock()?;
let registered = state.connected.get(&session.session_id);
if let Some(registered) = registered
&& (registered.binding.generation_identity.generation_id != session.generation_id
|| registered.binding.cli_id != session.cli_id
|| registered.session_anchor_id != session.session_anchor_id)
{
return Err(ControlError::InvalidRequest);
}
if session.session_phase == SessionPhase::Attached && registered.is_none() {
return Err(ControlError::InvalidRequest);
}
state.connected.remove(&session.session_id);
drop(state);
self.store
.apply(&[DomainStoreMutation::PutSession(Session {
session_phase: SessionPhase::Closed,
..session.clone()
})])?;
Ok(())
}
}
struct ControlPaths {
root: PathBuf,
root_file: File,

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

@ -47,6 +47,37 @@ pub trait LifecycleAdapter {
) -> Result<(), Self::Error>;
}
#[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,
@ -120,9 +151,10 @@ impl RuntimeProgress {
}
}
pub struct HostCoordinatorLifecycle<'store, A> {
pub struct HostCoordinatorLifecycle<'store, A, S> {
store: &'store HostCoordinatorDomainStore,
adapter: A,
session_adapter: S,
phase: DomainPhase,
generation: Option<Generation>,
bootstrap_owner: Option<SessionId>,
@ -133,12 +165,17 @@ pub struct HostCoordinatorLifecycle<'store, A> {
registered: HashSet<SessionId>,
}
impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
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) -> Self {
pub fn new(store: &'store HostCoordinatorDomainStore, adapter: A, session_adapter: S) -> Self {
Self {
store,
adapter,
session_adapter,
phase: DomainPhase::Absent,
generation: None,
bootstrap_owner: None,
@ -215,26 +252,47 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
}
/// # Errors
/// Refuses an unpublished, stale, mismatched, or duplicate session.
pub fn start_ready_session(
/// Refuses an unpublished, stale, mismatched, duplicate, or unauthenticated session.
pub fn admit_session(
&mut self,
session: Session,
fingerprint: ConfigurationFingerprint,
admission: S::Admission,
) -> Result<(), LifecycleError<A::Error>> {
self.require_ready_join(LifecycleAction::StartReadySession, &session, fingerprint)?;
let identity = self.session_adapter.admission_identity(&admission);
let entry_action = match self.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(
LifecycleAction::StartReadySession,
session.session_phase == SessionPhase::New
&& !self.sessions.contains_key(&session.session_id),
"session is not a distinct New session",
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",
)?;
self.execute_current(LifecycleAction::StartReadySession)?;
self.put_new_session(
LifecycleAction::StartReadySession,
session,
SessionPhase::Registering,
true,
)
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.connected.insert(attached.session_id);
self.registered.insert(attached.session_id);
self.sessions.insert(attached.session_id, attached);
Ok(())
}
/// # Errors
@ -285,7 +343,7 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
generation.durable_decision == DurableDecision::None,
"new generation decision is not None",
)?;
let mut session = self
let session = self
.session(session_id, LifecycleAction::RestartWaitingDomain)?
.clone();
Self::require(
@ -303,21 +361,36 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
&successor_session,
)?;
self.execute(LifecycleAction::RestartWaitingDomain, &generation)?;
let previous = Session {
session_phase: SessionPhase::Closed,
..session.clone()
};
session.generation_id = generation.generation_id;
self.store.apply(&[
DomainStoreMutation::PutSession(previous),
DomainStoreMutation::PutGeneration(generation.clone()),
DomainStoreMutation::PutSession(session.clone()),
])?;
let waiters = self
.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)?;
self.phase = DomainPhase::Starting;
self.bootstrap_owner = Some(session_id);
let generation_id = generation.generation_id;
self.generation = Some(generation);
self.progress = RuntimeProgress::default();
self.sessions.insert(session_id, session);
for mut waiter in waiters {
waiter.generation_id = generation_id;
self.sessions.insert(waiter.session_id, waiter);
}
Ok(())
}
@ -463,31 +536,6 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
Ok(())
}
/// # Errors
/// Refuses a non-waiting session, configuration mismatch, or unavailable Domain.
pub fn advance_waiting_session(
&mut self,
session_id: SessionId,
fingerprint: ConfigurationFingerprint,
) -> Result<(), LifecycleError<A::Error>> {
self.require_fingerprint(LifecycleAction::AdvanceWaitingSession, fingerprint)?;
let mut session = self
.session(session_id, LifecycleAction::AdvanceWaitingSession)?
.clone();
Self::require(
LifecycleAction::AdvanceWaitingSession,
session.session_phase == SessionPhase::Waiting && self.admission_ready(),
"waiting session is not admissible",
)?;
self.execute_current(LifecycleAction::AdvanceWaitingSession)?;
session.session_phase = SessionPhase::Registering;
self.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])?;
self.connected.insert(session_id);
self.sessions.insert(session_id, session);
Ok(())
}
/// # Errors
/// Refuses unless a new or waiting session differs from the ready generation.
pub fn reject_new_configuration_mismatch(
@ -558,31 +606,6 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
Ok(())
}
/// # Errors
/// Refuses a disconnected, stale, or non-registering session.
pub fn register_session(
&mut self,
session_id: SessionId,
) -> Result<(), LifecycleError<A::Error>> {
let mut session = self
.session(session_id, LifecycleAction::RegisterSession)?
.clone();
Self::require(
LifecycleAction::RegisterSession,
self.admission_ready()
&& session.session_phase == SessionPhase::Registering
&& self.connected.contains(&session_id),
"session is not admissible",
)?;
self.execute_current(LifecycleAction::RegisterSession)?;
session.session_phase = SessionPhase::Attached;
self.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])?;
self.registered.insert(session_id);
self.sessions.insert(session_id, session);
Ok(())
}
/// # Errors
/// Refuses an absent or terminal session, or an adapter/store failure.
pub fn disconnect_session(
@ -604,12 +627,12 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
),
"session is not live",
)?;
self.execute_current(LifecycleAction::DisconnectSession)?;
let leader_lost =
self.phase == DomainPhase::Starting && self.bootstrap_owner == Some(session_id);
self.session_adapter
.disconnect_session(self.store, &session)
.map_err(LifecycleError::Adapter)?;
session.session_phase = SessionPhase::Closed;
self.store
.apply(&[DomainStoreMutation::PutSession(session.clone())])?;
self.connected.remove(&session_id);
self.registered.remove(&session_id);
self.sessions.insert(session_id, session);
@ -741,8 +764,8 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
Ok(())
}
pub fn into_adapter(self) -> A {
self.adapter
pub fn into_adapters(self) -> (A, S) {
(self.adapter, self.session_adapter)
}
fn startup_action(
@ -804,29 +827,6 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
Ok(())
}
fn require_ready_join(
&self,
action: LifecycleAction,
session: &Session,
fingerprint: ConfigurationFingerprint,
) -> Result<(), LifecycleError<A::Error>> {
Self::require(
action,
self.admission_ready(),
"Domain is not published and ready",
)?;
Self::require(
action,
session.generation_id == self.current_generation(action)?.generation_id,
"session generation differs",
)?;
Self::require(
action,
fingerprint == self.current_generation(action)?.configuration_fingerprint,
"configuration differs",
)
}
fn require_new_generation(
&self,
action: LifecycleAction,
@ -843,23 +843,14 @@ impl<'store, A: LifecycleAdapter> HostCoordinatorLifecycle<'store, A> {
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 require_fingerprint(
&self,
action: LifecycleAction,
fingerprint: ConfigurationFingerprint,
) -> Result<(), LifecycleError<A::Error>> {
Self::require(
action,
fingerprint == self.current_generation(action)?.configuration_fingerprint,
"configuration differs",
)
}
fn admission_ready(&self) -> bool {
self.phase == DomainPhase::Ready && self.published
}