Harden session terminal release authority

crates/lazaret-host-coordinator/src/terminal/tests.rs

@@ -62,16 +62,25 @@ impl Fixture {
             HostCoordinatorDomainStore::open(lock, &directory.path().join("domain.redb"), &domain)
                 .unwrap(),
         );
+        let mut generation = Generation {
+            domain_id: domain.domain_id,
+            generation_id: GenerationId::new(7),
+            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: DurableDecision::None,
+        };
         store
-            .apply(&[DomainStoreMutation::PutGeneration(Generation {
-                domain_id: domain.domain_id,
-                generation_id: GenerationId::new(7),
-                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: DurableDecision::None,
-            })])
+            .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])
+            .unwrap();
+        generation.durable_decision = DurableDecision::Preparing;
+        store
+            .apply(&[DomainStoreMutation::PutGeneration(generation.clone())])
+            .unwrap();
+        generation.durable_decision = DurableDecision::Published;
+        store
+            .apply(&[DomainStoreMutation::PutGeneration(generation)])
             .unwrap();
         store
             .apply(&[DomainStoreMutation::PutParticipant(
@@ -123,42 +132,175 @@ impl Fixture {
                 sandbox,
             ))])
             .unwrap();
-        for (index, session_id) in session_ids.iter().enumerate() {
-            let mut session = Session {
-                generation_id: GenerationId::new(7),
-                session_id: *session_id,
-                cli_id: ClientId::new(bytes(20 + u8::try_from(index).unwrap())),
-                session_anchor_id: SessionAnchorId::new(bytes(30 + u8::try_from(index).unwrap())),
-                session_phase: SessionPhase::Registering,
-            };
-            store
-                .apply(&[DomainStoreMutation::PutSession(session.clone())])
-                .unwrap();
-            session.session_phase = SessionPhase::Attached;
-            store
-                .apply(&[DomainStoreMutation::PutSession(session)])
-                .unwrap();
-        }
+        insert_attached_sessions(&store, session_ids);
         Self {
             _directory: directory,
             store,
         }
     }
+
+    fn insert_session(&self, session_id: SessionId, phase: SessionPhase) {
+        self.store
+            .apply(&[DomainStoreMutation::PutSession(Session {
+                generation_id: GenerationId::new(7),
+                session_id,
+                cli_id: ClientId::new(bytes(20)),
+                session_anchor_id: SessionAnchorId::new(bytes(30)),
+                session_phase: phase,
+            })])
+            .unwrap();
+    }
+
+    fn set_session_phase(&self, session_id: SessionId, phase: SessionPhase) {
+        let mut session = self
+            .store
+            .snapshot()
+            .unwrap()
+            .sessions
+            .into_iter()
+            .find(|session| session.session_id == session_id)
+            .unwrap();
+        session.session_phase = phase;
+        self.store
+            .apply(&[DomainStoreMutation::PutSession(session)])
+            .unwrap();
+    }
+
+    fn release_sandbox(&self) {
+        let sandbox = self
+            .store
+            .snapshot()
+            .unwrap()
+            .participants
+            .into_iter()
+            .find(|participant| participant.role() == ParticipantRole::Sandbox)
+            .unwrap();
+        self.store
+            .apply(&[DomainStoreMutation::PutParticipant(
+                Participant::snapshot(
+                    sandbox.generation_id(),
+                    sandbox.role(),
+                    ParticipantState::ReleaseIntent,
+                    sandbox.ownership().cloned(),
+                )
+                .unwrap(),
+            )])
+            .unwrap();
+    }
+
+    fn add_newer_generation(&self) {
+        self.store
+            .apply(&[DomainStoreMutation::PutGeneration(Generation {
+                domain_id: DomainId::new(bytes(1)),
+                generation_id: GenerationId::new(8),
+                configuration_fingerprint: ConfigurationFingerprint::new(bytes(42)),
+                coordinator_id: CoordinatorId::new(bytes(43)),
+                network_sidecar_id: NetworkSidecarId::new(bytes(44)),
+                sandbox_id: SandboxId::new(bytes(45)),
+                durable_decision: DurableDecision::None,
+            })])
+            .unwrap();
+    }
+}
+
+fn insert_attached_sessions(store: &HostCoordinatorDomainStore, session_ids: &[SessionId]) {
+    for (index, session_id) in session_ids.iter().enumerate() {
+        let mut session = Session {
+            generation_id: GenerationId::new(7),
+            session_id: *session_id,
+            cli_id: ClientId::new(bytes(20 + u8::try_from(index).unwrap())),
+            session_anchor_id: SessionAnchorId::new(bytes(30 + u8::try_from(index).unwrap())),
+            session_phase: SessionPhase::Registering,
+        };
+        store
+            .apply(&[DomainStoreMutation::PutSession(session.clone())])
+            .unwrap();
+        session.session_phase = SessionPhase::Attached;
+        store
+            .apply(&[DomainStoreMutation::PutSession(session)])
+            .unwrap();
+    }
 }
 
 #[derive(Debug)]
 struct MockState {
     output: VecDeque<Vec<u8>>,
     log: Arc<Mutex<Vec<String>>>,
+    guest: MockGuestState,
+    resources: MockResourceState,
+}
+
+#[derive(Debug)]
+struct MockGuestState {
     group_present: bool,
+    leader_present: bool,
     identity_mismatch: bool,
+}
+
+#[derive(Debug)]
+struct MockResourceState {
     exec_exited: bool,
+    exec_reaped: bool,
+    pty_open: bool,
 }
 
 struct MockProcess {
     state: Arc<Mutex<MockState>>,
 }
 
