From 3a74bd9c353b795b3dfe7b705d19a23354ad9087 Mon Sep 17 00:00:00 2001 From: iris <2+iris@noreply.localhost> Date: Sat, 29 Aug 2026 05:37:42 -0400 Subject: [PATCH] Make the process record survive a crash mid-write Reviewing the reattach code found the fault it exists to prevent, sitting in its own save point. `process::write` used `fs::write`, which truncates before it fills. A crash inside that window leaves no readable record -- and a missing record reads as "nothing is running", which is the single answer that makes the next launch start a *second* CLI against a conversation that already has one. The window is not rare: the record is rewritten on every read that makes progress, so many times a second while a turn is producing output. Written to a neighbouring file and renamed over the real name now. The rename is atomic, so a reader sees the whole old record or the whole new one. That also makes the fixed-size padding pointless -- a rename replaces the file rather than overwriting part of it -- so it goes. Two more from the same pass: - A failed read of the stdout log was logged and nothing else. The session then went deaf with nothing on screen: no more output, no error, a status that stayed wherever it was. It now says so, closes the queue rather than stranding messages in it, and reports `Unknown` -- not `Exited`, because the process may well still be running; what failed is this server's ability to hear it. - Sizing the stderr log by reading it. `read_from` with a large offset answers "how long is it" by allocating the whole file first, which on a chatty process is a large pointless read on every reattach. `size_of` asks the filesystem. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01VETa8afmpWaYezLCqJhDB8 --- server/src/session/claude.rs | 37 ++++++++++---- server/src/session/process.rs | 92 +++++++++++++++++++++++++++-------- 2 files changed, 101 insertions(+), 28 deletions(-) diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 80a2821..53b8199 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -137,7 +137,7 @@ impl Queue { /// Reported rather than dropped. These are messages somebody typed /// that never reached the session and never reached the transcript, so /// this is the only place they can be mentioned at all. - fn close(&mut self, sink: &EventSink) { + fn close(&mut self, sink: &EventSink, why: &str) { self.closed = true; self.running = false; let lost: Vec = self.waiting.drain(..).map(|(text, _)| text).collect(); @@ -146,7 +146,7 @@ impl Queue { } let _ = sink.send(Event::Error { message: format!( - "the session ended before it read {}: {}", + "{why} before it read {}: {}", if lost.len() == 1 { "this message".to_string() } else { @@ -526,17 +526,35 @@ async fn follow( // run of this server was watching when it was written, so a reattach // starts at the end of it rather than repeating it. The tail is still // read from the file if the process dies, which is when it matters. - let mut stderr_at = match process::read_from(&stderr_path, u64::MAX) { - Ok((_, at)) => at, - Err(_) => 0, - }; + let mut stderr_at = process::size_of(&stderr_path); let mut said_unknown = false; while reading.load(Ordering::SeqCst) { let (bytes, _) = match process::read_from(&stdout_path, offset) { Ok(found) => found, Err(err) => { + // Reported, not only logged. This is the end of the + // session's output as far as anyone watching is + // concerned, and a phone told nothing shows a session + // that is merely quiet -- indistinguishable from one + // thinking. The status is `Unknown` rather than `Exited` + // because the process may well still be running; what + // has failed is this server's ability to hear it. tracing::error!("couldn't read {}: {err:#}", stdout_path.display()); + let _ = sink.send(Event::Error { + message: format!( + "lost track of {label}: its output can't be read ({err:#}). The process \ + may still be running; restarting the backend will try to pick it up \ + again." + ), + }); + queue + .lock() + .unwrap() + .close(&sink, "this server lost track of the session"); + let _ = sink.send(Event::Status { + state: SessionStatus::Unknown, + }); return; } }; @@ -596,7 +614,7 @@ async fn follow( // for ever and the exit would never be reported. process::Liveness::Dead if complete > 0 => {} process::Liveness::Dead => { - queue.lock().unwrap().close(&sink); + queue.lock().unwrap().close(&sink, "the session ended"); let detail = stderr_tail(&stderr_path); if !detail.is_empty() { let _ = sink.send(Event::Error { @@ -901,7 +919,7 @@ mod tests { }; queue.waiting.push_back(("first".into(), "{}".into())); queue.waiting.push_back(("second".into(), "{}".into())); - queue.close(&sink); + queue.close(&sink, "the session ended"); // Named rather than counted, because these never reached the // transcript: this message is the only record they existed. @@ -909,6 +927,7 @@ mod tests { panic!("closing a queue holding messages must report them"); }; assert!(message.contains("2 queued messages"), "{message}"); + assert!(message.starts_with("the session ended"), "{message}"); assert!( message.contains("first") && message.contains("second"), "{message}" @@ -924,7 +943,7 @@ mod tests { fn closing_an_empty_queue_says_nothing() { let (sink, mut received) = mpsc::unbounded_channel(); let mut queue = Queue::default(); - queue.close(&sink); + queue.close(&sink, "the session ended"); // A session that exits with nothing held has lost nothing, and an // error saying so would be noise on every ordinary exit. assert!(received.try_recv().is_err()); diff --git a/server/src/session/process.rs b/server/src/session/process.rs index 8c9e705..d2f2a26 100644 --- a/server/src/session/process.rs +++ b/server/src/session/process.rs @@ -28,6 +28,7 @@ //! live process" -- so the failure is the old behaviour (start one with //! `--resume`) rather than a wrong adoption. +use std::os::unix::fs::OpenOptionsExt; use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; @@ -35,12 +36,6 @@ use serde::{Deserialize, Serialize}; const RECORD_FILE: &str = "process.json"; -/// Fixed size for the record, so advancing the offset overwrites the file -/// rather than rewriting it -- there is no moment when it is shorter than -/// what was there before, and so no moment when a stale tail is readable -/// as part of the new value. -const RECORD_BYTES: usize = 256; - /// A process this server started and expects to outlive it. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct Record { @@ -136,36 +131,68 @@ pub fn live(session_dir: &Path) -> Option { } } -/// Writes `record` where [`live`] will find it. +/// Writes `record` where [`live`] will find it, atomically. +/// +/// Written to a neighbouring file and renamed over the real name. The +/// rename is what makes this safe: a reader sees either the whole old +/// record or the whole new one, never a partial. +/// +/// Writing in place would not be, and the consequence is severe rather +/// than untidy. `fs::write` truncates before it fills, so a crash inside +/// that window leaves no readable record -- and a missing record reads as +/// "nothing is running", which is the single answer that makes the next +/// launch start a *second* process against a conversation that already has +/// one. That is the fault this whole module exists to prevent, and writing +/// the record carelessly would reintroduce it at its own save point. The +/// window is not rare either: this runs on every read that makes progress, +/// so many times a second while a turn is producing output. /// /// Errors are logged rather than returned: this runs on the reading path, -/// and a session that cannot save its position is still a session worth -/// having -- it just cannot be reattached to, which is what the log says. +/// and a session that cannot save its position is still worth having -- it +/// just cannot be reattached to, which is what the log says. pub fn write(session_dir: &Path, record: &Record) { let path = path(session_dir); - let mut text = match serde_json::to_string(record) { + let text = match serde_json::to_string(record) { Ok(text) => text, Err(err) => { tracing::error!("couldn't serialize the process record: {err}"); return; } }; - if text.len() > RECORD_BYTES { - tracing::error!("process record is longer than its fixed size; not writing it"); - return; - } - // Padded to the fixed size so a shorter value never leaves a readable - // tail of the longer one it replaced. - text.push('\n'); - let padded = format!("{text: u64 { + std::fs::metadata(path).map(|meta| meta.len()).unwrap_or(0) +} + /// Forgets the recorded process -- for one confirmed dead, or a session /// being deleted. The path out for [`write`]. pub fn clear(session_dir: &Path) { @@ -313,6 +340,33 @@ mod tests { assert_eq!(live(dir.path()), None); } + #[test] + fn writing_leaves_no_temporary_behind_and_stays_readable() { + let dir = tempfile::tempdir().expect("tempdir"); + let mut record = + Record::of(std::process::id(), Detail::Stdio { stdout_read: 0 }).expect("start time"); + // Rewritten the way the reader rewrites it: constantly, as the + // position advances. Each one must land whole. + for read in [1u64, 4096, 2, 999_999] { + record.detail = Detail::Stdio { stdout_read: read }; + write(dir.path(), &record); + assert_eq!( + live(dir.path()), + Some(record.clone()), + "after offset {read}" + ); + } + // The rename is what makes it atomic; a leftover neighbour would + // mean it had not happened. + let stray: Vec<_> = std::fs::read_dir(dir.path()) + .expect("read dir") + .filter_map(Result::ok) + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|name| name != RECORD_FILE) + .collect(); + assert!(stray.is_empty(), "left behind {stray:?}"); + } + #[test] fn a_dead_or_unreadable_record_is_not_live() { let dir = tempfile::tempdir().expect("tempdir");