Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
132 changes: 132 additions & 0 deletions cmux-tui/crates/cmux-tui-core/src/mux.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5530,6 +5530,76 @@ impl Mux {
}
}

/// A positive terminal-host liveness death proof is evidence that the
/// exact runtime incarnation is gone. Record the interruption before exit
/// cleanup can collapse it into an ordinary detached runtime.
pub(crate) fn terminal_host_liveness_dead(
&self,
surface_id: SurfaceId,
identity: &TerminalHostIdentity,
) -> bool {
if self.shutting_down.load(Ordering::Acquire) {
return false;
}
let Some((runtime_id, host_epoch, lease_generation)) = identity.runtime_attachment_tokens()
else {
return false;
};
let mut registry = self.workspace_registry.lock().unwrap();
let state = self.state.lock().unwrap();
let identity_matches = state
.surfaces
.get(&surface_id)
.or_else(|| state.terminal_runtime_by_id(surface_id))
.and_then(|surface| surface.terminal_host_identity())
.is_some_and(|current| current == *identity);
drop(state);
if !identity_matches {
return false;
}
let Ok(Some(terminal)) = registry.terminal_record(&identity.terminal_id) else {
return false;
};
if terminal.incarnation.as_deref() != Some(identity.incarnation.as_str())
|| matches!(
terminal.lifecycle,
TerminalLifecycle::Exited | TerminalLifecycle::Tombstoned
)
{
return false;
}
let Ok(Some(public_id)) = registry.terminal_resource_id(&identity.terminal_id) else {
return false;
};
let idempotency_key = format!(
"runtime-host-loss-{}-{runtime_id}-{host_epoch}-{lease_generation}-host-liveness-dead",
identity.terminal_id
);
match registry.record_runtime_host_loss_proof(
"cmux-tui-runtime",
&idempotency_key,
&public_id,
runtime_id,
host_epoch,
lease_generation,
"host_liveness_dead",
) {
Ok(commit) => {
if !commit.replayed {
self.publish_journal_event();
}
true
}
Err(error) => {
self.emit(MuxEvent::Status(format!(
"could not persist terminal {} host loss proof: {error}",
identity.terminal_id
)));
false
}
}
}

pub(crate) fn terminal_host_reconnected(
self: &Arc<Self>,
surface_id: SurfaceId,
Expand Down Expand Up @@ -8736,6 +8806,11 @@ impl Mux {
let Some(public_id) = registry.terminal_resource_id(&identity.terminal_id)? else {
return Ok(());
};
if state == "detached"
&& registry.runtime_attachment_state(&public_id)?.as_deref() == Some("interrupted")
{
return Ok(());
}
let idempotency_key =
format!("runtime-attachment-{state}-{}-{runtime_id}", identity.terminal_id);
registry.record_runtime_attachment_update(
Expand Down Expand Up @@ -16417,6 +16492,18 @@ mod tests {
.collect()
}

fn journal_event_payloads(mux: &Mux, kind: &str) -> Vec<Value> {
let registry = mux.workspace_registry.lock().unwrap();
registry
.session_journal_after(0, 1024)
.unwrap()
.records
.into_iter()
.filter(|record| record.kind == kind)
.map(|record| record.payload)
.collect()
}

fn public_request(
mux: &Arc<Mux>,
id: &str,
Expand Down Expand Up @@ -20517,6 +20604,51 @@ mod tests {
);
}

#[cfg(unix)]
#[test]
fn host_liveness_dead_journals_interruption_before_exit_detach() {
const TERMINAL: &str = "00000000000040008000000000000063";
const INCARNATION: &str = "10000000000040008000000000000063";
let mux = test_mux();
let workspace = mux
.create_empty_workspace(
Some("runtime-host-loss-detector".into()),
Some("018f6e21-7b70-7e70-8000-000000001063".into()),
None,
)
.unwrap();
let surface_id =
mux.seed_running_terminal_for_test(TERMINAL, INCARNATION, &workspace.key).unwrap();
let surface = mux.surface(surface_id).unwrap();
let identity =
mux.resource_terminal_host_identity(&surface).expect("test terminal has host identity");
let public_id =
mux.workspace_registry.lock().unwrap().terminal_resource_id(TERMINAL).unwrap().unwrap();

assert!(mux.terminal_host_liveness_dead(surface_id, &identity));
assert!(mux.terminal_host_liveness_dead(surface_id, &identity));
mux.surface_exited(surface_id);

let loss_payloads = journal_event_payloads(&mux, "runtime.host_loss.proven");
assert_eq!(loss_payloads.len(), 1);
assert_eq!(loss_payloads[0]["terminal_id"], public_id.as_str());
assert_eq!(loss_payloads[0]["runtime_id"], INCARNATION);
assert_eq!(loss_payloads[0]["host_epoch"], INCARNATION);
assert_eq!(
loss_payloads[0]["lease_generation"],
runtime_attachment_lease_generation_for_test(INCARNATION)
);
assert_eq!(loss_payloads[0]["proof"], "host_liveness_dead");
let states = runtime_attachment_event_payloads(&mux)
.into_iter()
.map(|payload| payload["state"].as_str().unwrap().to_string())
.collect::<Vec<_>>();
assert_eq!(states, vec!["attached"]);
let kinds = journal_event_kinds(&mux);
assert!(!kinds.iter().any(|kind| kind.starts_with("session.hibernate")));
assert!(!kinds.iter().any(|kind| kind.starts_with("session.recover")));
}

#[cfg(unix)]
#[test]
fn stale_host_exit_does_not_create_first_runtime_attachment_state() {
Expand Down
1 change: 1 addition & 0 deletions cmux-tui/crates/cmux-tui-core/src/surface.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3085,6 +3085,7 @@ impl Surface {
);
pty.stream_progress.notify();
if let Some(mux) = mux.upgrade() {
mux.terminal_host_liveness_dead(surface.id, &identity);
mux.surface_exited(surface.id);
}
return;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1234,3 +1234,20 @@ fn upsert_session_effect_workflow(
)?;
Ok(())
}

impl WorkspaceRegistry {
pub(crate) fn runtime_attachment_state(
&self,
terminal_id: &TerminalPublicId,
) -> anyhow::Result<Option<String>> {
self.connection
.query_row(
"SELECT state FROM journal_runtime_attachment_states
WHERE terminal_id = ?1",
[terminal_id.as_str()],
|row| row.get(0),
)
.optional()
.map_err(Into::into)
}
}
Loading