+#[derive(Clone, Copy)]
+enum StoreFailurePoint {
+    Snapshot,
+    Write,
+    Commit,
+}
+
+struct FailingRunningStore {
+    inner: Arc<HostCoordinatorDomainStore>,
+    failure: Mutex<Option<(StoreFailurePoint, bool)>>,
+}
+
+impl FailingRunningStore {
+    fn new(inner: Arc<HostCoordinatorDomainStore>) -> Self {
+        Self {
+            inner,
+            failure: Mutex::new(None),
+        }
+    }
+
+    fn arm(&self, point: StoreFailurePoint, persistent: bool) {
+        *self.failure.lock().unwrap() = Some((point, persistent));
+    }
+}
+
+impl TerminalDomainStore for FailingRunningStore {
+    fn snapshot(&self) -> Result<lazaret_core::records::DomainStore, DomainStoreError> {
+        self.inner.snapshot()
+    }
+
+    fn apply(&self, mutations: &[DomainStoreMutation]) -> Result<(), DomainStoreError> {
+        let mut failure = self.failure.lock().unwrap();
+        if let Some((point, persistent)) = *failure {
+            let running = mutations.iter().any(|mutation| {
+                matches!(mutation, DomainStoreMutation::PutSessionTerminal(terminal) if terminal.state() == SessionTerminalState::Running)
+            });
+            if running || persistent {
+                if !persistent {
+                    *failure = None;
+                }
+                let point = match point {
+                    StoreFailurePoint::Snapshot => "injected snapshot failure",
+                    StoreFailurePoint::Write => "injected write failure",
+                    StoreFailurePoint::Commit => "injected commit failure",
+                };
+                return Err(DomainStoreError::Io(io::Error::other(point)));
+            }
+        }
+        drop(failure);
+        self.inner.apply(mutations)
+    }
+}
+
 impl SessionTerminalProcess for MockProcess {
     fn write_input(&mut self, bytes: &[u8]) -> io::Result<usize> {
         self.state
@@ -194,44 +336,42 @@ impl SessionTerminalProcess for MockProcess {
     }
 
     fn exec_exited(&self) -> io::Result<bool> {
-        Ok(self.state.lock().unwrap().exec_exited)
+        Ok(self.state.lock().unwrap().resources.exec_exited)
     }
 
     fn signal_group(&self, signal: &str) -> Result<(), TerminalError> {
         let mut state = self.state.lock().unwrap();
-        if state.identity_mismatch {
+        if !state.guest.leader_present || state.guest.identity_mismatch {
             return Err(TerminalError::GuestIdentityMismatch);
         }
         state.log.lock().unwrap().push(format!("signal:{signal}"));
         if signal == "HUP" {
-            state.group_present = false;
+            state.guest.group_present = false;
         }
         Ok(())
     }
 
     fn group_absent(&self) -> Result<bool, TerminalError> {
         let state = self.state.lock().unwrap();
-        if state.identity_mismatch {
+        if state.guest.identity_mismatch {
             return Err(TerminalError::GuestIdentityMismatch);
         }
-        Ok(!state.group_present)
+        Ok(!state.guest.group_present)
     }
 
     fn terminate_exec(&self) -> io::Result<()> {
         let mut state = self.state.lock().unwrap();
         state.log.lock().unwrap().push("terminate-exec".into());
-        state.exec_exited = true;
+        state.resources.exec_exited = true;
         Ok(())
     }
 
     fn reap_exec(&mut self) -> io::Result<ProcessExit> {
-        self.state
-            .lock()
-            .unwrap()
-            .log
-            .lock()
-            .unwrap()
-            .push("reap-exec".into());
+        let mut state = self.state.lock().unwrap();
+        if !state.resources.exec_reaped {
+            state.log.lock().unwrap().push("reap-exec".into());
+            state.resources.exec_reaped = true;
+        }
         Ok(ProcessExit {
             exit_code: Some(17),
             terminating_signal: None,
@@ -239,8 +379,9 @@ impl SessionTerminalProcess for MockProcess {
     }
 
     fn close_pty(&mut self) {
-        let state = self.state.lock().unwrap();
+        let mut state = self.state.lock().unwrap();
         state.log.lock().unwrap().push("close-pty".into());
+        state.resources.pty_open = false;
     }
 }
 
@@ -301,9 +442,16 @@ impl SessionTerminalFactory for MockFactory {
         let state = Arc::new(Mutex::new(MockState {
             output: VecDeque::new(),
             log: Arc::clone(&self.log),
-            group_present: true,
-            identity_mismatch: false,
-            exec_exited: false,
+            guest: MockGuestState {
+                group_present: true,
+                leader_present: true,
+                identity_mismatch: false,
+            },
+            resources: MockResourceState {
+                exec_exited: false,
+                exec_reaped: false,
+                pty_open: true,
+            },
         }));
         self.states.lock().unwrap().push(Arc::clone(&state));
         Ok((running, Box::new(MockProcess { state })))
@@ -336,13 +484,18 @@ fn acquisition(session_id: SessionId) -> TerminalAcquisition {
 
 fn setup(ids: &[SessionId]) -> (Fixture, Arc<MockFactory>, SessionTerminalEngine) {
     let fixture = Fixture::new(ids);
+    let (factory, engine) = engine_for(&fixture);
+    (fixture, factory, engine)
+}
+
+fn engine_for(fixture: &Fixture) -> (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))),
     );
-    (fixture, factory, engine)
+    (factory, engine)
 }
 
 fn id(value: u8) -> SessionId {
@@ -369,6 +522,88 @@ fn acquisition_refuses_authority_outside_the_committed_sandbox() {
     );
 }
 
+#[test]
+fn acquisition_requires_attached_current_session() {
+    for phase in [
+        SessionPhase::New,
+        SessionPhase::Waiting,
+        SessionPhase::Registering,
+    ] {
+        let session = id(6);
+        let fixture = Fixture::new(&[]);
+        fixture.insert_session(session, phase);
+        let (_factory, mut engine) = engine_for(&fixture);
+        assert!(matches!(
+            engine.acquire(&acquisition(session)),
+            Err(TerminalError::AuthorityMismatch)
+        ));
+        assert!(
+            fixture
+                .store
+                .snapshot()
+                .unwrap()
+                .session_terminals
+                .is_empty()
+        );
+    }
+}
+
+#[test]
+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]);
+        engine.acquire(&acquisition(session)).unwrap();
+        match revoke {
+            "closed" => fixture.set_session_phase(session, SessionPhase::Closed),
+            "failed" => fixture.set_session_phase(session, SessionPhase::Failed),
+            "sandbox-release" => fixture.release_sandbox(),
+            _ => unreachable!(),
+        }
+        assert!(matches!(
+            engine.handle(
+                GenerationId::new(7),
+                SessionTerminalEvent::Input(TerminalInput {
+                    session_id: session,
+                    bytes: b"refused".to_vec(),
+                }),
+            ),
+            Err(TerminalError::AuthorityMismatch)
+        ));
+        assert!(!engine.contains(session));
+        assert_eq!(
+            fixture.store.snapshot().unwrap().session_terminals[0].state(),
+            SessionTerminalState::Clean
+        );
+    }
+}
+
+#[test]
+fn stale_caller_and_superseded_generation_refuse_without_cross_generation_action() {
+    let session = id(6);
+    let (fixture, factory, mut engine) = setup(&[session]);
+    engine.acquire(&acquisition(session)).unwrap();
+    assert!(matches!(
+        engine.handle(
+            GenerationId::new(8),
+            SessionTerminalEvent::Signal(TerminalSignal {
+                session_id: session,
+                signal: TerminalSignalKind::Interrupt,
+            }),
+        ),
+        Err(TerminalError::StaleGeneration)
+    ));
+    assert!(factory.log.lock().unwrap().is_empty());
+    assert!(engine.contains(session));
+
+    fixture.add_newer_generation();
+    assert!(matches!(
+        engine.poll(session, GenerationId::new(7)),
+        Err(TerminalError::AuthorityMismatch)
+    ));
+    assert!(!engine.contains(session));
+}
+
 #[test]
 fn resize_and_terminal_signals_use_the_exact_group_map() {
     let session = id(6);
@@ -411,9 +646,9 @@ fn resize_and_terminal_signals_use_the_exact_group_map() {
 #[test]
 fn changed_guest_identity_refuses_numeric_group_action() {
     let session = id(6);
-    let (_fixture, factory, mut engine) = setup(&[session]);
+    let (fixture, factory, mut engine) = setup(&[session]);
     engine.acquire(&acquisition(session)).unwrap();
-    factory.state(0).lock().unwrap().identity_mismatch = true;
+    factory.state(0).lock().unwrap().guest.identity_mismatch = true;
     assert!(matches!(
         engine.handle(
             GenerationId::new(7),
@@ -425,13 +660,186 @@ fn changed_guest_identity_refuses_numeric_group_action() {
         Err(TerminalError::GuestIdentityMismatch)
     ));
     assert!(factory.log.lock().unwrap().is_empty());
+    assert!(matches!(
+        engine.handle(
+            GenerationId::new(7),
+            SessionTerminalEvent::Close(lazaret_core::records::SessionClose {
+                session_id: session,
+            }),
+        ),
+        Err(TerminalError::GuestIdentityMismatch)
+    ));
+    assert_eq!(
+        fixture.store.snapshot().unwrap().session_terminals[0].state(),
+        SessionTerminalState::ReleaseIntent
+    );
+}
+
+#[test]
+fn absent_leader_with_live_group_retains_release_until_group_is_empty() {
+    let session = id(6);
+    let (fixture, factory, mut engine) = setup(&[session]);
+    engine.acquire(&acquisition(session)).unwrap();
+    {
+        let state = factory.state(0);
+        let mut state = state.lock().unwrap();
+        state.guest.leader_present = false;
+        state.guest.group_present = true;
+    }
+    assert!(matches!(
+        engine.handle(
+            GenerationId::new(7),
+            SessionTerminalEvent::Close(lazaret_core::records::SessionClose {
+                session_id: session,
+            }),
+        ),
+        Err(TerminalError::GuestIdentityMismatch)
+    ));
+    let state = factory.state(0);
+    assert!(state.lock().unwrap().resources.exec_reaped);
+    assert!(!state.lock().unwrap().resources.pty_open);
+    assert_eq!(
+        fixture.store.snapshot().unwrap().session_terminals[0].state(),
+        SessionTerminalState::ReleaseIntent
+    );
+
+    state.lock().unwrap().guest.group_present = false;
+    engine
+        .handle(
+            GenerationId::new(7),
+            SessionTerminalEvent::Close(lazaret_core::records::SessionClose {
+                session_id: session,
+            }),
+        )
+        .unwrap();
+    assert_eq!(
+        fixture.store.snapshot().unwrap().session_terminals[0].state(),
+        SessionTerminalState::Clean
+    );
+}
+
+#[test]
+#[allow(clippy::too_many_lines)]
+fn running_store_failures_release_every_live_partial_handle() {
+    for point in [
+        StoreFailurePoint::Snapshot,
+        StoreFailurePoint::Write,
+        StoreFailurePoint::Commit,
+    ] {
+        for persistent in [false, true] {
+            let session = id(6);
+            let fixture = Fixture::new(&[session]);
+            let store = Arc::new(FailingRunningStore::new(Arc::clone(&fixture.store)));
+            let recorder = HostCoordinatorTerminalRecorder::with_store(store.clone());
+            let request = acquisition(session);
+            recorder
+                .put(
+                    terminal_snapshot(
+                        &request,
+                        SessionTerminalState::Intent,
+                        PersistedHostProcessOwnership::unassigned(),
+                        None,
+                        PersistedHostProcessOwnership::unassigned(),
+                        None,
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+            let pty = PtyIdentity {
+                master_fd: FileDescriptor::new(9),
+                device: DeviceId::new(10),
+                inode: InodeId::new(11),
+            };
+            recorder
+                .put(
+                    staged_terminal(
+                        &request,
+                        pty,
+                        StagedProcessOwnership::v1(
+                            request.domain_id,
+                            request.generation_id,
+                            request.coordinator_id,
+                            lazaret_core::records::HostProcessOwnershipSlot::SessionTerminalExecClient(
+                                request.session_id,
+                            ),
+                            process(90),
+                        ),
+                    )
+                    .unwrap(),
+                )
+                .unwrap();
+            let guest = GuestProcessIdentity {
+                leader_pid: GuestProcessId::new(41),
+                leader_start_time: GuestProcessStartTimeTicks::new(101),
+                process_group_id: GuestProcessGroupId::new(41),
+            };
+            let running = terminal_snapshot(
+                &request,
+                SessionTerminalState::Running,
+                PersistedHostProcessOwnership::committed(request.host_coordinator),
+                Some(pty),
+                PersistedHostProcessOwnership::committed(process(90)),
+                Some(guest),
+            )
+            .unwrap();
+            let state = Arc::new(Mutex::new(MockState {
+                output: VecDeque::new(),
+                log: Arc::new(Mutex::new(Vec::new())),
+                guest: MockGuestState {
+                    group_present: true,
+                    leader_present: true,
+                    identity_mismatch: false,
+                },
+                resources: MockResourceState {
+                    exec_exited: false,
+                    exec_reaped: false,
+                    pty_open: true,
+                },
+            }));
+            let mut process_handle = MockProcess {
+                state: Arc::clone(&state),
+            };
+            store.arm(point, persistent);
+            let error = commit_running_or_cleanup(
+                &recorder,
+                &request,
+                running,
+                pty,
+                process(90),
+                guest,
+                &mut process_handle,
+            )
+            .unwrap_err();
+            match error {
+                TerminalError::AcquisitionCleanup { cleanup: None, .. } if !persistent => {}
+                TerminalError::AcquisitionCleanup {
+                    cleanup: Some(_), ..
+                } if persistent => {}
+                error => panic!("unexpected cleanup result: {error}"),
+            }
+            let state = state.lock().unwrap();
+            assert!(!state.guest.group_present);
+            assert!(state.resources.exec_exited);
+            assert!(state.resources.exec_reaped);
+            assert!(!state.resources.pty_open);
+            drop(state);
+            assert_eq!(
+                fixture.store.snapshot().unwrap().session_terminals[0].state(),
+                if persistent {
+                    SessionTerminalState::Intent
+                } else {
+                    SessionTerminalState::Clean
+                }
+            );
+        }
+    }
 }
 
 #[test]
 fn merged_output_is_ordered_and_prompt_route_isolated_by_session() {
     let first = id(6);
     let second = id(7);
-    let (_fixture, factory, mut engine) = setup(&[first, second]);
+    let (fixture, factory, mut engine) = setup(&[first, second]);
     engine.acquire(&acquisition(first)).unwrap();
     engine.acquire(&acquisition(second)).unwrap();
     factory
@@ -446,6 +854,7 @@ fn merged_output_is_ordered_and_prompt_route_isolated_by_session() {
         .unwrap()
         .output
         .push_back(b"other".to_vec());
+    fixture.set_session_phase(first, SessionPhase::Prompting);
     engine
         .set_route(first, GenerationId::new(7), TerminalRoute::Prompt)
         .unwrap();
@@ -457,6 +866,7 @@ fn merged_output_is_ordered_and_prompt_route_isolated_by_session() {
             bytes: b"other".to_vec(),
         })]
     );
+    fixture.set_session_phase(first, SessionPhase::Attached);
     engine
         .set_route(first, GenerationId::new(7), TerminalRoute::Sandbox)
         .unwrap();