ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Gate terminal group cleanup on live authority

Browse source
This commit is contained in:
ms2k 2026-07-17 06:17:33 +09:00
commit c5ed009fb8
2 changed files with 355 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

230
crates/lazaret-host-coordinator/src/terminal.rs
View file

@ -57,6 +57,12 @@ pub enum TerminalLoss {
HostCoordinatorLost,
}
#[derive(Clone, Copy)]
enum GuestReleaseAuthority {
Validate,
HostOnly,
}
#[derive(Debug)]
pub enum TerminalError {
Io(io::Error),
@ -279,6 +285,57 @@ impl HostCoordinatorTerminalRecorder {
Ok(route)
}
fn require_guest_signal_authority(
&self,
request: &TerminalAcquisition,
terminal: &SessionTerminal,
) -> Result<(), TerminalError> {
let snapshot = self.store.snapshot()?;
let current = snapshot
.generations
.iter()
.all(|candidate| candidate.generation_id <= request.generation_id);
let generation = snapshot.generations.iter().any(|generation| {
generation.generation_id == request.generation_id
&& generation.coordinator_id == request.coordinator_id
&& generation.durable_decision == DurableDecision::Published
});
let session = snapshot.sessions.iter().any(|session| {
session.generation_id == request.generation_id
&& session.session_id == request.session_id
&& matches!(
session.session_phase,
SessionPhase::Attached | SessionPhase::Closed
)
});
let sandbox = snapshot.participants.iter().any(|participant| {
participant.generation_id() == request.generation_id
&& participant.role() == ParticipantRole::Sandbox
&& participant.state() == ParticipantState::Committed
&& matches!(
participant.ownership(),
Some(ParticipantOwnership::Sandbox(sandbox))
if sandbox.container_id == Some(request.container_id)
&& sandbox.host_coordinator.serviceable_pidfs().ok()
== Some(&request.host_coordinator)
)
});
let terminal = snapshot.session_terminals.iter().any(|record| {
record == terminal && record.state() == SessionTerminalState::ReleaseIntent
});
if snapshot.domain.domain_id == request.domain_id
&& current
&& generation
&& session
&& sandbox
&& terminal
{
Ok(())
} else {
Err(TerminalError::AuthorityMismatch)
}
}
fn begin_release(&self, terminal: SessionTerminal) -> Result<(), TerminalError> {
let snapshot = self.store.snapshot()?;
let mut session = snapshot
@ -426,7 +483,13 @@ impl SessionTerminalEngine {
SessionTerminalEvent::Close(event) => event.session_id,
};
if matches!(event, SessionTerminalEvent::Close(_)) {
return self.close(session_id, generation_id, None, true);
return self.close(
session_id,
generation_id,
None,
true,
GuestReleaseAuthority::Validate,
);
}
let authoritative = self.refresh_route(session_id, generation_id)?;
if authoritative != TerminalRoute::Sandbox {
@ -478,7 +541,13 @@ impl SessionTerminalEngine {
let authoritative = self.refresh_route(session_id, generation_id)?;
let terminal = self.active_mut(session_id, generation_id)?;
if terminal.process.exec_exited()? {
return self.close(session_id, generation_id, None, true);
return self.close(
session_id,
generation_id,
None,
true,
GuestReleaseAuthority::Validate,
);
}
if authoritative != TerminalRoute::Sandbox || terminal.releasing {
return Ok(Vec::new());
@ -486,13 +555,25 @@ impl SessionTerminalEngine {
flush_input(terminal)?;
let mut bytes = [0_u8; MAX_OUTPUT_BYTES];
match terminal.process.read_output(&mut bytes) {
Ok(0) => self.close(session_id, generation_id, Some(TerminalLoss::PtyLost), true),
Ok(0) => self.close(
session_id,
generation_id,
Some(TerminalLoss::PtyLost),
true,
GuestReleaseAuthority::Validate,
),
Ok(length) => Ok(vec![SessionTerminalMessage::Output(TerminalOutput {
session_id,
bytes: bytes[..length].to_vec(),
})]),
Err(error) if error.kind() == io::ErrorKind::WouldBlock => Ok(Vec::new()),
Err(_) => self.close(session_id, generation_id, Some(TerminalLoss::PtyLost), true),
Err(_) => self.close(
session_id,
generation_id,
Some(TerminalLoss::PtyLost),
true,
GuestReleaseAuthority::Validate,
),
}
}
@ -520,7 +601,13 @@ impl SessionTerminalEngine {
loss,
TerminalLoss::ClientDisconnected | TerminalLoss::HostCoordinatorLost
);
self.close(session_id, generation_id, Some(loss), connected)
self.close(
session_id,
generation_id,
Some(loss),
connected,
GuestReleaseAuthority::Validate,
)
}
#[must_use]
@ -533,21 +620,46 @@ impl SessionTerminalEngine {
session_id: SessionId,
generation_id: GenerationId,
) -> Result<TerminalRoute, TerminalError> {
let active_generation = self
.active
.get(&session_id)
.ok_or(TerminalError::UnknownSession)?
.generation_id;
if active_generation != generation_id {
let cleanup = self.close(
session_id,
active_generation,
Some(TerminalLoss::ExecClientLost),
false,
GuestReleaseAuthority::HostOnly,
);
return Err(with_cleanup(TerminalError::StaleGeneration, cleanup));
}
let request = {
let terminal = self.active_mut(session_id, generation_id)?;
terminal.request.clone()
};
match self.recorder.require_authority(&request) {
Ok(TerminalRoute::Detached) | Err(TerminalError::AuthorityMismatch) => {
Ok(TerminalRoute::Detached) => {
let cleanup = self.close(
session_id,
generation_id,
Some(TerminalLoss::ExecClientLost),
false,
GuestReleaseAuthority::HostOnly,
);
Err(with_cleanup(TerminalError::AuthorityMismatch, cleanup))
}
Err(error) => Err(error),
Err(error) => {
let cleanup = self.close(
session_id,
generation_id,
Some(TerminalLoss::ExecClientLost),
false,
GuestReleaseAuthority::HostOnly,
);
Err(with_cleanup(error, cleanup))
}
Ok(route) => {
self.active_mut(session_id, generation_id)?.route = route;
Ok(route)
@ -576,15 +688,36 @@ impl SessionTerminalEngine {
generation_id: GenerationId,
loss: Option<TerminalLoss>,
connected: bool,
authority: GuestReleaseAuthority,
) -> Result<Vec<SessionTerminalMessage>, TerminalError> {
let active_generation = self
.active
.get(&session_id)
.ok_or(TerminalError::UnknownSession)?
.generation_id;
if active_generation != generation_id {
let cleanup = self.close(
session_id,
active_generation,
loss,
false,
GuestReleaseAuthority::HostOnly,
);
return Err(with_cleanup(TerminalError::StaleGeneration, cleanup));
}
let mut terminal = self
.active
.remove(&session_id)
.ok_or(TerminalError::UnknownSession)?;
if terminal.generation_id != generation_id {
self.active.insert(session_id, terminal);
return Err(TerminalError::StaleGeneration);
}
let was_serviceable = matches!(authority, GuestReleaseAuthority::Validate)
&& (matches!(
self.recorder.require_authority(&terminal.request),
Ok(TerminalRoute::Sandbox | TerminalRoute::Prompt | TerminalRoute::HostProcess)
) || (terminal.record.state() == SessionTerminalState::ReleaseIntent
&& self
.recorder
.require_guest_signal_authority(&terminal.request, &terminal.record)
.is_ok()));
let result = (|| {
terminal.releasing = true;
terminal.route = TerminalRoute::Detached;
@ -594,14 +727,23 @@ impl SessionTerminalEngine {
terminal.record = release;
self.cancellation.cancel_for_close(session_id)?;
let (mut output, exit) = release_physical(
let mut authorize_guest = || {
was_serviceable
&& self
.recorder
.require_guest_signal_authority(&terminal.request, &terminal.record)
.is_ok()
};
let (mut output, exit, guest_absent) = release_physical(
terminal.process.as_mut(),
session_id,
connected,
!matches!(loss, Some(TerminalLoss::HostCoordinatorLost)),
&mut authorize_guest,
)?;
let clean = clean_terminal(generation_id, session_id)?;
self.recorder.clean(clean)?;
if guest_absent {
let clean = clean_terminal(generation_id, session_id)?;
self.recorder.clean(clean)?;
}
if connected {
output.push(SessionTerminalMessage::Exit(exit_message(
session_id, loss, exit,
@ -620,11 +762,16 @@ fn release_physical(
process: &mut dyn SessionTerminalProcess,
session_id: SessionId,
connected: bool,
signal_guest: bool,
) -> Result<(Vec<SessionTerminalMessage>, ProcessExit), TerminalError> {
authorize_guest: &mut dyn FnMut() -> bool,
) -> Result<(Vec<SessionTerminalMessage>, ProcessExit, bool), TerminalError> {
let mut failure = None;
if signal_guest {
let mut guest_authorized = authorize_guest();
if guest_authorized {
for signal in ["HUP", "TERM", "KILL"] {
if !authorize_guest() {
guest_authorized = false;
break;
}
match process.group_absent() {
Ok(true) => break,
Ok(false) => {}
@ -633,6 +780,10 @@ fn release_physical(
break;
}
}
if !authorize_guest() {
guest_authorized = false;
break;
}
if let Err(error) = process.signal_group(signal) {
failure.get_or_insert(error);
break;
@ -685,14 +836,19 @@ fn release_physical(
}
}
process.close_pty();
match process.group_absent() {
Ok(true) => {}
Ok(false) => {
failure.get_or_insert(TerminalError::GuestGroupStillPresent);
}
Err(error) => {
failure.get_or_insert(error);
if guest_authorized && authorize_guest() {
match process.group_absent() {
Ok(true) if authorize_guest() => {}
Ok(true) => guest_authorized = false,
Ok(false) => {
failure.get_or_insert(TerminalError::GuestGroupStillPresent);
}
Err(error) => {
failure.get_or_insert(error);
}
}
} else {
guest_authorized = false;
}
if let Some(error) = failure {
return Err(error);
@ -700,6 +856,7 @@ fn release_physical(
Ok((
output,
exit.ok_or_else(|| io::Error::other("exec client exit is unavailable"))?,
guest_authorized,
))
}
@ -951,13 +1108,17 @@ fn cleanup_failed_running_commit(
PersistedHostProcessOwnership::committed(exec_client),
Some(guest),
)?;
let durable = recorder.put(observed.clone()).and_then(|()| {
recorder.put(copy_terminal_state(
&observed,
SessionTerminalState::ReleaseIntent,
)?)
});
let physical = release_physical(process, request.session_id, false, true).map(|_| ());
let release = copy_terminal_state(&observed, SessionTerminalState::ReleaseIntent)?;
let durable = recorder
.put(observed)
.and_then(|()| recorder.put(release.clone()));
let mut authorize_guest = || {
recorder
.require_guest_signal_authority(request, &release)
.is_ok()
};
let physical = release_physical(process, request.session_id, false, &mut authorize_guest)
.map(|(_, _, guest_absent)| guest_absent);
match (durable, physical) {
(Err(durable), Err(physical)) => {
return Err(TerminalError::AcquisitionCleanup {
@ -965,8 +1126,9 @@ fn cleanup_failed_running_commit(
cleanup: Some(Box::new(physical)),
});
}
(Err(error), Ok(())) | (Ok(()), Err(error)) => return Err(error),
(Ok(()), Ok(())) => {}
(Err(error), Ok(_)) | (Ok(()), Err(error)) => return Err(error),
(Ok(()), Ok(false)) => return Ok(()),
(Ok(()), Ok(true)) => {}
}
recorder.clean(clean_terminal(request.generation_id, request.session_id)?)
}

173
crates/lazaret-host-coordinator/src/terminal/tests.rs
View file

@ -235,6 +235,7 @@ struct MockGuestState {
group_present: bool,
leader_present: bool,
identity_mismatch: bool,
clears_after_signal: &'static str,
}
#[derive(Debug)]
@ -258,6 +259,7 @@ enum StoreFailurePoint {
struct FailingRunningStore {
inner: Arc<HostCoordinatorDomainStore>,
failure: Mutex<Option<(StoreFailurePoint, bool)>>,
snapshot_failure: Mutex<bool>,
}
impl FailingRunningStore {
@ -265,16 +267,29 @@ impl FailingRunningStore {
Self {
inner,
failure: Mutex::new(None),
snapshot_failure: Mutex::new(false),
}
}
fn arm(&self, point: StoreFailurePoint, persistent: bool) {
*self.failure.lock().unwrap() = Some((point, persistent));
}
fn arm_snapshot_failure(&self) {
*self.snapshot_failure.lock().unwrap() = true;
}
}
impl TerminalDomainStore for FailingRunningStore {
fn snapshot(&self) -> Result<lazaret_core::records::DomainStore, DomainStoreError> {
let mut failure = self.snapshot_failure.lock().unwrap();
if *failure {
*failure = false;
return Err(DomainStoreError::Io(io::Error::other(
"injected authority read failure",
)));
}
drop(failure);
self.inner.snapshot()
}
@ -345,7 +360,7 @@ impl SessionTerminalProcess for MockProcess {
return Err(TerminalError::GuestIdentityMismatch);
}
state.log.lock().unwrap().push(format!("signal:{signal}"));
if signal == "HUP" {
if signal == state.guest.clears_after_signal {
state.guest.group_present = false;
}
Ok(())
@ -446,6 +461,7 @@ impl SessionTerminalFactory for MockFactory {
group_present: true,
leader_present: true,
identity_mismatch: false,
clears_after_signal: "HUP",
},
resources: MockResourceState {
exec_exited: false,
@ -489,12 +505,20 @@ fn setup(ids: &[SessionId]) -> (Fixture, Arc<MockFactory>, SessionTerminalEngine
}
fn engine_for(fixture: &Fixture) -> (Arc<MockFactory>, SessionTerminalEngine) {
let store: Arc<dyn TerminalDomainStore> = fixture.store.clone();
engine_for_store(store)
}
fn engine_for_store(
store: Arc<dyn TerminalDomainStore>,
) -> (Arc<MockFactory>, SessionTerminalEngine) {
let factory = Arc::new(MockFactory::default());
let engine = SessionTerminalEngine::new(
Arc::clone(&fixture.store),
factory.clone(),
Arc::new(Cancellation(Arc::clone(&factory.log))),
);
let engine = SessionTerminalEngine {
recorder: HostCoordinatorTerminalRecorder::with_store(store),
factory: factory.clone(),
cancellation: Arc::new(Cancellation(Arc::clone(&factory.log))),
active: HashMap::new(),
};
(factory, engine)
}
@ -502,6 +526,17 @@ fn id(value: u8) -> SessionId {
SessionId::new(bytes(value))
}
fn assert_no_group_signal(factory: &MockFactory) {
assert!(
factory
.log
.lock()
.unwrap()
.iter()
.all(|action| !action.starts_with("signal:"))
);
}
#[test]
fn acquisition_refuses_authority_outside_the_committed_sandbox() {
let session = id(6);
@ -552,7 +587,7 @@ fn acquisition_requires_attached_current_session() {
fn durable_detach_failure_and_sandbox_release_revoke_active_terminal() {
for revoke in ["closed", "failed", "sandbox-release"] {
let session = id(6);
let (fixture, _factory, mut engine) = setup(&[session]);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
match revoke {
"closed" => fixture.set_session_phase(session, SessionPhase::Closed),
@ -571,13 +606,94 @@ fn durable_detach_failure_and_sandbox_release_revoke_active_terminal() {
Err(TerminalError::AuthorityMismatch)
));
assert!(!engine.contains(session));
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::Clean
SessionTerminalState::ReleaseIntent
);
}
}
#[test]
fn domain_container_hc_and_authority_read_mismatch_use_host_only_release() {
for mismatch in ["domain", "container", "host-coordinator"] {
let session = id(6);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
let request = &mut engine.active.get_mut(&session).unwrap().request;
match mismatch {
"domain" => request.domain_id = DomainId::new(bytes(80)),
"container" => request.container_id = ContainerId::new(bytes(81)),
"host-coordinator" => request.host_coordinator = process(800),
_ => unreachable!(),
}
assert!(matches!(
engine.handle(
GenerationId::new(7),
SessionTerminalEvent::Input(TerminalInput {
session_id: session,
bytes: b"refused".to_vec(),
}),
),
Err(TerminalError::AuthorityMismatch)
));
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
let session = id(6);
let fixture = Fixture::new(&[session]);
let store = Arc::new(FailingRunningStore::new(Arc::clone(&fixture.store)));
let (factory, mut engine) = engine_for_store(store.clone());
engine.acquire(&acquisition(session)).unwrap();
store.arm_snapshot_failure();
assert!(matches!(
engine.poll(session, GenerationId::new(7)),
Err(TerminalError::Store(DomainStoreError::Io(_)))
));
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
#[test]
fn explicit_close_and_loss_revalidate_before_guest_cleanup() {
let session = id(6);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
fixture.release_sandbox();
engine
.handle(
GenerationId::new(7),
SessionTerminalEvent::Close(lazaret_core::records::SessionClose {
session_id: session,
}),
)
.unwrap();
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
fixture.add_newer_generation();
engine
.lose(session, GenerationId::new(7), TerminalLoss::PtyLost)
.unwrap();
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
#[test]
fn stale_caller_and_superseded_generation_refuse_without_cross_generation_action() {
let session = id(6);
@ -593,15 +709,26 @@ fn stale_caller_and_superseded_generation_refuse_without_cross_generation_action
),
Err(TerminalError::StaleGeneration)
));
assert!(factory.log.lock().unwrap().is_empty());
assert!(engine.contains(session));
assert_no_group_signal(&factory);
assert!(!engine.contains(session));
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
fixture.add_newer_generation();
assert!(matches!(
engine.poll(session, GenerationId::new(7)),
Err(TerminalError::AuthorityMismatch)
));
assert!(!engine.contains(session));
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::ReleaseIntent
);
}
#[test]
@ -789,6 +916,7 @@ fn running_store_failures_release_every_live_partial_handle() {
group_present: true,
leader_present: true,
identity_mismatch: false,
clears_after_signal: "HUP",
},
resources: MockResourceState {
exec_exited: false,
@ -818,7 +946,16 @@ fn running_store_failures_release_every_live_partial_handle() {
error => panic!("unexpected cleanup result: {error}"),
}
let state = state.lock().unwrap();
assert!(!state.guest.group_present);
assert_eq!(state.guest.group_present, persistent);
assert_eq!(
state
.log
.lock()
.unwrap()
.iter()
.any(|action| action.starts_with("signal:")),
!persistent
);
assert!(state.resources.exec_exited);
assert!(state.resources.exec_reaped);
assert!(!state.resources.pty_open);
@ -886,6 +1023,7 @@ fn close_cancels_route_then_bounds_drain_and_emits_one_final_exit() {
let session = id(6);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
factory.state(0).lock().unwrap().guest.clears_after_signal = "KILL";
factory
.state(0)
.lock()
@ -911,10 +1049,16 @@ fn close_cancels_route_then_bounds_drain_and_emits_one_final_exit() {
assert!(
matches!(messages.last(), Some(SessionTerminalMessage::Exit(exit)) if exit.exit_code() == Some(17))
);
let log = factory.log.lock().unwrap();
assert_eq!(log[0], "cancel-route");
assert_eq!(
&factory.log.lock().unwrap()[..2],
&["cancel-route", "signal:HUP"]
log.iter()
.filter(|action| action.starts_with("signal:"))
.map(String::as_str)
.collect::<Vec<_>>(),
["signal:HUP", "signal:TERM", "signal:KILL"]
);
drop(log);
assert!(!engine.contains(session));
assert!(matches!(
engine.handle(
@ -951,7 +1095,7 @@ fn disconnect_component_loss_and_hard_hc_loss_never_reattach() {
}
let session = id(6);
let (fixture, _factory, mut engine) = setup(&[session]);
let (fixture, factory, mut engine) = setup(&[session]);
engine.acquire(&acquisition(session)).unwrap();
assert!(
engine
@ -964,6 +1108,7 @@ fn disconnect_component_loss_and_hard_hc_loss_never_reattach() {
.is_empty()
);
assert!(!engine.contains(session));
assert_no_group_signal(&factory);
assert_eq!(
fixture.store.snapshot().unwrap().session_terminals[0].state(),
SessionTerminalState::Running