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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VETa8afmpWaYezLCqJhDB8
This commit is contained in:
1 parent
aff2cb90e2
commit
3a74bd9c35
2 files changed
+101
-28
No files matched your search
@@ -137,7 +137,7 @@ impl Queue {
|
|||||||
/// Reported rather than dropped. These are messages somebody typed
|
/// Reported rather than dropped. These are messages somebody typed
|
||||||
/// that never reached the session and never reached the transcript, so
|
/// that never reached the session and never reached the transcript, so
|
||||||
/// this is the only place they can be mentioned at all.
|
/// 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.closed = true;
|
||||||
self.running = false;
|
self.running = false;
|
||||||
let lost: Vec<String> = self.waiting.drain(..).map(|(text, _)| text).collect();
|
let lost: Vec<String> = self.waiting.drain(..).map(|(text, _)| text).collect();
|
||||||
@@ -146,7 +146,7 @@ impl Queue {
|
|||||||
}
|
}
|
||||||
let _ = sink.send(Event::Error {
|
let _ = sink.send(Event::Error {
|
||||||
message: format!(
|
message: format!(
|
||||||
"the session ended before it read {}: {}",
|
"{why} before it read {}: {}",
|
||||||
if lost.len() == 1 {
|
if lost.len() == 1 {
|
||||||
"this message".to_string()
|
"this message".to_string()
|
||||||
} else {
|
} else {
|
||||||
@@ -526,17 +526,35 @@ async fn follow(
|
|||||||
// run of this server was watching when it was written, so a reattach
|
// 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
|
// 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.
|
// 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) {
|
let mut stderr_at = process::size_of(&stderr_path);
|
||||||
Ok((_, at)) => at,
|
|
||||||
Err(_) => 0,
|
|
||||||
};
|
|
||||||
let mut said_unknown = false;
|
let mut said_unknown = false;
|
||||||
|
|
||||||
while reading.load(Ordering::SeqCst) {
|
while reading.load(Ordering::SeqCst) {
|
||||||
let (bytes, _) = match process::read_from(&stdout_path, offset) {
|
let (bytes, _) = match process::read_from(&stdout_path, offset) {
|
||||||
Ok(found) => found,
|
Ok(found) => found,
|
||||||
Err(err) => {
|
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());
|
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;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -596,7 +614,7 @@ async fn follow(
|
|||||||
// for ever and the exit would never be reported.
|
// for ever and the exit would never be reported.
|
||||||
process::Liveness::Dead if complete > 0 => {}
|
process::Liveness::Dead if complete > 0 => {}
|
||||||
process::Liveness::Dead => {
|
process::Liveness::Dead => {
|
||||||
queue.lock().unwrap().close(&sink);
|
queue.lock().unwrap().close(&sink, "the session ended");
|
||||||
let detail = stderr_tail(&stderr_path);
|
let detail = stderr_tail(&stderr_path);
|
||||||
if !detail.is_empty() {
|
if !detail.is_empty() {
|
||||||
let _ = sink.send(Event::Error {
|
let _ = sink.send(Event::Error {
|
||||||
@@ -901,7 +919,7 @@ mod tests {
|
|||||||
};
|
};
|
||||||
queue.waiting.push_back(("first".into(), "{}".into()));
|
queue.waiting.push_back(("first".into(), "{}".into()));
|
||||||
queue.waiting.push_back(("second".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
|
// Named rather than counted, because these never reached the
|
||||||
// transcript: this message is the only record they existed.
|
// transcript: this message is the only record they existed.
|
||||||
@@ -909,6 +927,7 @@ mod tests {
|
|||||||
panic!("closing a queue holding messages must report them");
|
panic!("closing a queue holding messages must report them");
|
||||||
};
|
};
|
||||||
assert!(message.contains("2 queued messages"), "{message}");
|
assert!(message.contains("2 queued messages"), "{message}");
|
||||||
|
assert!(message.starts_with("the session ended"), "{message}");
|
||||||
assert!(
|
assert!(
|
||||||
message.contains("first") && message.contains("second"),
|
message.contains("first") && message.contains("second"),
|
||||||
"{message}"
|
"{message}"
|
||||||
@@ -924,7 +943,7 @@ mod tests {
|
|||||||
fn closing_an_empty_queue_says_nothing() {
|
fn closing_an_empty_queue_says_nothing() {
|
||||||
let (sink, mut received) = mpsc::unbounded_channel();
|
let (sink, mut received) = mpsc::unbounded_channel();
|
||||||
let mut queue = Queue::default();
|
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
|
// A session that exits with nothing held has lost nothing, and an
|
||||||
// error saying so would be noise on every ordinary exit.
|
// error saying so would be noise on every ordinary exit.
|
||||||
assert!(received.try_recv().is_err());
|
assert!(received.try_recv().is_err());
|
||||||
|
|||||||
@@ -28,6 +28,7 @@
|
|||||||
//! live process" -- so the failure is the old behaviour (start one with
|
//! live process" -- so the failure is the old behaviour (start one with
|
||||||
//! `--resume`) rather than a wrong adoption.
|
//! `--resume`) rather than a wrong adoption.
|
||||||
|
|
||||||
|
use std::os::unix::fs::OpenOptionsExt;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
@@ -35,12 +36,6 @@ use serde::{Deserialize, Serialize};
|
|||||||
|
|
||||||
const RECORD_FILE: &str = "process.json";
|
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.
|
/// A process this server started and expects to outlive it.
|
||||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||||
pub struct Record {
|
pub struct Record {
|
||||||
@@ -136,36 +131,68 @@ pub fn live(session_dir: &Path) -> Option<Record> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 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,
|
/// 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
|
/// and a session that cannot save its position is still worth having -- it
|
||||||
/// having -- it just cannot be reattached to, which is what the log says.
|
/// just cannot be reattached to, which is what the log says.
|
||||||
pub fn write(session_dir: &Path, record: &Record) {
|
pub fn write(session_dir: &Path, record: &Record) {
|
||||||
let path = path(session_dir);
|
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,
|
Ok(text) => text,
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
tracing::error!("couldn't serialize the process record: {err}");
|
tracing::error!("couldn't serialize the process record: {err}");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
if text.len() > RECORD_BYTES {
|
// Beside the real file so the rename stays within one filesystem,
|
||||||
tracing::error!("process record is longer than its fixed size; not writing it");
|
// which is what makes it atomic.
|
||||||
return;
|
let temp = path.with_extension("json.new");
|
||||||
}
|
let written = std::fs::OpenOptions::new()
|
||||||
// Padded to the fixed size so a shorter value never leaves a readable
|
.create(true)
|
||||||
// tail of the longer one it replaced.
|
.write(true)
|
||||||
text.push('\n');
|
.truncate(true)
|
||||||
let padded = format!("{text:<RECORD_BYTES$}");
|
// Owner-only, like everything else in a session directory.
|
||||||
if let Err(err) = std::fs::write(&path, padded) {
|
.mode(0o600)
|
||||||
|
.open(&temp)
|
||||||
|
.and_then(|mut file| {
|
||||||
|
use std::io::Write;
|
||||||
|
file.write_all(text.as_bytes())?;
|
||||||
|
file.write_all(b"\n")
|
||||||
|
})
|
||||||
|
.and_then(|()| std::fs::rename(&temp, &path));
|
||||||
|
if let Err(err) = written {
|
||||||
tracing::error!(
|
tracing::error!(
|
||||||
"couldn't record the session process in {}: {err}",
|
"couldn't record the session process in {}: {err}",
|
||||||
path.display()
|
path.display()
|
||||||
);
|
);
|
||||||
|
let _ = std::fs::remove_file(&temp);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// How many bytes `path` holds, or 0 if it is not there.
|
||||||
|
///
|
||||||
|
/// Exists so a caller wanting only the length does not have to read the
|
||||||
|
/// file to find it -- [`read_from`] with a large offset answers the
|
||||||
|
/// question, but allocates the whole file on the way.
|
||||||
|
pub fn size_of(path: &Path) -> u64 {
|
||||||
|
std::fs::metadata(path).map(|meta| meta.len()).unwrap_or(0)
|
||||||
|
}
|
||||||
|
|
||||||
/// Forgets the recorded process -- for one confirmed dead, or a session
|
/// Forgets the recorded process -- for one confirmed dead, or a session
|
||||||
/// being deleted. The path out for [`write`].
|
/// being deleted. The path out for [`write`].
|
||||||
pub fn clear(session_dir: &Path) {
|
pub fn clear(session_dir: &Path) {
|
||||||
@@ -313,6 +340,33 @@ mod tests {
|
|||||||
assert_eq!(live(dir.path()), None);
|
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]
|
#[test]
|
||||||
fn a_dead_or_unreadable_record_is_not_live() {
|
fn a_dead_or_unreadable_record_is_not_live() {
|
||||||
let dir = tempfile::tempdir().expect("tempdir");
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
|||||||
Reference in new issue
Block a user