File size: 1,982 Bytes
afa0cbf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | //! Saves persistent root thread IDs and restores them through the shared internal resume path.
//! Recovery uses normal cold-resume semantics without delaying readiness for runtime loading.
//! Already-loaded runtimes remain owned by their current clients.
use std::io;
use std::path::PathBuf;
use codex_app_server_transport::daemon_recovery;
pub(crate) async fn snapshot(
path: PathBuf,
saved: daemon_recovery::RecoverySnapshot,
) -> io::Result<()> {
// A forced exit must not wait for file I/O in Tokio's blocking pool.
let (result_tx, result_rx) = tokio::sync::oneshot::channel();
std::thread::Builder::new()
.name("daemon-snapshot".into())
.spawn(move || {
let result = daemon_recovery::write_snapshot(&path, &saved);
if result.is_err()
&& let Err(err) = std::fs::remove_file(&path)
&& err.kind() != io::ErrorKind::NotFound
{
tracing::warn!("failed to clear stale daemon recovery file: {err}");
}
let _ = result_tx.send(result);
})?;
result_rx.await.map_err(io::Error::other)?
}
/// Consume the handoff before serving requests, then restore runtimes in the background.
pub(crate) async fn start_recovery(
path: PathBuf,
processor: std::sync::Arc<crate::message_processor::MessageProcessor>,
) -> io::Result<tokio::task::JoinHandle<()>> {
let candidates = tokio::task::spawn_blocking(move || {
let candidates = daemon_recovery::read_snapshot(&path);
// Even malformed or temporarily unreadable snapshots belong to this generation only.
match std::fs::remove_file(&path) {
Ok(()) => candidates,
Err(err) if err.kind() == io::ErrorKind::NotFound => candidates,
Err(err) => Err(err),
}
})
.await
.map_err(io::Error::other)??;
Ok(tokio::spawn(async move {
processor.restore_daemon_threads(candidates).await;
}))
}
|