ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Recover pre-identity session terminals safely

Browse source
This commit is contained in:
ms2k 2026-07-29 00:42:11 +09:00
commit e3e21fff45
2 changed files with 332 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

218
crates/lazaret-host-coordinator/src/recovery/reconciliation.rs
View file

@ -9,7 +9,8 @@ use lazaret_core::configuration::{ConfinedPathType, ResolvedDomainConfiguration}
use lazaret_core::identity::{GenerationId, HostPath, MountIdentity, PidfsProcessIdentityV1};
use lazaret_core::records::{
ArtifactLease, DurableDecision, Generation, Participant, ParticipantOwnership, ParticipantRole,
ParticipantState, SandboxOwnership, SessionPhase,
ParticipantState, PersistedHostProcessOwnership, SandboxOwnership, SessionPhase,
SessionTerminal, SessionTerminalFields, SessionTerminalState,
};
use lazaret_linux::filesystem::{current_mount_identity, mount_identity};
use lazaret_linux::process::{
@ -148,6 +149,7 @@ pub(crate) enum RecoveryBoundary {
GitRouting,
StateAnchor,
PreparedLower,
SessionTerminals,
PrepareOperations,
ArtifactGc,
FinishGeneration,
@ -265,15 +267,11 @@ where
let generation_id = generation.generation_id;
let snapshot = store.snapshot().map_err(|error| error.to_string())?;
if snapshot
.session_terminals
.mailboxes
.iter()
.any(|terminal| terminal.generation_id() == generation_id)
|| snapshot
.mailboxes
.iter()
.any(|mailbox| mailbox.generation_id == generation_id)
.any(|mailbox| mailbox.generation_id == generation_id)
{
return Err("unfinished terminal or mailbox state cannot be recovered safely".into());
return Err("unfinished mailbox state cannot be recovered safely".into());
}
let session_updates = snapshot
.sessions
@ -300,9 +298,20 @@ where
.snapshot()
.map_err(|error| format!("Participant snapshot failed: {error}"))?
.participants;
let sandbox_container = participants
.iter()
.find(|participant| {
participant.generation_id() == generation_id
&& participant.role() == ParticipantRole::Sandbox
})
.and_then(Participant::ownership)
.and_then(|ownership| match ownership {
ParticipantOwnership::Sandbox(sandbox) => sandbox.container_id,
_ => None,
});
let ownership = runtime_recovery_ownership(&participants, generation)
.map_err(|error| format!("runtime ownership inventory failed: {error}"))?;
let mut owners = seal_runtime_owners(&ownership, open)
let mut owners = seal_runtime_owners(&ownership, &mut *open)
.map_err(|error| format!("runtime owner acquisition failed: {error}"))?;
failpoint(RecoveryBoundary::Publication, generation_id)?;
@ -341,6 +350,36 @@ where
recover_partial_prepared_lower(configuration, store, catalog, generation_id)
.map_err(|error| format!("prepared-lower recovery failed: {error}"))?;
finish_runtime_and_terminal_recovery(
store,
generation_id,
&ownership,
sandbox_container,
&mut owners,
&mut *open,
failpoint,
)?;
failpoint(RecoveryBoundary::FinishGeneration, generation_id)?;
let mut clean = generation.clone();
clean.durable_decision = DurableDecision::Clean;
store
.apply(&[DomainStoreMutation::PutGeneration(clean)])
.map_err(|error| error.to_string())
}
fn finish_runtime_and_terminal_recovery<Open, Failpoint>(
store: &HostCoordinatorDomainStore,
generation_id: GenerationId,
runtime: &RuntimeRecoveryOwnership,
sandbox_container: Option<lazaret_core::identity::ContainerId>,
owners: &mut SealedRuntimeOwners,
open: &mut Open,
failpoint: &mut Failpoint,
) -> Result<(), String>
where
Open: FnMut(PidfsProcessIdentityV1) -> io::Result<PidfsRecoveryMatch>,
Failpoint: FnMut(RecoveryBoundary, GenerationId) -> Result<(), String>,
{
let snapshot = store.snapshot().map_err(|error| error.to_string())?;
if snapshot.participants.iter().any(|participant| {
participant.generation_id() == generation_id
@ -348,12 +387,161 @@ where
}) {
return Err("generation retains non-Clean participants after recovery".into());
}
failpoint(RecoveryBoundary::FinishGeneration, generation_id)?;
let mut clean = generation.clone();
clean.durable_decision = DurableDecision::Clean;
store
.apply(&[DomainStoreMutation::PutGeneration(clean)])
.map_err(|error| error.to_string())
owners
.prove_absent()
.map_err(|error| format!("runtime absence proof failed: {error}"))?;
failpoint(RecoveryBoundary::SessionTerminals, generation_id)?;
recover_session_terminals_after_runtime_absence(
store,
generation_id,
runtime,
sandbox_container,
open,
)
}
fn recover_session_terminals_after_runtime_absence<Open>(
store: &HostCoordinatorDomainStore,
generation_id: GenerationId,
runtime: &RuntimeRecoveryOwnership,
sandbox_container: Option<lazaret_core::identity::ContainerId>,
open: &mut Open,
) -> Result<(), String>
where
Open: FnMut(PidfsProcessIdentityV1) -> io::Result<PidfsRecoveryMatch>,
{
let snapshot = store.snapshot().map_err(|error| error.to_string())?;
for terminal in snapshot
.session_terminals
.iter()
.filter(|terminal| terminal.generation_id() == generation_id)
{
if terminal.state() == SessionTerminalState::Clean {
continue;
}
if !snapshot.sessions.iter().any(|session| {
session.generation_id == generation_id
&& session.session_id == terminal.session_id()
&& matches!(
session.session_phase,
SessionPhase::Closed | SessionPhase::Failed
)
}) {
return Err("Session Terminal has no closed or failed Session".into());
}
require_terminal_runtime_binding(terminal, runtime, sandbox_container)?;
require_terminal_exec_absent(terminal.exec_client(), open)?;
let bare_intent = terminal.state() == SessionTerminalState::Intent
&& terminal.container_id().is_none()
&& terminal.host_coordinator().is_unassigned()
&& terminal.pty_master().is_none()
&& terminal.exec_client().is_unassigned()
&& terminal.guest_process().is_none();
let current = if terminal.state() == SessionTerminalState::ReleaseIntent || bare_intent {
terminal.clone()
} else {
let release = terminal_with_state(terminal, SessionTerminalState::ReleaseIntent)?;
store
.apply(&[DomainStoreMutation::PutSessionTerminal(release.clone())])
.map_err(|error| format!("Session Terminal release failed: {error}"))?;
release
};
let clean = SessionTerminal::snapshot(SessionTerminalFields {
generation_id,
session_id: terminal.session_id(),
state: SessionTerminalState::Clean,
container_id: None,
host_coordinator: PersistedHostProcessOwnership::unassigned(),
pty_master: None,
exec_client: PersistedHostProcessOwnership::unassigned(),
guest_process: None,
})
.map_err(|_| "Session Terminal Clean snapshot is invalid")?;
current
.validate_transition_to_clean_after_absence(&clean)
.map_err(|error| format!("Session Terminal cleanup is invalid: {error}"))?;
store
.apply(&[DomainStoreMutation::CleanSessionTerminalAfterAbsence(clean)])
.map_err(|error| format!("Session Terminal cleanup failed: {error}"))?;
}
Ok(())
}
fn require_terminal_runtime_binding(
terminal: &SessionTerminal,
runtime: &RuntimeRecoveryOwnership,
sandbox_container: Option<lazaret_core::identity::ContainerId>,
) -> Result<(), String> {
if let Some(container_id) = terminal.container_id()
&& Some(container_id) != sandbox_container
{
return Err("Session Terminal container differs from runtime authority".into());
}
if terminal.guest_process().is_some() && terminal.container_id().is_none() {
return Err("Session Terminal guest identity has no container authority".into());
}
match terminal.host_coordinator().authoritative_pidfs() {
Ok(identity) if runtime.process(RuntimeProcessRole::HostCoordinator) == Some(*identity) => {
}
Ok(_) => {
return Err("Session Terminal Host Coordinator differs from runtime authority".into());
}
Err(_) if terminal.pty_master().is_none() => {}
Err(_) => return Err("Session Terminal PTY has no Host Coordinator authority".into()),
}
if let Ok(identity) = terminal.exec_client().authoritative_pidfs()
&& runtime
.processes
.iter()
.any(|owner| owner.identity == *identity)
{
return Err("Session Terminal exec identity aliases runtime authority".into());
}
Ok(())
}
fn require_terminal_exec_absent<Open>(
ownership: &PersistedHostProcessOwnership,
open: &mut Open,
) -> Result<(), String>
where
Open: FnMut(PidfsProcessIdentityV1) -> io::Result<PidfsRecoveryMatch>,
{
let Ok(identity) = ownership.authoritative_pidfs() else {
return Ok(());
};
match open(*identity) {
Ok(PidfsRecoveryMatch::BootMismatch) => Ok(()),
Ok(PidfsRecoveryMatch::Matched(process))
if process.has_exited().map_err(|error| error.to_string())? =>
{
Ok(())
}
Ok(PidfsRecoveryMatch::Matched(_)) => {
Err("Session Terminal exec client remains present".into())
}
Err(error) if error.raw_os_error() == Some(rustix::io::Errno::SRCH.raw_os_error()) => {
Ok(())
}
Err(error) => Err(format!("Session Terminal exec identity refused: {error}")),
}
}
fn terminal_with_state(
terminal: &SessionTerminal,
state: SessionTerminalState,
) -> Result<SessionTerminal, String> {
SessionTerminal::snapshot(SessionTerminalFields {
generation_id: terminal.generation_id(),
session_id: terminal.session_id(),
state,
container_id: terminal.container_id(),
host_coordinator: *terminal.host_coordinator(),
pty_master: terminal.pty_master(),
exec_client: *terminal.exec_client(),
guest_process: terminal.guest_process(),
})
.map_err(|_| "Session Terminal release snapshot is invalid".into())
}
pub(super) fn recover_publication(

136
crates/lazaret-host-coordinator/src/recovery/tests.rs
View file

@ -11,18 +11,22 @@ use lazaret_core::configuration::{
ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_core::identity::{
BootId, BundleDigest, ContainerId, CoordinatorId, DeviceId, HostPidLocator, IdMapDigest,
InodeId, MountId, NetworkSidecarId, PidfsDeviceId, PidfsInodeId, PidfsProcessAuthorityV1,
PolicyId, PrepareOperationId, PreparedArtifactId, ProcessStartTimeTicks, RunscBuildId,
RunscDigest, SandboxId, SelectedManifestDigest, SourceIdentity,
BootId, BundleDigest, ClientId, ContainerId, CoordinatorId, DeviceId, FileDescriptor,
HostPidLocator, IdMapDigest, InodeId, MountId, NetworkSidecarId, PidfsDeviceId, PidfsInodeId,
PidfsProcessAuthorityV1, PolicyId, PrepareOperationId, PreparedArtifactId,
ProcessStartTimeTicks, RunscBuildId, RunscDigest, SandboxId, SelectedManifestDigest,
SessionAnchorId, SessionId, SourceIdentity,
};
use lazaret_core::records::{
ArtifactLease, ArtifactState, DurableDecision, GitRoutingOwnership, HelperOwnership,
PrepareOperation, PrepareStageState, PreparedArtifact, PreparedLowerOwnership,
PublicationOwnership, SandboxOwnership, StateAnchorOwnership,
PrepareOperation, PrepareStageState, PreparedArtifact, PreparedLowerOwnership, PtyIdentity,
PublicationOwnership, SandboxOwnership, Session, SessionPhase, SessionTerminal,
SessionTerminalFields, SessionTerminalState, StateAnchorOwnership,
};
use lazaret_linux::filesystem::{current_mount_identity, try_exclusive_flock};
use lazaret_linux::process::{PidfsRecoveryMatch, match_recovery_process};
use lazaret_linux::process::{
PidfsRecoveryMatch, current_process_identity, match_recovery_process,
};
use lazaret_prepared_lower::PreparedLowerStore;
use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
use lazaret_state_anchor::StateAnchorLayout;
@ -361,6 +365,7 @@ fn recovery_preflight_retries_every_recovery_boundary() {
RecoveryBoundary::GitRouting,
RecoveryBoundary::StateAnchor,
RecoveryBoundary::PreparedLower,
RecoveryBoundary::SessionTerminals,
RecoveryBoundary::PrepareOperations,
RecoveryBoundary::ArtifactGc,
RecoveryBoundary::FinishGeneration,
@ -411,6 +416,52 @@ fn recovery_preflight_opens_every_exact_owner_once() {
assert!(opened.values().all(|count| *count == 1));
}
#[test]
fn recovery_cleans_pre_identity_terminal_after_runtime_absence() {
let fixture = RecoveryFixture::new();
fixture.add_released_terminal(process(7), process(1));
fixture.recover_with(|_, _| Ok(())).unwrap();
let snapshot = fixture.store.snapshot().unwrap();
assert_eq!(
snapshot.generations[0].durable_decision,
DurableDecision::Clean
);
assert_eq!(
snapshot.session_terminals[0].state(),
SessionTerminalState::Clean
);
}
#[test]
fn recovery_retains_pre_identity_terminal_when_exec_client_is_live() {
let fixture = RecoveryFixture::new();
fixture.add_released_terminal(current_process_identity().unwrap(), process(1));
let error = fixture.recover_with(|_, _| Ok(())).unwrap_err();
assert!(error.contains("Session Terminal exec client remains present"));
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
#[test]
fn recovery_retains_terminal_on_runtime_authority_mismatch() {
let fixture = RecoveryFixture::new();
fixture.add_released_terminal(process(7), process(8));
let error = fixture.recover_with(|_, _| Ok(())).unwrap_err();
assert!(error.contains("Host Coordinator differs from runtime authority"));
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
#[test]
fn publication_replacement_and_runsc_root_replacement_refuse_untouched() {
let fixture = RecoveryFixture::new();
@ -544,6 +595,77 @@ impl RecoveryFixture {
)
}
fn add_released_terminal(
&self,
exec_client: PidfsProcessIdentityV1,
host_coordinator: PidfsProcessIdentityV1,
) {
let session = Session {
generation_id: GenerationId::new(1),
session_id: SessionId::new([31; 32]),
cli_id: ClientId::new([32; 32]),
session_anchor_id: SessionAnchorId::new([33; 32]),
session_phase: SessionPhase::New,
};
let mut registering = session.clone();
registering.session_phase = SessionPhase::Registering;
let mut attached = registering.clone();
attached.session_phase = SessionPhase::Attached;
let mut closed = attached.clone();
closed.session_phase = SessionPhase::Closed;
let bare = SessionTerminal::snapshot(SessionTerminalFields {
generation_id: session.generation_id,
session_id: session.session_id,
state: SessionTerminalState::Intent,
container_id: None,
host_coordinator: PersistedHostProcessOwnership::unassigned(),
pty_master: None,
exec_client: PersistedHostProcessOwnership::unassigned(),
guest_process: None,
})
.unwrap();
let partial = SessionTerminal::snapshot(SessionTerminalFields {
generation_id: session.generation_id,
session_id: session.session_id,
state: SessionTerminalState::Intent,
container_id: Some(ContainerId::new([6; 32])),
host_coordinator: PersistedHostProcessOwnership::committed(host_coordinator),
pty_master: Some(PtyIdentity {
master_fd: FileDescriptor::new(17),
device: DeviceId::new(18),
inode: InodeId::new(19),
}),
exec_client: PersistedHostProcessOwnership::staged(exec_client),
guest_process: None,
})
.unwrap();
let release = SessionTerminal::snapshot(SessionTerminalFields {
state: SessionTerminalState::ReleaseIntent,
..SessionTerminalFields {
generation_id: partial.generation_id(),
session_id: partial.session_id(),
state: partial.state(),
container_id: partial.container_id(),
host_coordinator: *partial.host_coordinator(),
pty_master: partial.pty_master(),
exec_client: *partial.exec_client(),
guest_process: partial.guest_process(),
}
})
.unwrap();
self.store
.apply(&[
DomainStoreMutation::PutSession(session),
DomainStoreMutation::PutSession(registering),
DomainStoreMutation::PutSession(attached),
DomainStoreMutation::PutSession(closed),
DomainStoreMutation::PutSessionTerminal(bare),
DomainStoreMutation::PutSessionTerminal(partial),
DomainStoreMutation::PutSessionTerminal(release),
])
.unwrap();
}
#[allow(clippy::too_many_lines)]
fn new() -> Self {
let tree = tempfile::tempdir().unwrap();