clarify terminal persistence cleanup ownership

This commit is contained in:
Anton Panasenko
2026-07-06 20:47:12 -07:00
parent 05eb6e5e8b
commit fbf5bfd1a9
3 changed files with 14 additions and 4 deletions
+4 -3
View File
@@ -630,9 +630,10 @@ async fn shutdown_thread_persistence(
return ThreadPersistenceShutdown::Closed;
};
let failure_context = failure_context.to_owned();
// Terminal cleanup must outlive the submission loop. If the loop is canceled while a
// graceful shutdown is still waiting on I/O, the same task still reaches the discard
// fallback and releases the live-writer lease.
// Await a child task so ordinary shutdown still waits for cleanup before completion. If the
// submission loop is canceled while graceful shutdown is waiting on I/O, dropping the
// JoinHandle detaches the child; it still reaches the discard fallback and releases the
// live-writer lease.
tokio::spawn(async move {
if let Err(err) = live_thread.shutdown().await {
warn!("{failure_context}: {err}");
+5 -1
View File
@@ -60,12 +60,16 @@ impl LiveThreadInitGuard {
}
pub async fn discard(&mut self) {
let Some(live_thread) = self.live_thread.take() else {
let Some(live_thread) = self.live_thread.as_ref().cloned() else {
return;
};
if let Err(err) = live_thread.discard().await {
warn!("failed to discard thread persistence for failed session init: {err}");
return;
}
// Keep the guard armed until discard completes so cancellation still leaves Drop able to
// retry terminal cleanup.
self.live_thread = None;
}
}
@@ -162,6 +162,9 @@ pub(super) async fn shutdown_thread(
) -> ThreadStoreResult<()> {
let recorder = store.live_recorder(thread_id).await?;
let store = store.clone();
// The writer can close before this future removes its live-recorder entry. Keep those steps
// in one detached task so canceling an arbitrary ThreadStore caller cannot strand a closed
// recorder that blocks a later resume.
tokio::spawn(async move {
let rollout_path = recorder.rollout_path().to_path_buf();
recorder.shutdown().await.map_err(thread_store_io_error)?;
@@ -190,6 +193,8 @@ pub(super) async fn discard_thread(
Err(ThreadStoreError::ThreadNotFound { .. }) => return Ok(()),
Err(err) => return Err(err),
};
// Discard closes stale recorder clones before removing the lease. Keep that sequence alive if
// an arbitrary ThreadStore caller is canceled between the two steps.
spawn_discard_cleanup(store.clone(), thread_id, recorder)
.await
.map_err(|err| ThreadStoreError::Internal {