From 2917b3c344dbbb737017240ff3067017ed1b7143 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 28 Jul 2026 06:35:43 +0200 Subject: [PATCH 1/5] fix(storage, workspace, cli): Report disk-full and write errors clearly Previously a failed conversation write from `Drop` (the safety net for mutation scopes that unwind before an explicit `flush()`) was logged to stderr and swallowed: the CLI could exit zero after silently losing data. `ConversationLock` and `ConversationMut` now record a drop-time persist failure in shared state, and the next `flush()`, or a `take_persist_failure()` drain at teardown, surfaces it so the process exits non-zero and the printer reports it. `jp query`, `jp config set`, and the `jp conversation compact`/`edit`/`fork` subcommands now call `flush()` before printing success, so a write failure is reported as an error instead of a confirmation the user cannot trust. `jp_storage::Error` gains `WriteFailed` and `OutOfSpace` variants that carry the failing path alongside the OS error, replacing the bare, pathless `Io` variant on every write call site. A write that fails because the filesystem is full is classified separately: once seen, later write attempts on the same conversation lock are skipped rather than retried, since none can succeed until space is freed. The CLI renders `OutOfSpace` with a `suggestion` field telling the user to free disk space and re-run, and notes that anything after the last successful write was not saved. Updates RFD 069 (guard-scoped persistence for conversations) to describe the drop-time failure reporting path and the one remaining gap: a lock-owning scope (`into_mut`) that drops without a later `flush()` or `take_persist_failure()` call still only logs the failure. Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd.rs | 22 ++ crates/jp_cli/src/cmd/config/set.rs | 5 +- crates/jp_cli/src/cmd/conversation/compact.rs | 8 +- crates/jp_cli/src/cmd/conversation/edit.rs | 4 + crates/jp_cli/src/cmd/conversation/fork.rs | 16 +- crates/jp_cli/src/cmd/query.rs | 35 ++- crates/jp_cli/src/cmd_tests.rs | 92 ++++++ crates/jp_storage/src/error.rs | 70 ++++- crates/jp_storage/src/error_tests.rs | 74 +++++ crates/jp_storage/src/lib.rs | 2 +- crates/jp_storage/src/value.rs | 13 +- crates/jp_storage/src/value_tests.rs | 19 ++ crates/jp_workspace/src/conversation_lock.rs | 104 ++++++- .../src/conversation_lock_tests.rs | 267 +++++++++++++++++- crates/jp_workspace/src/error.rs | 14 + ...rd-scoped-persistence-for-conversations.md | 38 ++- 16 files changed, 743 insertions(+), 40 deletions(-) create mode 100644 crates/jp_cli/src/cmd_tests.rs create mode 100644 crates/jp_storage/src/error_tests.rs diff --git a/crates/jp_cli/src/cmd.rs b/crates/jp_cli/src/cmd.rs index abdef8487..f1e8ae084 100644 --- a/crates/jp_cli/src/cmd.rs +++ b/crates/jp_cli/src/cmd.rs @@ -738,6 +738,24 @@ impl From for Error { ("path", path.to_string().into()), ] .into(), + Error::OutOfSpace { path, source } => [ + ("message", "No space left on device".into()), + ("path", path.to_string().into()), + ("error", source.to_string().into()), + ( + "suggestion", + "Free up disk space and re-run. Anything after the last successful write was \ + not saved." + .into(), + ), + ] + .into(), + Error::WriteFailed { path, source } => [ + ("message", "Failed to write file".into()), + ("path", path.to_string().into()), + ("error", source.to_string().into()), + ] + .into(), Error::ConversationNotFound(id) => [ ("message", "Conversation not found.".into()), ("id", id.to_string().into()), @@ -794,3 +812,7 @@ impl From for Error { Self::from(metadata) } } + +#[cfg(test)] +#[path = "cmd_tests.rs"] +mod tests; diff --git a/crates/jp_cli/src/cmd/config/set.rs b/crates/jp_cli/src/cmd/config/set.rs index aca90843e..a3bacac57 100644 --- a/crates/jp_cli/src/cmd/config/set.rs +++ b/crates/jp_cli/src/cmd/config/set.rs @@ -53,9 +53,12 @@ impl Set { LockOutcome::ForkConversation(_) => unreachable!("fork not allowed"), }; - let conv = lock.into_mut(); + let mut conv = lock.into_mut(); let id = conv.id(); conv.update_events(|events| events.add_config_delta(config_delta.clone())); + // Write before reporting success, so a failed write is an error + // rather than a confirmation the user cannot trust. + conv.flush()?; ctx.printer .println(format!("Set configuration in conversation {id}")); } diff --git a/crates/jp_cli/src/cmd/conversation/compact.rs b/crates/jp_cli/src/cmd/conversation/compact.rs index 160c4dd62..dd90e090a 100644 --- a/crates/jp_cli/src/cmd/conversation/compact.rs +++ b/crates/jp_cli/src/cmd/conversation/compact.rs @@ -602,7 +602,7 @@ impl Compact { }; let cfg = ctx.config(); - let conv = lock.into_mut(); + let mut conv = lock.into_mut(); let events_snapshot = conv.events().clone(); if self.reset { @@ -617,6 +617,9 @@ impl Compact { } } else { let removed = conv.update_events(ConversationStream::remove_compactions); + // Write before reporting, so a failed write is an error rather + // than a confirmation the user cannot trust. + conv.flush()?; if removed > 0 { ctx.printer .println(format!("Removed {removed} compaction event(s).")); @@ -681,6 +684,9 @@ impl Compact { &conv.id().to_string(), )); apply_compactions(&conv, compactions); + // Write before reporting the timeline, so a failed write is an error + // rather than a confirmation the user cannot trust. + conv.flush()?; for line in timeline_lines(&segments, last_turn, false) { ctx.printer.println(line); } diff --git a/crates/jp_cli/src/cmd/conversation/edit.rs b/crates/jp_cli/src/cmd/conversation/edit.rs index 994d5e9fd..ba0b8d45c 100644 --- a/crates/jp_cli/src/cmd/conversation/edit.rs +++ b/crates/jp_cli/src/cmd/conversation/edit.rs @@ -301,6 +301,10 @@ impl Edit { } else if self.no_expires_at { conv.update_metadata(|m| m.expires_at = None); } + + // Write before reporting success, so a failed write is an error + // rather than a confirmation the user cannot trust. + conv.flush()?; } ctx.printer.println("Conversation(s) updated."); diff --git a/crates/jp_cli/src/cmd/conversation/fork.rs b/crates/jp_cli/src/cmd/conversation/fork.rs index e242009d8..98626b760 100644 --- a/crates/jp_cli/src/cmd/conversation/fork.rs +++ b/crates/jp_cli/src/cmd/conversation/fork.rs @@ -124,9 +124,13 @@ impl Fork { } })?; + // One mutable scope for both post-fork mutations, so they share a + // single write at the closing flush. + let mut conv = lock.as_mut(); + if self.compact.should_compact() { let cfg = ctx.config(); - let events_snapshot = lock.events().clone(); + let events_snapshot = conv.events().clone(); let rules = self .compact .effective_rules(&cfg.conversation.compaction.rules) @@ -143,17 +147,21 @@ impl Fork { ) .await?; for compaction in compactions { - lock.as_mut() - .update_events(|events| events.add_compaction(compaction)); + conv.update_events(|events| events.add_compaction(compaction)); } } if let Some(title) = &self.title { - lock.as_mut().update_metadata(|m| { + conv.update_metadata(|m| { m.title = Some(title.clone()); }); } + // Write before reporting success, so a failed write is an error + // rather than a confirmation the user cannot trust. + conv.flush()?; + drop(conv); + if self.activate && let Some(session) = &ctx.session && let Err(error) = diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 83e92bd8f..d1c223ada 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -499,21 +499,24 @@ impl Query { echo.render_user_request(&chat_request); } + // One mutable scope for the whole pre-turn setup. Every mutation below + // shares its single write at the closing `flush`, instead of each + // statement persisting the entire conversation on its own drop. + let mut setup = lock.as_mut(); + // Record the CLI-provided config delta (`--cfg`) now that the query is // known to be non-empty. Recording it before the empty-query check // would leave a config event behind for a query that was ultimately // ignored. if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? { - lock.as_mut() - .update_events(|events| events.add_config_delta(delta)); + setup.update_events(|events| events.add_config_delta(delta)); } if !editor_provided_config.is_empty() { // Resolve any model aliases before storing in the stream so // that per-event configs always contain concrete model IDs. editor_provided_config.resolve_model_aliases(&cfg.providers.llm.aliases); - lock.as_mut() - .update_events(|events| events.add_config_delta(editor_provided_config)); + setup.update_events(|events| events.add_config_delta(editor_provided_config)); } // Snapshot the stream for title generation and thread assembly. The @@ -522,7 +525,7 @@ impl Query { // stream is only trimmed at the turn-start commit point), the new // request not yet appended. let stream = { - let mut stream = lock.events().clone(); + let mut stream = setup.events().clone(); pending_trim.apply(&mut stream); stream }; @@ -546,8 +549,7 @@ impl Query { ) { NewTitle::FromHeading(title) => { debug!("Using leading markdown heading as conversation title"); - lock.as_mut() - .update_metadata(|m| m.title = Some(title.clone())); + setup.update_metadata(|m| m.title = Some(title.clone())); if ctx.term.is_tty { jp_term::osc::set_title(format!("{cid}: {title}")); } @@ -595,7 +597,13 @@ impl Query { // Sanitize any structural issues (orphaned tool calls, missing // user messages, etc.) before sending the stream to the provider. - lock.as_mut().update_events(ConversationStream::sanitize); + setup.update_events(ConversationStream::sanitize); + + // Commit the setup phase before the turn starts. Errors propagate here + // rather than being swallowed by a drop, and the turn loop's own + // checkpoints take over from this point. + setup.flush()?; + drop(setup); let invocation = InvocationContext { workspace_id: ctx.workspace.id().to_string(), @@ -645,6 +653,17 @@ impl Query { cleanup_query_message_file(ctx.fs_backend.as_deref(), &cid); } + // A mutation scope that dropped while dirty could not propagate its + // persist failure, so it recorded it on the lock instead. Surface it + // here so an unsaved conversation never exits zero. Only when the turn + // itself succeeded: a turn error is the more specific diagnostic, and + // the persist failure is usually a consequence of the same condition. + if turn_result.is_ok() + && let Some(error) = lock.take_persist_failure() + { + return Err(cmd::Error::from(Error::Workspace(error))); + } + turn_result } diff --git a/crates/jp_cli/src/cmd_tests.rs b/crates/jp_cli/src/cmd_tests.rs new file mode 100644 index 000000000..044bf4e5c --- /dev/null +++ b/crates/jp_cli/src/cmd_tests.rs @@ -0,0 +1,92 @@ +use std::io; + +use camino::Utf8Path; +use test_log::test; + +use super::*; + +/// `ENOSPC` on every Unix platform this crate targets. +/// +/// These tests assert the exact text the user sees, which includes the OS's own +/// message, so they build the error from the raw code the kernel delivers +/// rather than naming [`io::ErrorKind::StorageFull`] directly. +#[cfg(unix)] +const ENOSPC: i32 = 28; + +/// Render a CLI error the way the terminal output does: the message, then each +/// metadata entry as `key=value`. +fn rendered(error: &Error) -> String { + let mut lines = vec![error.message.clone().unwrap_or_default()]; + for (key, value) in &error.metadata { + lines.push(format!("{key}={}", value.as_str().unwrap_or_default())); + } + lines.join("\n") +} + +#[cfg(unix)] +#[test] +fn out_of_space_renders_path_cause_and_action() { + let error = Error::from(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + io::Error::from_raw_os_error(ENOSPC), + )); + + assert_eq!( + rendered(&error), + "No space left on device\npath=/data/conv/events.json\nerror=No space left on device (os \ + error 28)\nsuggestion=Free up disk space and re-run. Anything after the last successful \ + write was not saved." + ); + assert_eq!(error.code.get(), 1); +} + +#[test] +fn write_failure_renders_the_path_and_the_os_error() { + let error = Error::from(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + io::Error::from(io::ErrorKind::PermissionDenied), + )); + + assert_eq!( + rendered(&error), + "Failed to write file\npath=/data/conv/events.json\nerror=permission denied" + ); +} + +#[cfg(unix)] +#[test] +fn a_workspace_persist_failure_keeps_the_full_cause_chain() { + // The path a failed drop-time persist takes to the terminal: storage error + // wrapped by the workspace, wrapped by the CLI. Each layer has to keep the + // one below it renderable, or the user is told "IO error" and nothing else. + let error = Error::from(crate::error::Error::Workspace( + jp_workspace::Error::Storage(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + io::Error::from_raw_os_error(ENOSPC), + )), + )); + + assert_eq!( + rendered(&error), + "No space left on device\npath=/data/conv/events.json\nerror=No space left on device (os \ + error 28)\nsuggestion=Free up disk space and re-run. Anything after the last successful \ + write was not saved." + ); +} + +#[cfg(unix)] +#[test] +fn a_bare_io_error_renders_without_a_path() { + // `Error::Io` has no path to name, so it renders through the generic + // cause-chain walk: an "IO error" headline plus the OS message. Pinned to + // show what a caller gives up by not routing through + // `jp_storage::Error::write_failed`. + let error = Error::from(jp_storage::Error::from(io::Error::from_raw_os_error( + ENOSPC, + ))); + + assert_eq!( + rendered(&error), + "IO error\n=No space left on device (os error 28)" + ); +} diff --git a/crates/jp_storage/src/error.rs b/crates/jp_storage/src/error.rs index 007ec986c..d244f0f14 100644 --- a/crates/jp_storage/src/error.rs +++ b/crates/jp_storage/src/error.rs @@ -1,3 +1,5 @@ +use std::io; + use camino::Utf8PathBuf; use jp_conversation::ConversationId; @@ -17,9 +19,32 @@ pub enum Error { #[error("configuration error")] Config(#[from] jp_config::error::Error), - #[error("IO error")] + #[error(transparent)] Io(#[from] std::io::Error), + /// A write to `path` failed. + /// + /// `std::io::Error` carries no path, so a bare I/O failure cannot tell the + /// user which file it was. + #[error("failed to write {path}")] + WriteFailed { + path: Utf8PathBuf, + #[source] + source: io::Error, + }, + + /// The filesystem holding `path` has no room left. + /// + /// Separate from [`Error::WriteFailed`] because it is the one write failure + /// with an obvious user action, and because no retry can succeed until + /// space is freed. + #[error("no space left on device while writing {path}")] + OutOfSpace { + path: Utf8PathBuf, + #[source] + source: io::Error, + }, + #[error("invalid JSON data")] Json(#[from] serde_json::Error), @@ -30,6 +55,45 @@ pub enum Error { ConversationNotFound(ConversationId), } +impl Error { + /// Classify a failed write to `path`. + /// + /// Yields [`Error::OutOfSpace`] when the operating system reported a full + /// filesystem, and [`Error::WriteFailed`] otherwise. + #[must_use] + pub fn write_failed(path: impl Into, source: io::Error) -> Self { + let path = path.into(); + if is_storage_full(&source) { + Self::OutOfSpace { path, source } + } else { + Self::WriteFailed { path, source } + } + } + + /// Whether the error means the filesystem is full. + /// + /// A caller that sees this should stop attempting writes rather than move + /// on to the next one: none of them can succeed until space is freed. + #[must_use] + pub fn is_out_of_space(&self) -> bool { + match self { + Self::OutOfSpace { .. } => true, + Self::Io(error) => is_storage_full(error), + _ => false, + } + } +} + +/// Whether an OS error reports a full filesystem. +/// +/// Recognises both a raw OS error (`ENOSPC`, `ERROR_DISK_FULL`) and an error +/// constructed directly from the kind, since `std` maps the platform codes onto +/// [`io::ErrorKind::StorageFull`]. +#[must_use] +pub fn is_storage_full(error: &io::Error) -> bool { + error.kind() == io::ErrorKind::StorageFull +} + #[cfg(test)] impl PartialEq for Error { fn eq(&self, other: &Self) -> bool { @@ -41,3 +105,7 @@ impl PartialEq for Error { format!("{self:?}") == format!("{other:?}") } } + +#[cfg(test)] +#[path = "error_tests.rs"] +mod tests; diff --git a/crates/jp_storage/src/error_tests.rs b/crates/jp_storage/src/error_tests.rs new file mode 100644 index 000000000..2a507c568 --- /dev/null +++ b/crates/jp_storage/src/error_tests.rs @@ -0,0 +1,74 @@ +use camino::Utf8Path; + +use super::*; + +/// `ENOSPC` on every Unix platform this crate targets. +/// +/// Building the error from the raw code keeps the classification tests honest: +/// they run against what the kernel actually delivers, not against an error +/// synthesized from the very [`io::ErrorKind`] the classifier reads back. +#[cfg(unix)] +const ENOSPC: i32 = 28; + +/// An out-of-space error as the operating system reports it. +#[cfg(unix)] +fn os_disk_full() -> io::Error { + io::Error::from_raw_os_error(ENOSPC) +} + +#[cfg(not(unix))] +fn os_disk_full() -> io::Error { + io::Error::from(io::ErrorKind::StorageFull) +} + +#[test] +fn write_failed_classifies_full_disk_as_out_of_space() { + let error = Error::write_failed(Utf8Path::new("/data/conv/events.json"), os_disk_full()); + + assert!(matches!(error, Error::OutOfSpace { .. })); + assert!(error.is_out_of_space()); + assert_eq!( + error.to_string(), + "no space left on device while writing /data/conv/events.json" + ); +} + +#[test] +fn write_failed_classifies_other_errors_as_write_failed() { + let error = Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + io::Error::from(io::ErrorKind::PermissionDenied), + ); + + assert!(matches!(error, Error::WriteFailed { .. })); + assert!(!error.is_out_of_space()); + assert_eq!(error.to_string(), "failed to write /data/conv/events.json"); +} + +#[test] +fn write_failed_keeps_the_os_error_as_the_source() { + let error = Error::write_failed(Utf8Path::new("/data/conv/events.json"), os_disk_full()); + + let source = std::error::Error::source(&error) + .and_then(|source| source.downcast_ref::()) + .expect("the os error is retained as the source, unwrapped"); + assert_eq!(source.kind(), io::ErrorKind::StorageFull); +} + +#[cfg(unix)] +#[test] +fn io_variant_renders_the_os_error_verbatim() { + // `Error::Io` is transparent: a bare I/O failure must not be flattened into + // an opaque "IO error", or the cause is lost wherever only `Display` is + // rendered. + let error = Error::from(os_disk_full()); + + assert_eq!(error.to_string(), "No space left on device (os error 28)"); + assert!(error.is_out_of_space()); +} + +#[test] +fn unrelated_errors_are_not_out_of_space() { + assert!(!Error::NotDir(Utf8Path::new("/data").to_owned()).is_out_of_space()); + assert!(!Error::from(io::Error::from(io::ErrorKind::NotFound)).is_out_of_space()); +} diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs index 9c93ba78b..13c27188d 100644 --- a/crates/jp_storage/src/lib.rs +++ b/crates/jp_storage/src/lib.rs @@ -237,7 +237,7 @@ impl Storage { // Bring any existing copy to the current directory name (e.g. after a // title change) and drop other stale copies for this id. reconcile_conversation_dir(id, conversations_dir, &conv_dir)?; - fs::create_dir_all(&conv_dir)?; + fs::create_dir_all(&conv_dir).map_err(|error| Error::write_failed(&conv_dir, error))?; let (base_config, events_json) = events .to_parts() diff --git a/crates/jp_storage/src/value.rs b/crates/jp_storage/src/value.rs index 7519fb33f..a6608afc0 100644 --- a/crates/jp_storage/src/value.rs +++ b/crates/jp_storage/src/value.rs @@ -4,7 +4,7 @@ use camino::{Utf8Path, Utf8PathBuf}; use serde::{Serialize, de::DeserializeOwned}; use serde_json::Value; -use crate::error::Result; +use crate::error::{Error, Result}; /// The temp-file suffix used by [`write_json`] for atomic writes. pub const TMP_SUFFIX: &str = ".tmp"; @@ -57,11 +57,16 @@ pub fn read_json(path: &Utf8Path) -> Result { /// target. /// If anything fails before the rename, the original file is left untouched and /// the temp file is cleaned up on a best-effort basis. +/// The temp file lives on the same filesystem as `path`, so the write needs +/// room for a full second copy of the content. /// /// The write is skipped when `path` already holds the exact same bytes, so an /// unchanged file keeps its modification time. /// Use [`write_json_force`] when a refreshed mtime is required regardless of /// content. +/// +/// A failure names the path it was writing, and reports a full filesystem as +/// [`Error::OutOfSpace`]. pub fn write_json(path: &Utf8Path, value: &T) -> Result<()> { write_json_impl(path, value, false) } @@ -84,7 +89,7 @@ fn write_json_impl(path: &Utf8Path, value: &T, force: bool) -> Res /// is skipped so the file's modification time is preserved. fn write_bytes(path: &Utf8Path, bytes: &[u8], force: bool) -> Result<()> { if let Some(parent) = path.parent() { - fs::create_dir_all(parent)?; + fs::create_dir_all(parent).map_err(|error| Error::write_failed(parent, error))?; } if !force @@ -97,12 +102,12 @@ fn write_bytes(path: &Utf8Path, bytes: &[u8], force: bool) -> Result<()> { let tmp_path = tmp_path_for(path); if let Err(error) = fs::write(&tmp_path, bytes) { let _err = fs::remove_file(&tmp_path); - return Err(error.into()); + return Err(Error::write_failed(&tmp_path, error)); } if let Err(error) = fs::rename(&tmp_path, path) { let _err = fs::remove_file(&tmp_path); - return Err(error.into()); + return Err(Error::write_failed(path, error)); } Ok(()) diff --git a/crates/jp_storage/src/value_tests.rs b/crates/jp_storage/src/value_tests.rs index a40a0b38e..2732ea4b6 100644 --- a/crates/jp_storage/src/value_tests.rs +++ b/crates/jp_storage/src/value_tests.rs @@ -54,6 +54,25 @@ fn write_json_preserves_original_on_write_failure() { assert_eq!(content, json!({"original": true})); } +#[test] +fn write_json_failure_names_the_path_it_was_writing() { + let tmp = tempdir().unwrap(); + let path = tmp.path().join("out.json"); + + // A directory at the .tmp path makes the temp write fail on all platforms. + let blocker = Utf8PathBuf::from(format!("{path}{TMP_SUFFIX}")); + fs::create_dir(&blocker).unwrap(); + + let error = write_json(&path, &json!({"v": 1})).expect_err("temp write should fail"); + + // The failing path is the temp sibling, not the target: naming the target + // would point the user at a file that was never touched. + match error { + Error::WriteFailed { path: failed, .. } => assert_eq!(failed, blocker), + other => panic!("expected WriteFailed, got {other:?}"), + } +} + #[test] fn write_json_overwrites_existing_file() { let tmp = tempdir().unwrap(); diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs index f444b39c2..0efe8f1e3 100644 --- a/crates/jp_workspace/src/conversation_lock.rs +++ b/crates/jp_workspace/src/conversation_lock.rs @@ -61,10 +61,24 @@ //! - **`Drop`** — safety net. //! If the `ConversationMut` drops while dirty (e.g., due to `?` unwinding), //! `Drop` persists the data. -//! Errors are logged but cannot be propagated from `Drop`. //! //! Long-running loops should call `flush()` at each checkpoint so disk errors //! halt immediately rather than letting the loop continue with unsaved data. +//! +//! # Reporting Drop-Time Failures +//! +//! `Drop` cannot propagate, so a failed drop-time persist is recorded on state +//! shared with the originating lock instead of being written to the terminal. +//! The next `flush()` returns it, and +//! [`take_persist_failure`][ConversationLock::take_persist_failure] drains it +//! for a caller that wants to report at teardown. +//! Either way, one failing disk yields one diagnostic rather than one per +//! mutation scope, and the reporting happens where the output channel and the +//! exit code live. +//! +//! A failure that shows the filesystem is full also marks the shared state, and +//! subsequent write attempts are skipped: nothing can succeed until space is +//! freed. use std::sync::{ Arc, @@ -73,10 +87,29 @@ use std::sync::{ use jp_conversation::{Conversation, ConversationId, ConversationStream}; use jp_storage::backend::{ConversationLockGuard, PersistBackend, Projection}; -use parking_lot::{RwLock, RwLockReadGuard}; -use tracing::info; +use parking_lot::{Mutex, RwLock, RwLockReadGuard}; +use tracing::{info, warn}; -use crate::handle::ConversationHandle; +use crate::{error::Error, handle::ConversationHandle}; + +/// Shared record of drop-time persistence outcomes for one conversation lock. +/// +/// See the module docs for how a recorded failure reaches the user. +#[derive(Debug, Default)] +struct PersistState { + /// The first drop-time failure no caller has been told about yet. + /// + /// The first is kept rather than the last: later failures are consequences + /// of the same condition, while the first names the write that actually + /// broke. + failure: Option, + + /// Set once a failure showed the filesystem is full. + /// + /// Sticky, and never cleared: further write attempts are skipped rather + /// than retried against a disk with no room. + out_of_space: bool, +} /// Result of attempting to acquire a conversation lock. #[derive(Debug)] @@ -105,6 +138,7 @@ pub struct ConversationLock { writer: Arc, lock_guard: Arc>, projection: Projection, + persist: Arc>, } impl ConversationLock { @@ -128,9 +162,21 @@ impl ConversationLock { writer, lock_guard: Arc::new(lock_guard), projection, + persist: Arc::default(), } } + /// Take the drop-time persist failure recorded since the last call. + /// + /// A caller reports it once, through whatever output channel and exit code + /// it owns. + /// Failures already surfaced through [`ConversationMut::flush`] are not + /// recorded here and are never returned twice. + #[must_use] + pub fn take_persist_failure(&self) -> Option { + self.persist.lock().failure.take() + } + /// The write projection this lock persists with. /// /// Resolved from storage presence at acquisition (or the creation flags for @@ -184,6 +230,7 @@ impl ConversationLock { dirty: AtomicBool::new(false), writer: Arc::clone(&self.writer), projection: self.projection, + persist: Arc::clone(&self.persist), _lock_guard: Arc::clone(&self.lock_guard), } } @@ -202,6 +249,7 @@ impl ConversationLock { dirty: AtomicBool::new(false), writer: self.writer, projection: self.projection, + persist: self.persist, _lock_guard: self.lock_guard, } } @@ -235,6 +283,10 @@ pub struct ConversationMut { writer: Arc, projection: Projection, + // Shared with the originating lock and every other scope derived from it, + // so a failure recorded here survives this scope's drop. + persist: Arc>, + // Holds the lock guard alive. Released when the last Arc drops. _lock_guard: Arc>, } @@ -331,8 +383,10 @@ impl ConversationMut { /// Long-running loops **must** call this at each checkpoint (e.g., after /// each turn in the LLM loop) so that I/O errors propagate immediately via /// `?`. - /// The `Drop` implementation is a safety net for `?` unwinding — it - /// persists partial state but swallows errors. + /// + /// Returns a persist failure recorded by an earlier drop-time write before + /// attempting its own, so a failure that could not propagate from `Drop` + /// still reaches a caller that can act on it. /// /// Takes `&mut self` to prevent calling while a write guard from /// `update_events()` or `update_metadata()` is held (which would deadlock). @@ -341,19 +395,38 @@ impl ConversationMut { /// /// After a successful flush, the dirty flag is cleared. pub fn flush(&mut self) -> crate::error::Result<()> { + if let Some(error) = self.persist.lock().failure.take() { + return Err(error); + } + if !self.dirty.load(Ordering::Relaxed) { return Ok(()); } let meta = self.metadata.read(); let evts = self.events.read(); - self.writer.write(&self.id, &meta, &evts, self.projection)?; + if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) { + // Recorded as futile but not as a pending failure: this error is + // returned, so the caller owns reporting it. + let error = Error::from(error); + self.persist.lock().out_of_space |= error.is_out_of_space(); + return Err(error); + } self.dirty.store(false, Ordering::Relaxed); info!(id = %self.id, "Flushed conversation to disk."); Ok(()) } + /// Take the drop-time persist failure recorded since the last call. + /// + /// Counterpart of [`ConversationLock::take_persist_failure`] for a scope + /// that owns the lock, where no `ConversationLock` remains to drain. + #[must_use] + pub fn take_persist_failure(&self) -> Option { + self.persist.lock().failure.take() + } + /// The write projection this scope persists with. #[must_use] pub fn projection(&self) -> Projection { @@ -399,12 +472,23 @@ impl Drop for ConversationMut { return; } + if self.persist.lock().out_of_space { + warn!(id = %self.id, "Skipping persist: no space left on device."); + return; + } + let meta = self.metadata.read(); let evts = self.events.read(); - #[expect(clippy::print_stderr)] - if let Err(e) = self.writer.write(&self.id, &meta, &evts, self.projection) { - eprintln!("Failed to persist conversation {}: {e}", self.id); + if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) { + let error = Error::from(error); + warn!(id = %self.id, %error, "Failed to persist conversation."); + + let mut state = self.persist.lock(); + state.out_of_space |= error.is_out_of_space(); + if state.failure.is_none() { + state.failure = Some(error); + } } } } diff --git a/crates/jp_workspace/src/conversation_lock_tests.rs b/crates/jp_workspace/src/conversation_lock_tests.rs index cda218ee6..fa40f6e96 100644 --- a/crates/jp_workspace/src/conversation_lock_tests.rs +++ b/crates/jp_workspace/src/conversation_lock_tests.rs @@ -1,5 +1,12 @@ -use std::sync::{Arc, Mutex}; - +use std::{ + io, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering as AtomicOrdering}, + }, +}; + +use camino::Utf8Path; use jp_conversation::{Conversation, ConversationId, ConversationStream}; use jp_storage::backend::{NoopLockGuard, NullPersistBackend, PersistBackend, Projection}; use parking_lot::RwLock; @@ -53,6 +60,87 @@ impl PersistBackend for MockPersistBackend { } } +/// Persistence backend that fails every write with a fixed error, counting the +/// attempts. +/// +/// The count is what makes the "one failing disk, one report" claim testable: a +/// backend that only records the error cannot distinguish "the write was +/// skipped" from "the write was retried and failed again". +#[derive(Debug)] +struct FailingPersistBackend { + attempts: AtomicUsize, + out_of_space: bool, +} + +impl FailingPersistBackend { + fn out_of_space() -> Self { + Self { + attempts: AtomicUsize::new(0), + out_of_space: true, + } + } + + fn permission_denied() -> Self { + Self { + attempts: AtomicUsize::new(0), + out_of_space: false, + } + } + + fn attempts(&self) -> usize { + self.attempts.load(AtomicOrdering::Relaxed) + } +} + +impl PersistBackend for FailingPersistBackend { + fn write( + &self, + _id: &ConversationId, + _metadata: &Conversation, + _events: &ConversationStream, + _projection: Projection, + ) -> Result<(), jp_storage::Error> { + self.attempts.fetch_add(1, AtomicOrdering::Relaxed); + + let source = if self.out_of_space { + io::Error::from(io::ErrorKind::StorageFull) + } else { + io::Error::from(io::ErrorKind::PermissionDenied) + }; + Err(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + source, + )) + } + + fn remove(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> { + Ok(()) + } + + fn archive(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> { + Ok(()) + } + + fn unarchive(&self, _id: &ConversationId) -> Result<(), jp_storage::Error> { + Ok(()) + } +} + +fn test_lock_with_failing_backend( + backend: FailingPersistBackend, +) -> (ConversationLock, Arc) { + let backend = Arc::new(backend); + let lock = ConversationLock::new( + test_handle(), + Arc::new(RwLock::new(Conversation::default())), + Arc::new(RwLock::new(ConversationStream::new_test())), + Arc::clone(&backend) as _, + Box::new(NoopLockGuard), + Projection::Projected, + ); + (lock, backend) +} + fn test_id() -> ConversationId { ConversationId::try_from(chrono::DateTime::::UNIX_EPOCH).unwrap() } @@ -291,6 +379,181 @@ fn as_mut_mutations_visible_through_lock() { assert_eq!(lock.metadata().title.as_deref(), Some("visible")); } +#[test] +fn no_persist_failure_recorded_on_a_healthy_backend() { + let (lock, _mock) = test_lock_with_mock(); + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("fine".into())); + } + assert!(lock.take_persist_failure().is_none()); +} + +#[test] +fn drop_records_persist_failure_on_the_lock() { + let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("unsaved".into())); + } + + assert_eq!(backend.attempts(), 1); + let error = lock + .take_persist_failure() + .expect("the failed drop-time write is recorded on the lock"); + assert!(error.is_out_of_space()); + assert_eq!( + error.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); +} + +#[test] +fn recorded_persist_failure_is_returned_only_once() { + let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + { + let conv = lock.as_mut(); + conv.update_metadata(|_| {}); + } + + assert!(lock.take_persist_failure().is_some()); + assert!( + lock.take_persist_failure().is_none(), + "draining the failure must not report it a second time" + ); +} + +#[test] +fn out_of_space_skips_every_later_write_attempt() { + let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + // Three mutation scopes, each of which would persist on drop. + for _ in 0..3 { + let conv = lock.as_mut(); + conv.update_metadata(|_| {}); + } + + assert_eq!( + backend.attempts(), + 1, + "a full disk must be hit once, not once per mutation scope" + ); +} + +#[test] +fn out_of_space_records_only_the_first_failure() { + let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("first".into())); + } + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("second".into())); + } + + assert!(lock.take_persist_failure().is_some()); + assert!(lock.take_persist_failure().is_none()); +} + +#[test] +fn a_recoverable_failure_does_not_skip_later_writes() { + // Only a full filesystem makes further writes futile. A permission error may + // clear (the user chmods the directory mid-run), so each scope still tries. + let (lock, backend) = + test_lock_with_failing_backend(FailingPersistBackend::permission_denied()); + + for _ in 0..3 { + let conv = lock.as_mut(); + conv.update_metadata(|_| {}); + } + + assert_eq!(backend.attempts(), 3); + let error = lock.take_persist_failure().expect("failure is recorded"); + assert!(!error.is_out_of_space()); +} + +#[test] +fn flush_returns_a_failure_recorded_by_an_earlier_drop() { + let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + { + let conv = lock.as_mut(); + conv.update_metadata(|_| {}); + } + assert_eq!(backend.attempts(), 1); + + let mut conv = lock.as_mut(); + let error = conv + .flush() + .expect_err("flush surfaces the earlier drop-time failure"); + + assert!(error.is_out_of_space()); + assert_eq!( + backend.attempts(), + 1, + "flush returns the recorded failure without touching a full disk again" + ); + assert!( + lock.take_persist_failure().is_none(), + "a failure surfaced through flush is not reported a second time" + ); +} + +#[test] +fn flush_does_not_record_its_own_failure() { + // `flush` propagates, so its caller owns reporting. Recording it too would + // produce the double report this mechanism exists to prevent. + let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + let mut conv = lock.as_mut(); + conv.update_metadata(|_| {}); + assert!(conv.flush().is_err()); + drop(conv); + + assert!(lock.take_persist_failure().is_none()); +} + +#[test] +fn a_failed_flush_leaves_the_scope_dirty_for_the_drop_safety_net() { + let (lock, backend) = + test_lock_with_failing_backend(FailingPersistBackend::permission_denied()); + + let mut conv = lock.as_mut(); + conv.update_metadata(|_| {}); + assert!(conv.flush().is_err()); + assert!(conv.is_dirty()); + drop(conv); + + assert_eq!( + backend.attempts(), + 2, + "the drop safety net retries a recoverable flush failure" + ); +} + +#[test] +fn a_scope_drains_a_failure_recorded_before_it_existed() { + // The lock-owning scope (`into_mut`) is the only drain point once the lock + // has been consumed, so it must see failures recorded by earlier scopes. + let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + + { + let conv = lock.as_mut(); + conv.update_metadata(|_| {}); + } + + let conv = lock.into_mut(); + let error = conv + .take_persist_failure() + .expect("the owning scope sees the earlier failure"); + assert!(error.is_out_of_space()); + assert!(conv.take_persist_failure().is_none()); +} + #[test] fn multiple_as_mut_each_persist_independently() { let (lock, mock) = test_lock_with_mock(); diff --git a/crates/jp_workspace/src/error.rs b/crates/jp_workspace/src/error.rs index 767ed3c59..953ba586b 100644 --- a/crates/jp_workspace/src/error.rs +++ b/crates/jp_workspace/src/error.rs @@ -1,4 +1,5 @@ use camino::Utf8PathBuf; +use jp_storage::error::is_storage_full; pub(crate) type Result = std::result::Result; @@ -42,6 +43,19 @@ pub enum Error { } impl Error { + /// Whether the error means the filesystem is full. + /// + /// A caller that sees this should stop attempting writes rather than move + /// on to the next one: none of them can succeed until space is freed. + #[must_use] + pub fn is_out_of_space(&self) -> bool { + match self { + Self::Storage(error) => error.is_out_of_space(), + Self::Io(error) => is_storage_full(error), + _ => false, + } + } + pub fn not_found(target: &'static str, id: &impl ToString) -> Self { Self::NotFound(target, id.to_string()) } diff --git a/docs/rfd/069-guard-scoped-persistence-for-conversations.md b/docs/rfd/069-guard-scoped-persistence-for-conversations.md index 89e49d194..5116aafe8 100644 --- a/docs/rfd/069-guard-scoped-persistence-for-conversations.md +++ b/docs/rfd/069-guard-scoped-persistence-for-conversations.md @@ -166,6 +166,9 @@ impl ConversationMut { /// Long-running loops must call this at each checkpoint so I/O /// errors propagate via `?`. Drop is the safety net for unwinding. /// + /// Returns a failure recorded by an earlier drop-time write before + /// attempting its own. + /// /// Takes `&mut self` to prevent calling while a write guard from /// update_events() is held (which would deadlock). pub fn flush(&mut self) -> Result<()>; @@ -174,17 +177,32 @@ impl ConversationMut { impl Drop for ConversationMut { fn drop(&mut self) { if !self.dirty.load(Ordering::Relaxed) { return; } - if let Some(writer) = &self.writer { - let meta = self.metadata.read(); - let evts = self.events.read(); - if let Err(e) = writer.write(&self.id, &meta, &evts) { - eprintln!("Failed to persist conversation {}: {e}", self.id); - } + if self.persist.lock().out_of_space { return; } + + let meta = self.metadata.read(); + let evts = self.events.read(); + if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) { + warn!(id = %self.id, %error, "Failed to persist conversation."); + let mut state = self.persist.lock(); + state.out_of_space |= error.is_out_of_space(); + if state.failure.is_none() { state.failure = Some(error); } } } } ``` +`Drop` cannot propagate, so it records the failure on state shared with the +originating lock rather than writing to the terminal. +`ConversationLock::take_persist_failure` (and its `ConversationMut` +counterpart) drains it, so the shell reports once, through the printer, with a +non-zero exit code. +One failing disk therefore yields one diagnostic instead of one per mutation +scope. + +A failure that shows the filesystem is full marks the shared state, and +subsequent write attempts are skipped: nothing can succeed until space is +freed. + `AtomicBool` is used for the dirty flag instead of `Cell`. `Cell` is `!Sync`, which would make `ConversationMut` `!Sync` and cause async futures holding `&ConversationMut` across `.await` points to become @@ -352,8 +370,12 @@ This is slightly more verbose but structurally prevents `.await`-across-lock-guard bugs. `?` composes naturally since the callback's return type is forwarded. -**Errors in `Drop` are swallowed.** If persist fails during `ConversationMut`'s -drop, the error is logged to stderr but cannot be propagated. +**Errors in `Drop` cannot be propagated.** A persist failure during +`ConversationMut`'s drop is recorded on the shared persist state and surfaced by +the next `flush()` or by a `take_persist_failure()` drain at teardown, rather +than returned from `drop`. +A scope that owns the lock (`into_mut`) and drops without either is the one +remaining hole: its failure reaches the log file but no caller. Long-running loops must call `flush()?` at checkpoints so that I/O failures halt immediately. From ed7887801c49e74e4451e6293b9902831606d686 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 28 Jul 2026 08:41:30 +0200 Subject: [PATCH 2/5] review feedback Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd.rs | 8 ++ crates/jp_cli/src/cmd/query.rs | 76 ++++++++--- crates/jp_cli/src/cmd/query_tests.rs | 53 ++++++++ crates/jp_storage/src/error.rs | 16 +-- crates/jp_storage/src/error_tests.rs | 9 -- crates/jp_storage/src/lib.rs | 12 +- crates/jp_storage/src/lib_tests.rs | 23 ++++ crates/jp_workspace/src/conversation_lock.rs | 37 ++--- .../src/conversation_lock_tests.rs | 128 ++++++------------ crates/jp_workspace/src/error.rs | 14 -- docs/.vitepress/rfd-summaries.json | 2 +- ...rd-scoped-persistence-for-conversations.md | 27 ++-- 12 files changed, 223 insertions(+), 182 deletions(-) diff --git a/crates/jp_cli/src/cmd.rs b/crates/jp_cli/src/cmd.rs index f1e8ae084..2150d796c 100644 --- a/crates/jp_cli/src/cmd.rs +++ b/crates/jp_cli/src/cmd.rs @@ -221,6 +221,14 @@ impl Error { ..self } } + + /// Append a metadata entry, rendered under the error's message. + /// + /// Used to attach a secondary fact to an error that already has a more + /// specific message of its own. + pub(super) fn push_metadata(&mut self, key: &str, value: impl Into) { + self.metadata.push((key.to_owned(), value.into())); + } } impl fmt::Display for Error { diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index d1c223ada..035f9d05a 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -344,7 +344,6 @@ pub(crate) struct Query { } impl Query { - #[expect(clippy::too_many_lines)] pub(crate) async fn run( self, ctx: &mut Ctx, @@ -353,8 +352,6 @@ impl Query { ) -> Output { debug!("Running `query` command."); trace!(args = ?self, "Received arguments."); - let now = ctx.now(); - let cfg = ctx.config(); // Resolve the target conversation and acquire an exclusive lock. // @@ -365,6 +362,24 @@ impl Query { // 4. Lock contention: user picks "new" or "fork" from the prompt. let lock = self.acquire_lock(ctx, handle, start_new).await?; + let result = self.run_locked(ctx, &lock).await; + + // Every exit from the locked region lands here, which is what makes + // this the one reliable drain point: a mutation scope that dropped + // while dirty recorded its persist failure on the lock, and any `?` in + // the region above returns past every other candidate site. + fold_persist_failure(result, lock.take_persist_failure()) + } + + /// Run the query against an already-locked conversation. + /// + /// Errors propagate freely: the caller drains any persist failure the + /// unwinding left behind. + #[expect(clippy::too_many_lines)] + async fn run_locked(self, ctx: &mut Ctx, lock: &ConversationLock) -> Output { + let now = ctx.now(); + let cfg = ctx.config(); + // Create symlinks and seed approvals for any `--mount` flags before the // turn runs, so tools can reach the mounted paths. create_mount_effects(&self.mount, &ctx.workspace, ctx.fs_backend.as_deref(), now)?; @@ -372,13 +387,13 @@ impl Query { // The two flags are mutually exclusive (enforced by clap), and the // resolved conversation may be new, freshly forked (which clones the // source's metadata, including any title), or resumed. - apply_title_override(&lock, self.title.as_deref(), self.no_title); + apply_title_override(lock, self.title.as_deref(), self.no_title); // Record this conversation as the session's active conversation. if let Some(session) = &ctx.session && let Err(error) = ctx .workspace - .activate_session_conversation(&lock, session, now) + .activate_session_conversation(lock, session, now) { warn!(%error, "Failed to record activation."); } @@ -400,7 +415,7 @@ impl Query { // Compact the conversation before querying, if requested. if self.compact.should_compact() { - self.apply_pre_query_compaction(&lock, &cfg).await?; + self.apply_pre_query_compaction(lock, &cfg).await?; } let mut mcp_servers_handle = ctx.configure_active_mcp_servers().await?; @@ -508,7 +523,7 @@ impl Query { // known to be non-empty. Recording it before the empty-query check // would leave a config event behind for a query that was ultimately // ignored. - if let Some(delta) = get_config_delta_from_cli(&cfg, &lock)? { + if let Some(delta) = get_config_delta_from_cli(&cfg, lock)? { setup.update_events(|events| events.add_config_delta(delta)); } @@ -610,7 +625,7 @@ impl Query { conversation_id: lock.id().to_string(), }; - let turn_result = self + let mut turn_result = self .handle_turn( &cfg, &ctx.signals, @@ -618,7 +633,7 @@ impl Query { root, ctx.term.is_tty, &thread.attachments, - &lock, + lock, cfg.assistant.tool_choice.clone(), &tools, ctx.printer.clone(), @@ -630,6 +645,19 @@ impl Query { .await .map_err(|error| cmd::Error::from(error).with_persistence(true)); + // Fold a drop-time persist failure into the turn's result before + // anything downstream treats the turn as a success. Both gates below + // key off `is_ok`, and the scratch file one of them deletes is the + // user's only recovery copy of a request that was never saved. + // + // `run` drains again for the exits that never reach this point; the + // failure is only ever handed out once. + if turn_result.is_ok() + && let Some(error) = lock.take_persist_failure() + { + turn_result = Err(cmd::Error::from(Error::Workspace(error))); + } + // Extract structured data from the conversation after the turn. if self.schema.is_some() && turn_result.is_ok() { let data = lock.events().iter().rev().find_map(|e| { @@ -653,17 +681,6 @@ impl Query { cleanup_query_message_file(ctx.fs_backend.as_deref(), &cid); } - // A mutation scope that dropped while dirty could not propagate its - // persist failure, so it recorded it on the lock instead. Surface it - // here so an unsaved conversation never exits zero. Only when the turn - // itself succeeded: a turn error is the more specific diagnostic, and - // the persist failure is usually a consequence of the same condition. - if turn_result.is_ok() - && let Some(error) = lock.take_persist_failure() - { - return Err(cmd::Error::from(Error::Workspace(error))); - } - turn_result } @@ -1350,6 +1367,25 @@ fn resolve_new_title(from_heading: bool, generate_auto: bool, content: &str) -> NewTitle::Skip } +/// Fold a drained persist failure into the run's result. +/// +/// With no failure the result passes through. +/// A failure on an otherwise successful run becomes the error, so an unsaved +/// conversation cannot exit zero. +/// Alongside an existing error the failure is attached as metadata: the two can +/// be independent (a provider error and a full disk), and the primary error is +/// the more specific diagnostic. +fn fold_persist_failure(result: Output, persist: Option) -> Output { + match (result, persist) { + (result, None) => result, + (Ok(()), Some(persist)) => Err(cmd::Error::from(Error::Workspace(persist))), + (Err(mut error), Some(persist)) => { + error.push_metadata("persist_failure", persist.to_string()); + Err(error) + } + } +} + /// Apply `--title` / `--no-title` to the resolved conversation. /// /// Both flags act on `metadata.title` directly so the run ends with the title diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index 33940e54a..69c61d86b 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/crates/jp_cli/src/cmd/query_tests.rs @@ -172,6 +172,59 @@ async fn run_mock_turn( .unwrap(); } +/// A drop-time persist failure as the workspace records one. +fn persist_failure() -> jp_workspace::Error { + jp_workspace::Error::Storage(jp_storage::Error::write_failed( + camino::Utf8Path::new("/data/conv/events.json"), + std::io::Error::from(std::io::ErrorKind::StorageFull), + )) +} + +#[test] +fn a_clean_run_passes_through_unchanged() { + assert!(fold_persist_failure(Ok(()), None).is_ok()); +} + +#[test] +fn a_persist_failure_fails_an_otherwise_successful_run() { + let error = + fold_persist_failure(Ok(()), Some(persist_failure())).expect_err("must not exit zero"); + + assert_eq!(error.message.as_deref(), Some("No space left on device")); + assert_eq!(error.code.get(), 1); +} + +#[test] +fn a_persist_failure_rides_along_with_an_existing_error() { + // The two can be independent — a provider error and a full disk — so the + // primary error stays the headline and the persist failure is attached + // rather than dropped. + let error = fold_persist_failure( + Err(cmd::Error::from("Rate limited")), + Some(persist_failure()), + ) + .expect_err("the primary error is preserved"); + + assert_eq!(error.message.as_deref(), Some("Rate limited")); + assert_eq!( + error + .metadata + .iter() + .find(|(key, _)| key == "persist_failure") + .map(|(_, value)| value.as_str().unwrap_or_default()), + Some("Storage error: no space left on device while writing /data/conv/events.json") + ); +} + +#[test] +fn an_existing_error_survives_when_nothing_was_recorded() { + let error = fold_persist_failure(Err(cmd::Error::from("Rate limited")), None) + .expect_err("the primary error is preserved"); + + assert_eq!(error.message.as_deref(), Some("Rate limited")); + assert!(error.metadata.is_empty()); +} + #[test] fn test_query_tools_and_no_tools() { // Create a partial configuration with a few tools. diff --git a/crates/jp_storage/src/error.rs b/crates/jp_storage/src/error.rs index d244f0f14..2e2316f73 100644 --- a/crates/jp_storage/src/error.rs +++ b/crates/jp_storage/src/error.rs @@ -69,19 +69,6 @@ impl Error { Self::WriteFailed { path, source } } } - - /// Whether the error means the filesystem is full. - /// - /// A caller that sees this should stop attempting writes rather than move - /// on to the next one: none of them can succeed until space is freed. - #[must_use] - pub fn is_out_of_space(&self) -> bool { - match self { - Self::OutOfSpace { .. } => true, - Self::Io(error) => is_storage_full(error), - _ => false, - } - } } /// Whether an OS error reports a full filesystem. @@ -89,8 +76,7 @@ impl Error { /// Recognises both a raw OS error (`ENOSPC`, `ERROR_DISK_FULL`) and an error /// constructed directly from the kind, since `std` maps the platform codes onto /// [`io::ErrorKind::StorageFull`]. -#[must_use] -pub fn is_storage_full(error: &io::Error) -> bool { +fn is_storage_full(error: &io::Error) -> bool { error.kind() == io::ErrorKind::StorageFull } diff --git a/crates/jp_storage/src/error_tests.rs b/crates/jp_storage/src/error_tests.rs index 2a507c568..daf8fbf82 100644 --- a/crates/jp_storage/src/error_tests.rs +++ b/crates/jp_storage/src/error_tests.rs @@ -26,7 +26,6 @@ fn write_failed_classifies_full_disk_as_out_of_space() { let error = Error::write_failed(Utf8Path::new("/data/conv/events.json"), os_disk_full()); assert!(matches!(error, Error::OutOfSpace { .. })); - assert!(error.is_out_of_space()); assert_eq!( error.to_string(), "no space left on device while writing /data/conv/events.json" @@ -41,7 +40,6 @@ fn write_failed_classifies_other_errors_as_write_failed() { ); assert!(matches!(error, Error::WriteFailed { .. })); - assert!(!error.is_out_of_space()); assert_eq!(error.to_string(), "failed to write /data/conv/events.json"); } @@ -64,11 +62,4 @@ fn io_variant_renders_the_os_error_verbatim() { let error = Error::from(os_disk_full()); assert_eq!(error.to_string(), "No space left on device (os error 28)"); - assert!(error.is_out_of_space()); -} - -#[test] -fn unrelated_errors_are_not_out_of_space() { - assert!(!Error::NotDir(Utf8Path::new("/data").to_owned()).is_out_of_space()); - assert!(!Error::from(io::Error::from(io::ErrorKind::NotFound)).is_out_of_space()); } diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs index 13c27188d..f941f204e 100644 --- a/crates/jp_storage/src/lib.rs +++ b/crates/jp_storage/src/lib.rs @@ -836,7 +836,8 @@ fn import_external_copy( return Ok(()); }; - fs::create_dir_all(user_conversations)?; + fs::create_dir_all(user_conversations) + .map_err(|error| Error::write_failed(user_conversations, error))?; copy_dir_all( &workspace_conv, &user_conversations.join(id.to_dirname(title)), @@ -844,8 +845,11 @@ fn import_external_copy( } /// Recursively copy directory `src` into `dst`. +/// +/// A failure names the destination path it was writing, and reports a full +/// filesystem as [`Error::OutOfSpace`]. fn copy_dir_all(src: &Utf8Path, dst: &Utf8Path) -> Result<()> { - fs::create_dir_all(dst)?; + fs::create_dir_all(dst).map_err(|error| Error::write_failed(dst, error))?; for entry in dir_entries(src) { let to = dst.join(entry.file_name()); let is_dir = entry.file_type().is_ok_and(|ty| ty.is_dir()); @@ -853,7 +857,7 @@ fn copy_dir_all(src: &Utf8Path, dst: &Utf8Path) -> Result<()> { if is_dir { copy_dir_all(&from, &to)?; } else { - fs::copy(&from, &to)?; + fs::copy(&from, &to).map_err(|error| Error::write_failed(&to, error))?; } } Ok(()) @@ -939,7 +943,7 @@ fn reconcile_conversation_dir( .into_iter() .find(|dir| dir != target) { - fs::rename(&src, target)?; + fs::rename(&src, target).map_err(|error| Error::write_failed(target, error))?; } for dir in conversation_dirs_for_id(conversations_dir, &prefix) { diff --git a/crates/jp_storage/src/lib_tests.rs b/crates/jp_storage/src/lib_tests.rs index 375f73e68..2dbe6a4a2 100644 --- a/crates/jp_storage/src/lib_tests.rs +++ b/crates/jp_storage/src/lib_tests.rs @@ -36,6 +36,29 @@ fn test_storage_new_errors_on_source_file() { } } +#[test] +fn copy_dir_all_failure_names_the_destination_file() { + // The import leg of a persist copies whole conversation directories, so a + // full disk can fail here rather than in `write_json`. Without the path the + // user is told only "IO error". + let tmp = tempdir().unwrap(); + let src = tmp.path().join("src"); + let dst = tmp.path().join("dst"); + fs::create_dir_all(&src).unwrap(); + fs::write(src.join("events.json"), "[]").unwrap(); + + // A directory where the copied file must land fails on all platforms. + let blocker = dst.join("events.json"); + fs::create_dir_all(&blocker).unwrap(); + + let error = copy_dir_all(&src, &dst).expect_err("copying onto a directory should fail"); + + match error { + Error::WriteFailed { path, .. } => assert_eq!(path, blocker), + other => panic!("expected WriteFailed, got {other:?}"), + } +} + #[test] fn test_conversation_dir_name_generation() { let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap(); diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs index 0efe8f1e3..ee1b41baa 100644 --- a/crates/jp_workspace/src/conversation_lock.rs +++ b/crates/jp_workspace/src/conversation_lock.rs @@ -72,13 +72,15 @@ //! The next `flush()` returns it, and //! [`take_persist_failure`][ConversationLock::take_persist_failure] drains it //! for a caller that wants to report at teardown. -//! Either way, one failing disk yields one diagnostic rather than one per -//! mutation scope, and the reporting happens where the output channel and the -//! exit code live. +//! Only the first failure is kept, so one failing disk yields one diagnostic +//! rather than one per mutation scope, and the reporting happens where the +//! output channel and the exit code live. //! -//! A failure that shows the filesystem is full also marks the shared state, and -//! subsequent write attempts are skipped: nothing can succeed until space is -//! freed. +//! Every later scope still attempts its own write. +//! A persist can span two filesystems (the durable user-local copy and the +//! workspace projection), so a failure against one of them says nothing about +//! the other: skipping subsequent writes would strand new events that the +//! healthy root could still accept. use std::sync::{ Arc, @@ -103,12 +105,6 @@ struct PersistState { /// of the same condition, while the first names the write that actually /// broke. failure: Option, - - /// Set once a failure showed the filesystem is full. - /// - /// Sticky, and never cleared: further write attempts are skipped rather - /// than retried against a disk with no room. - out_of_space: bool, } /// Result of attempting to acquire a conversation lock. @@ -405,13 +401,10 @@ impl ConversationMut { let meta = self.metadata.read(); let evts = self.events.read(); - if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) { - // Recorded as futile but not as a pending failure: this error is - // returned, so the caller owns reporting it. - let error = Error::from(error); - self.persist.lock().out_of_space |= error.is_out_of_space(); - return Err(error); - } + // Not recorded on the shared state: this error is returned, so the + // caller owns reporting it. The scope stays dirty, so if the caller + // swallows it, the drop below retries and records that failure. + self.writer.write(&self.id, &meta, &evts, self.projection)?; self.dirty.store(false, Ordering::Relaxed); info!(id = %self.id, "Flushed conversation to disk."); @@ -472,11 +465,6 @@ impl Drop for ConversationMut { return; } - if self.persist.lock().out_of_space { - warn!(id = %self.id, "Skipping persist: no space left on device."); - return; - } - let meta = self.metadata.read(); let evts = self.events.read(); @@ -485,7 +473,6 @@ impl Drop for ConversationMut { warn!(id = %self.id, %error, "Failed to persist conversation."); let mut state = self.persist.lock(); - state.out_of_space |= error.is_out_of_space(); if state.failure.is_none() { state.failure = Some(error); } diff --git a/crates/jp_workspace/src/conversation_lock_tests.rs b/crates/jp_workspace/src/conversation_lock_tests.rs index fa40f6e96..3ed7e7d84 100644 --- a/crates/jp_workspace/src/conversation_lock_tests.rs +++ b/crates/jp_workspace/src/conversation_lock_tests.rs @@ -60,33 +60,18 @@ impl PersistBackend for MockPersistBackend { } } -/// Persistence backend that fails every write with a fixed error, counting the +/// Persistence backend whose every write fails with a full disk, counting the /// attempts. /// -/// The count is what makes the "one failing disk, one report" claim testable: a -/// backend that only records the error cannot distinguish "the write was -/// skipped" from "the write was retried and failed again". -#[derive(Debug)] +/// The count is what makes the retry claims testable: a backend that only +/// records the error cannot distinguish "the write was skipped" from "the write +/// was retried and failed again". +#[derive(Debug, Default)] struct FailingPersistBackend { attempts: AtomicUsize, - out_of_space: bool, } impl FailingPersistBackend { - fn out_of_space() -> Self { - Self { - attempts: AtomicUsize::new(0), - out_of_space: true, - } - } - - fn permission_denied() -> Self { - Self { - attempts: AtomicUsize::new(0), - out_of_space: false, - } - } - fn attempts(&self) -> usize { self.attempts.load(AtomicOrdering::Relaxed) } @@ -102,14 +87,9 @@ impl PersistBackend for FailingPersistBackend { ) -> Result<(), jp_storage::Error> { self.attempts.fetch_add(1, AtomicOrdering::Relaxed); - let source = if self.out_of_space { - io::Error::from(io::ErrorKind::StorageFull) - } else { - io::Error::from(io::ErrorKind::PermissionDenied) - }; Err(jp_storage::Error::write_failed( Utf8Path::new("/data/conv/events.json"), - source, + io::Error::from(io::ErrorKind::StorageFull), )) } @@ -126,10 +106,8 @@ impl PersistBackend for FailingPersistBackend { } } -fn test_lock_with_failing_backend( - backend: FailingPersistBackend, -) -> (ConversationLock, Arc) { - let backend = Arc::new(backend); +fn test_lock_with_failing_backend() -> (ConversationLock, Arc) { + let backend = Arc::new(FailingPersistBackend::default()); let lock = ConversationLock::new( test_handle(), Arc::new(RwLock::new(Conversation::default())), @@ -391,7 +369,7 @@ fn no_persist_failure_recorded_on_a_healthy_backend() { #[test] fn drop_records_persist_failure_on_the_lock() { - let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + let (lock, backend) = test_lock_with_failing_backend(); { let conv = lock.as_mut(); @@ -402,7 +380,6 @@ fn drop_records_persist_failure_on_the_lock() { let error = lock .take_persist_failure() .expect("the failed drop-time write is recorded on the lock"); - assert!(error.is_out_of_space()); assert_eq!( error.to_string(), "Storage error: no space left on device while writing /data/conv/events.json" @@ -411,7 +388,7 @@ fn drop_records_persist_failure_on_the_lock() { #[test] fn recorded_persist_failure_is_returned_only_once() { - let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + let (lock, _backend) = test_lock_with_failing_backend(); { let conv = lock.as_mut(); @@ -426,25 +403,24 @@ fn recorded_persist_failure_is_returned_only_once() { } #[test] -fn out_of_space_skips_every_later_write_attempt() { - let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); +fn every_scope_attempts_its_own_write_after_a_failure() { + // A persist spans two roots that can be separate filesystems. A failure + // against one says nothing about the other, so a scope must never skip its + // write on the strength of an earlier failure: doing so strands new events + // that the healthy root would have accepted. + let (lock, backend) = test_lock_with_failing_backend(); - // Three mutation scopes, each of which would persist on drop. for _ in 0..3 { let conv = lock.as_mut(); conv.update_metadata(|_| {}); } - assert_eq!( - backend.attempts(), - 1, - "a full disk must be hit once, not once per mutation scope" - ); + assert_eq!(backend.attempts(), 3); } #[test] fn out_of_space_records_only_the_first_failure() { - let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + let (lock, _backend) = test_lock_with_failing_backend(); { let conv = lock.as_mut(); @@ -459,26 +435,9 @@ fn out_of_space_records_only_the_first_failure() { assert!(lock.take_persist_failure().is_none()); } -#[test] -fn a_recoverable_failure_does_not_skip_later_writes() { - // Only a full filesystem makes further writes futile. A permission error may - // clear (the user chmods the directory mid-run), so each scope still tries. - let (lock, backend) = - test_lock_with_failing_backend(FailingPersistBackend::permission_denied()); - - for _ in 0..3 { - let conv = lock.as_mut(); - conv.update_metadata(|_| {}); - } - - assert_eq!(backend.attempts(), 3); - let error = lock.take_persist_failure().expect("failure is recorded"); - assert!(!error.is_out_of_space()); -} - #[test] fn flush_returns_a_failure_recorded_by_an_earlier_drop() { - let (lock, backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + let (lock, backend) = test_lock_with_failing_backend(); { let conv = lock.as_mut(); @@ -491,11 +450,14 @@ fn flush_returns_a_failure_recorded_by_an_earlier_drop() { .flush() .expect_err("flush surfaces the earlier drop-time failure"); - assert!(error.is_out_of_space()); + assert_eq!( + error.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); assert_eq!( backend.attempts(), 1, - "flush returns the recorded failure without touching a full disk again" + "the recorded failure is returned without a further write of its own" ); assert!( lock.take_persist_failure().is_none(), @@ -504,34 +466,31 @@ fn flush_returns_a_failure_recorded_by_an_earlier_drop() { } #[test] -fn flush_does_not_record_its_own_failure() { - // `flush` propagates, so its caller owns reporting. Recording it too would - // produce the double report this mechanism exists to prevent. - let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); +fn a_swallowed_flush_failure_still_reaches_a_drain() { + // `flush` records nothing of its own: it propagates, so the caller owns + // reporting. A caller that logs and moves on (the turn loop does this when + // aborting on a fatal stream error) would otherwise lose the fact that + // nothing was saved. The scope stays dirty, so its drop retries and + // records that failure for a teardown drain. + let (lock, backend) = test_lock_with_failing_backend(); let mut conv = lock.as_mut(); conv.update_metadata(|_| {}); assert!(conv.flush().is_err()); - drop(conv); - - assert!(lock.take_persist_failure().is_none()); -} - -#[test] -fn a_failed_flush_leaves_the_scope_dirty_for_the_drop_safety_net() { - let (lock, backend) = - test_lock_with_failing_backend(FailingPersistBackend::permission_denied()); - - let mut conv = lock.as_mut(); - conv.update_metadata(|_| {}); - assert!(conv.flush().is_err()); - assert!(conv.is_dirty()); + assert!( + conv.is_dirty(), + "a failed flush must not clear the dirty flag" + ); drop(conv); assert_eq!( backend.attempts(), 2, - "the drop safety net retries a recoverable flush failure" + "the drop safety net retries the failed flush" + ); + assert!( + lock.take_persist_failure().is_some(), + "the retry's failure is recorded, so a swallowed flush is not silent" ); } @@ -539,7 +498,7 @@ fn a_failed_flush_leaves_the_scope_dirty_for_the_drop_safety_net() { fn a_scope_drains_a_failure_recorded_before_it_existed() { // The lock-owning scope (`into_mut`) is the only drain point once the lock // has been consumed, so it must see failures recorded by earlier scopes. - let (lock, _backend) = test_lock_with_failing_backend(FailingPersistBackend::out_of_space()); + let (lock, _backend) = test_lock_with_failing_backend(); { let conv = lock.as_mut(); @@ -550,7 +509,10 @@ fn a_scope_drains_a_failure_recorded_before_it_existed() { let error = conv .take_persist_failure() .expect("the owning scope sees the earlier failure"); - assert!(error.is_out_of_space()); + assert_eq!( + error.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); assert!(conv.take_persist_failure().is_none()); } diff --git a/crates/jp_workspace/src/error.rs b/crates/jp_workspace/src/error.rs index 953ba586b..767ed3c59 100644 --- a/crates/jp_workspace/src/error.rs +++ b/crates/jp_workspace/src/error.rs @@ -1,5 +1,4 @@ use camino::Utf8PathBuf; -use jp_storage::error::is_storage_full; pub(crate) type Result = std::result::Result; @@ -43,19 +42,6 @@ pub enum Error { } impl Error { - /// Whether the error means the filesystem is full. - /// - /// A caller that sees this should stop attempting writes rather than move - /// on to the next one: none of them can succeed until space is freed. - #[must_use] - pub fn is_out_of_space(&self) -> bool { - match self { - Self::Storage(error) => error.is_out_of_space(), - Self::Io(error) => is_storage_full(error), - _ => false, - } - } - pub fn not_found(target: &'static str, id: &impl ToString) -> Self { Self::NotFound(target, id.to_string()) } diff --git a/docs/.vitepress/rfd-summaries.json b/docs/.vitepress/rfd-summaries.json index 731e64e9b..96e8f3b2e 100644 --- a/docs/.vitepress/rfd-summaries.json +++ b/docs/.vitepress/rfd-summaries.json @@ -268,7 +268,7 @@ "summary": "Automatically retry forced tool calls with reasoning disabled if the model ignores soft-force directives." }, "069-guard-scoped-persistence-for-conversations.md": { - "hash": "19eae2427dbd5a7952b03406e1bf51b985e315786297e64fd659c252e3e5038e", + "hash": "8dc9e83e1b0f57e5ac28a4e16e3ef1feef092e0e33466c6f773a6446a2f528b9", "summary": "Guard-scoped persistence: ConversationMut auto-persists on drop while holding cross-process file lock, with callback-based writes and explicit flush for checkpoints." }, "072-command-plugin-system.md": { diff --git a/docs/rfd/069-guard-scoped-persistence-for-conversations.md b/docs/rfd/069-guard-scoped-persistence-for-conversations.md index 5116aafe8..7d2fe2b01 100644 --- a/docs/rfd/069-guard-scoped-persistence-for-conversations.md +++ b/docs/rfd/069-guard-scoped-persistence-for-conversations.md @@ -177,14 +177,12 @@ impl ConversationMut { impl Drop for ConversationMut { fn drop(&mut self) { if !self.dirty.load(Ordering::Relaxed) { return; } - if self.persist.lock().out_of_space { return; } let meta = self.metadata.read(); let evts = self.events.read(); if let Err(error) = self.writer.write(&self.id, &meta, &evts, self.projection) { warn!(id = %self.id, %error, "Failed to persist conversation."); let mut state = self.persist.lock(); - state.out_of_space |= error.is_out_of_space(); if state.failure.is_none() { state.failure = Some(error); } } } @@ -193,15 +191,22 @@ impl Drop for ConversationMut { `Drop` cannot propagate, so it records the failure on state shared with the originating lock rather than writing to the terminal. -`ConversationLock::take_persist_failure` (and its `ConversationMut` -counterpart) drains it, so the shell reports once, through the printer, with a -non-zero exit code. -One failing disk therefore yields one diagnostic instead of one per mutation -scope. - -A failure that shows the filesystem is full marks the shared state, and -subsequent write attempts are skipped: nothing can succeed until space is -freed. +`ConversationLock::take_persist_failure` (and its `ConversationMut` counterpart) +drains it, so the shell reports once, through the printer, with a non-zero exit +code. +Only the first failure is kept, so one failing disk yields one diagnostic +instead of one per mutation scope. + +Every later scope still attempts its own write. +A persist spans two roots that can be separate filesystems (the durable +user-local copy and the workspace projection), so a failure against one says +nothing about the other; skipping subsequent writes would strand new events that +the healthy root would have accepted. + +`flush` records nothing of its own, since it propagates and the caller owns +reporting. +A caller that swallows that error leaves the scope dirty, so the drop retries +and records the outcome — a swallowed flush failure still reaches a drain. `AtomicBool` is used for the dirty flag instead of `Cell`. `Cell` is `!Sync`, which would make `ConversationMut` `!Sync` and cause From 47d5e79bc462f7b8640862b7023162de85386731 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 28 Jul 2026 10:13:19 +0200 Subject: [PATCH 3/5] review feedback Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd.rs | 19 +++++ crates/jp_cli/src/cmd/query.rs | 21 +---- crates/jp_cli/src/cmd/query_tests.rs | 53 ------------ crates/jp_cli/src/cmd_tests.rs | 50 ++++++++++++ crates/jp_cli/src/lib.rs | 8 ++ crates/jp_cli/src/lib_tests.rs | 80 +++++++++++++++++++ crates/jp_storage/src/lib.rs | 54 ++++++++++++- crates/jp_storage/src/lib_tests.rs | 61 ++++++++++++++ crates/jp_storage/src/validate.rs | 9 ++- crates/jp_workspace/src/conversation_lock.rs | 40 +++++++--- .../src/conversation_lock_tests.rs | 43 +++++++++- crates/jp_workspace/src/lib.rs | 24 ++++++ crates/jp_workspace/src/lib_tests.rs | 61 ++++++++++++++ ...rd-scoped-persistence-for-conversations.md | 17 ++-- 14 files changed, 445 insertions(+), 95 deletions(-) diff --git a/crates/jp_cli/src/cmd.rs b/crates/jp_cli/src/cmd.rs index 2150d796c..71148ecc0 100644 --- a/crates/jp_cli/src/cmd.rs +++ b/crates/jp_cli/src/cmd.rs @@ -178,6 +178,25 @@ impl IntoPartialAppConfig for Commands { pub(crate) type Output = std::result::Result<(), Error>; +/// Fold a drained persist failure into a command's result. +/// +/// With no failure the result passes through. +/// A failure on an otherwise successful run becomes the error, so an unsaved +/// conversation cannot exit zero. +/// Alongside an existing error the failure is attached as metadata: the two can +/// be independent (a provider error and a full disk), and the primary error is +/// the more specific diagnostic. +pub(crate) fn fold_persist_failure(result: Output, persist: Option) -> Output { + match (result, persist) { + (result, None) => result, + (Ok(()), Some(persist)) => Err(Error::from(crate::error::Error::Workspace(persist))), + (Err(mut error), Some(persist)) => { + error.push_metadata("persist_failure", persist.to_string()); + Err(error) + } + } +} + #[derive(Debug, thiserror::Error)] pub(crate) struct Error { /// The error code. diff --git a/crates/jp_cli/src/cmd/query.rs b/crates/jp_cli/src/cmd/query.rs index 035f9d05a..f434af7f8 100644 --- a/crates/jp_cli/src/cmd/query.rs +++ b/crates/jp_cli/src/cmd/query.rs @@ -368,7 +368,7 @@ impl Query { // this the one reliable drain point: a mutation scope that dropped // while dirty recorded its persist failure on the lock, and any `?` in // the region above returns past every other candidate site. - fold_persist_failure(result, lock.take_persist_failure()) + cmd::fold_persist_failure(result, lock.take_persist_failure()) } /// Run the query against an already-locked conversation. @@ -1367,25 +1367,6 @@ fn resolve_new_title(from_heading: bool, generate_auto: bool, content: &str) -> NewTitle::Skip } -/// Fold a drained persist failure into the run's result. -/// -/// With no failure the result passes through. -/// A failure on an otherwise successful run becomes the error, so an unsaved -/// conversation cannot exit zero. -/// Alongside an existing error the failure is attached as metadata: the two can -/// be independent (a provider error and a full disk), and the primary error is -/// the more specific diagnostic. -fn fold_persist_failure(result: Output, persist: Option) -> Output { - match (result, persist) { - (result, None) => result, - (Ok(()), Some(persist)) => Err(cmd::Error::from(Error::Workspace(persist))), - (Err(mut error), Some(persist)) => { - error.push_metadata("persist_failure", persist.to_string()); - Err(error) - } - } -} - /// Apply `--title` / `--no-title` to the resolved conversation. /// /// Both flags act on `metadata.title` directly so the run ends with the title diff --git a/crates/jp_cli/src/cmd/query_tests.rs b/crates/jp_cli/src/cmd/query_tests.rs index 69c61d86b..33940e54a 100644 --- a/crates/jp_cli/src/cmd/query_tests.rs +++ b/crates/jp_cli/src/cmd/query_tests.rs @@ -172,59 +172,6 @@ async fn run_mock_turn( .unwrap(); } -/// A drop-time persist failure as the workspace records one. -fn persist_failure() -> jp_workspace::Error { - jp_workspace::Error::Storage(jp_storage::Error::write_failed( - camino::Utf8Path::new("/data/conv/events.json"), - std::io::Error::from(std::io::ErrorKind::StorageFull), - )) -} - -#[test] -fn a_clean_run_passes_through_unchanged() { - assert!(fold_persist_failure(Ok(()), None).is_ok()); -} - -#[test] -fn a_persist_failure_fails_an_otherwise_successful_run() { - let error = - fold_persist_failure(Ok(()), Some(persist_failure())).expect_err("must not exit zero"); - - assert_eq!(error.message.as_deref(), Some("No space left on device")); - assert_eq!(error.code.get(), 1); -} - -#[test] -fn a_persist_failure_rides_along_with_an_existing_error() { - // The two can be independent — a provider error and a full disk — so the - // primary error stays the headline and the persist failure is attached - // rather than dropped. - let error = fold_persist_failure( - Err(cmd::Error::from("Rate limited")), - Some(persist_failure()), - ) - .expect_err("the primary error is preserved"); - - assert_eq!(error.message.as_deref(), Some("Rate limited")); - assert_eq!( - error - .metadata - .iter() - .find(|(key, _)| key == "persist_failure") - .map(|(_, value)| value.as_str().unwrap_or_default()), - Some("Storage error: no space left on device while writing /data/conv/events.json") - ); -} - -#[test] -fn an_existing_error_survives_when_nothing_was_recorded() { - let error = fold_persist_failure(Err(cmd::Error::from("Rate limited")), None) - .expect_err("the primary error is preserved"); - - assert_eq!(error.message.as_deref(), Some("Rate limited")); - assert!(error.metadata.is_empty()); -} - #[test] fn test_query_tools_and_no_tools() { // Create a partial configuration with a few tools. diff --git a/crates/jp_cli/src/cmd_tests.rs b/crates/jp_cli/src/cmd_tests.rs index 044bf4e5c..9f7cf842b 100644 --- a/crates/jp_cli/src/cmd_tests.rs +++ b/crates/jp_cli/src/cmd_tests.rs @@ -23,6 +23,56 @@ fn rendered(error: &Error) -> String { lines.join("\n") } +/// A drop-time persist failure as the workspace records one. +fn persist_failure() -> jp_workspace::Error { + jp_workspace::Error::Storage(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + io::Error::from(io::ErrorKind::StorageFull), + )) +} + +#[test] +fn a_clean_run_passes_through_unchanged() { + assert!(fold_persist_failure(Ok(()), None).is_ok()); +} + +#[test] +fn a_persist_failure_fails_an_otherwise_successful_run() { + let error = + fold_persist_failure(Ok(()), Some(persist_failure())).expect_err("must not exit zero"); + + assert_eq!(error.message.as_deref(), Some("No space left on device")); + assert_eq!(error.code.get(), 1); +} + +#[test] +fn a_persist_failure_rides_along_with_an_existing_error() { + // The two can be independent — a provider error and a full disk — so the + // primary error stays the headline and the persist failure is attached + // rather than dropped. + let error = fold_persist_failure(Err(Error::from("Rate limited")), Some(persist_failure())) + .expect_err("the primary error is preserved"); + + assert_eq!(error.message.as_deref(), Some("Rate limited")); + assert_eq!( + error + .metadata + .iter() + .find(|(key, _)| key == "persist_failure") + .map(|(_, value)| value.as_str().unwrap_or_default()), + Some("Storage error: no space left on device while writing /data/conv/events.json") + ); +} + +#[test] +fn an_existing_error_survives_when_nothing_was_recorded() { + let error = fold_persist_failure(Err(Error::from("Rate limited")), None) + .expect_err("the primary error is preserved"); + + assert_eq!(error.message.as_deref(), Some("Rate limited")); + assert!(error.metadata.is_empty()); +} + #[cfg(unix)] #[test] fn out_of_space_renders_path_cause_and_action() { diff --git a/crates/jp_cli/src/lib.rs b/crates/jp_cli/src/lib.rs index e94db9cae..7a1cab533 100644 --- a/crates/jp_cli/src/lib.rs +++ b/crates/jp_cli/src/lib.rs @@ -466,6 +466,14 @@ fn run_inner(cli: Cli, format: OutputFormat) -> Result<()> { } }); + // The shutdown arm above drops the command future, so a conversation scope + // that was dirty persists from its `Drop` and records any failure on the + // workspace — with the command's own drain gone along with the future. This + // is the last place that can still tell the user nothing was saved. + // Commands that reported already left nothing behind: the record yields + // each failure once. + let output = cmd::fold_persist_failure(output, ctx.workspace.take_persist_failure()); + if let Err(error) = output.as_ref() && error.disable_persistence { diff --git a/crates/jp_cli/src/lib_tests.rs b/crates/jp_cli/src/lib_tests.rs index e715b0310..e677a4328 100644 --- a/crates/jp_cli/src/lib_tests.rs +++ b/crates/jp_cli/src/lib_tests.rs @@ -64,6 +64,86 @@ fn build_cfg( pipeline.partial_without_conversation() } +/// Persistence backend whose every write fails with a full disk. +#[derive(Debug)] +struct AlwaysFullBackend; + +impl jp_storage::backend::PersistBackend for AlwaysFullBackend { + fn write( + &self, + _id: &ConversationId, + _metadata: &Conversation, + _events: &jp_conversation::ConversationStream, + _projection: jp_storage::backend::Projection, + ) -> std::result::Result<(), jp_storage::Error> { + Err(jp_storage::Error::write_failed( + camino::Utf8Path::new("/data/conv/events.json"), + std::io::Error::from(std::io::ErrorKind::StorageFull), + )) + } + + fn remove(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } + + fn archive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } + + fn unarchive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } +} + +#[test] +fn a_cancelled_command_future_still_reports_its_persist_failure() { + // Mirrors `run_inner`'s shutdown arm: on Ctrl-C the command future is + // dropped mid-flight, so the drain inside it never runs. The command arm + // here parks forever after dirtying the conversation, which holds the + // window open — cancellation always lands while the scope is still alive, + // rather than racing a sleep. + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend)); + let lock = workspace + .create_and_lock_conversation( + Conversation::default(), + Arc::new(AppConfig::new_test()), + None, + ) + .unwrap(); + + let output: cmd::Output = rt.block_on(async { + tokio::select! { + biased; + () = async { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("unsaved".into())); + std::future::pending::<()>().await; + } => unreachable!("the command arm never completes"), + () = std::future::ready(()) => Err(cmd::Error::interrupted()), + } + }); + drop(lock); + + let error = cmd::fold_persist_failure(output, workspace.take_persist_failure()) + .expect_err("the run was interrupted"); + + // The interrupt stays the headline and keeps its exit code; the fact that + // nothing was saved rides along instead of being lost to the log file. + assert_eq!(error.message.as_deref(), Some("Interrupted")); + assert_eq!(error.code.get(), 130); + assert_eq!( + error + .metadata + .iter() + .find(|(key, _)| key == "persist_failure") + .map(|(_, value)| value.as_str().unwrap_or_default()), + Some("Storage error: no space left on device while writing /data/conv/events.json") + ); +} + #[test] fn tracing_guard_persist_returns_explicit_log_file_path() { // `--log-file `: the file lives wherever the caller put it; persist diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs index f941f204e..52b449613 100644 --- a/crates/jp_storage/src/lib.rs +++ b/crates/jp_storage/src/lib.rs @@ -27,6 +27,12 @@ pub(crate) const METADATA_FILE: &str = "metadata.json"; const EVENTS_FILE: &str = "events.json"; const BASE_CONFIG_FILE: &str = "base_config.json"; pub(crate) const CONVERSATIONS_DIR: &str = "conversations"; + +/// Name prefix for a conversation directory being staged by an import. +/// +/// Dot-prefixed so a leftover is invisible to conversation lookups and index +/// scans, both of which match on the bare conversation dirname. +const IMPORT_STAGING_PREFIX: &str = ".import-"; pub(crate) const ARCHIVE_DIR: &str = ".archive"; #[derive(Debug, Clone)] @@ -821,6 +827,10 @@ fn remove_conversation_dirs(id: &ConversationId, conversations_dir: &Utf8Path) - /// name the upcoming write will use. /// A conversation already present in user-local, or one with no workspace copy, /// is left untouched. +/// +/// The copy lands in a staging directory and is renamed into place, so the +/// import is all-or-nothing: a failed or killed run leaves nothing that later +/// runs can mistake for an imported conversation. fn import_external_copy( id: &ConversationId, title: Option<&str>, @@ -838,10 +848,46 @@ fn import_external_copy( fs::create_dir_all(user_conversations) .map_err(|error| Error::write_failed(user_conversations, error))?; - copy_dir_all( - &workspace_conv, - &user_conversations.join(id.to_dirname(title)), - ) + + // Copying straight to the final name would leave a partial directory behind + // on failure. The next run then finds it, treats the import as already done + // and skips it forever, while its fresh mtime makes the loader prefer it + // over the intact workspace copy. Staging plus a rename makes the visible + // result atomic. + let dirname = id.to_dirname(title); + let staging = user_conversations.join(format!("{IMPORT_STAGING_PREFIX}{dirname}")); + let target = user_conversations.join(&dirname); + + // A leftover staging directory is debris from an earlier crash; copying + // into it would carry stale entries across. + let _err = fs::remove_dir_all(&staging); + + if let Err(error) = copy_dir_all(&workspace_conv, &staging) { + let _err = fs::remove_dir_all(&staging); + return Err(error); + } + + if let Err(error) = fs::rename(&staging, &target) { + let _err = fs::remove_dir_all(&staging); + return Err(Error::write_failed(&target, error)); + } + + Ok(()) +} + +/// Remove leftover import staging directories from a `conversations/` +/// directory. +/// +/// One only survives a crash between the copy and the rename, and holds nothing +/// the workspace copy does not still have. +pub(crate) fn cleanup_import_staging(conversations_dir: &Utf8Path) { + for entry in dir_entries(conversations_dir) { + if entry.file_name().starts_with(IMPORT_STAGING_PREFIX) && entry.path().is_dir() { + let path = entry.into_path(); + trace!(%path, "Removing leftover import staging directory."); + let _err = fs::remove_dir_all(path); + } + } } /// Recursively copy directory `src` into `dst`. diff --git a/crates/jp_storage/src/lib_tests.rs b/crates/jp_storage/src/lib_tests.rs index 2dbe6a4a2..b70908b7b 100644 --- a/crates/jp_storage/src/lib_tests.rs +++ b/crates/jp_storage/src/lib_tests.rs @@ -36,6 +36,67 @@ fn test_storage_new_errors_on_source_file() { } } +#[cfg(unix)] +#[test] +fn a_failed_import_leaves_nothing_that_looks_like_a_conversation() { + let tmp = tempdir().unwrap(); + let workspace = tmp.path().join("ws").join("conversations"); + let user = tmp.path().join("user").join("conversations"); + let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap(); + let prefix = id.to_dirname(None); + + let src = workspace.join(&prefix); + fs::create_dir_all(&src).unwrap(); + fs::write(src.join("events.json"), "[]").unwrap(); + fs::write(src.join("notes.md"), "keep me").unwrap(); + // A dangling symlink makes exactly one `fs::copy` fail mid-directory, the + // way a disk filling up would, without having to fill a disk. + std::os::unix::fs::symlink("nowhere", src.join("dangling")).unwrap(); + + let error = import_external_copy(&id, Some("title"), &workspace, &user) + .expect_err("the copy fails on the dangling entry"); + assert!(matches!(error, Error::WriteFailed { .. })); + + // The poisoning this guards against: a partial directory under the real + // name makes every later run skip the import, and its fresh mtime makes the + // loader prefer it over the intact workspace copy. + assert!( + find_normal_conversation_dir_path(&user, &prefix).is_none(), + "a failed import must not leave a conversation directory" + ); + assert_eq!( + dir_entries(&user).count(), + 0, + "the staging directory is cleaned up too" + ); + + // Clear the obstruction: the retry now completes, non-managed file included. + fs::remove_file(src.join("dangling")).unwrap(); + import_external_copy(&id, Some("title"), &workspace, &user).expect("the retry succeeds"); + + let imported = + find_normal_conversation_dir_path(&user, &prefix).expect("the conversation is imported"); + assert_eq!( + fs::read_to_string(imported.join("notes.md")).unwrap(), + "keep me" + ); +} + +#[test] +fn cleanup_import_staging_removes_only_staging_dirs() { + let tmp = tempdir().unwrap(); + let conversations = tmp.path(); + fs::create_dir_all(conversations.join(".import-17457886043-title")).unwrap(); + fs::create_dir_all(conversations.join("17457886043-title")).unwrap(); + fs::create_dir_all(conversations.join(".archive")).unwrap(); + + cleanup_import_staging(conversations); + + assert!(!conversations.join(".import-17457886043-title").exists()); + assert!(conversations.join("17457886043-title").is_dir()); + assert!(conversations.join(".archive").is_dir()); +} + #[test] fn copy_dir_all_failure_names_the_destination_file() { // The import leg of a persist copies whole conversation directories, so a diff --git a/crates/jp_storage/src/validate.rs b/crates/jp_storage/src/validate.rs index 03edac8b6..d3b1be68a 100644 --- a/crates/jp_storage/src/validate.rs +++ b/crates/jp_storage/src/validate.rs @@ -6,8 +6,8 @@ use rayon::prelude::*; use tracing::{debug, trace}; use crate::{ - BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, METADATA_FILE, Storage, dir_entries, - trash::trash_conversation, value::cleanup_tmp_files, + BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, METADATA_FILE, Storage, + cleanup_import_staging, dir_entries, trash::trash_conversation, value::cleanup_tmp_files, }; /// Result of validating all conversation directories across storage roots. @@ -153,6 +153,11 @@ pub(crate) fn trash_invalid_conversation(entry: &InvalidConversation) -> crate:: fn validate_root(conversations_dir: &Utf8Path, result: &mut ValidationResult) { trace!(root = %conversations_dir, "Validating conversation root."); + // Debris from an import killed between its copy and its rename. Swept here + // rather than left in place so a large abandoned copy doesn't sit on the + // user's disk indefinitely. + cleanup_import_staging(conversations_dir); + let entries: Vec<_> = dir_entries(conversations_dir) .filter(|entry| { entry.file_type().ok().is_some_and(|ft| ft.is_dir()) diff --git a/crates/jp_workspace/src/conversation_lock.rs b/crates/jp_workspace/src/conversation_lock.rs index ee1b41baa..b02a00759 100644 --- a/crates/jp_workspace/src/conversation_lock.rs +++ b/crates/jp_workspace/src/conversation_lock.rs @@ -81,6 +81,12 @@ //! workspace projection), so a failure against one of them says nothing about //! the other: skipping subsequent writes would strand new events that the //! healthy root could still accept. +//! +//! The record is owned by the `Workspace`, not by the lock, so a failure +//! recorded while a dropped future unwinds is still there once the lock is +//! gone. +//! `Workspace::take_persist_failure` is the drain of last resort for a run +//! whose command future was cancelled before it could report. use std::sync::{ Arc, @@ -94,11 +100,11 @@ use tracing::{info, warn}; use crate::{error::Error, handle::ConversationHandle}; -/// Shared record of drop-time persistence outcomes for one conversation lock. +/// Shared record of drop-time persistence outcomes. /// /// See the module docs for how a recorded failure reaches the user. #[derive(Debug, Default)] -struct PersistState { +pub(crate) struct PersistState { /// The first drop-time failure no caller has been told about yet. /// /// The first is kept rather than the last: later failures are consequences @@ -107,6 +113,19 @@ struct PersistState { failure: Option, } +impl PersistState { + /// Take the recorded failure, leaving the slot empty. + pub(crate) fn take(&mut self) -> Option { + self.failure.take() + } +} + +/// Shared handle to the record of drop-time persistence failures. +/// +/// Held by the workspace and by every lock and scope derived from it, so a +/// failure recorded while a future unwinds outlives the scope that recorded it. +pub(crate) type PersistFailures = Arc>; + /// Result of attempting to acquire a conversation lock. #[derive(Debug)] pub enum LockResult { @@ -134,7 +153,7 @@ pub struct ConversationLock { writer: Arc, lock_guard: Arc>, projection: Projection, - persist: Arc>, + persist: PersistFailures, } impl ConversationLock { @@ -150,6 +169,7 @@ impl ConversationLock { writer: Arc, lock_guard: Box, projection: Projection, + persist: PersistFailures, ) -> Self { Self { id: handle.into_inner(), @@ -158,7 +178,7 @@ impl ConversationLock { writer, lock_guard: Arc::new(lock_guard), projection, - persist: Arc::default(), + persist, } } @@ -170,7 +190,7 @@ impl ConversationLock { /// recorded here and are never returned twice. #[must_use] pub fn take_persist_failure(&self) -> Option { - self.persist.lock().failure.take() + self.persist.lock().take() } /// The write projection this lock persists with. @@ -279,9 +299,9 @@ pub struct ConversationMut { writer: Arc, projection: Projection, - // Shared with the originating lock and every other scope derived from it, - // so a failure recorded here survives this scope's drop. - persist: Arc>, + // Shared with the workspace, the originating lock, and every other scope + // derived from it, so a failure recorded here survives this scope's drop. + persist: PersistFailures, // Holds the lock guard alive. Released when the last Arc drops. _lock_guard: Arc>, @@ -391,7 +411,7 @@ impl ConversationMut { /// /// After a successful flush, the dirty flag is cleared. pub fn flush(&mut self) -> crate::error::Result<()> { - if let Some(error) = self.persist.lock().failure.take() { + if let Some(error) = self.persist.lock().take() { return Err(error); } @@ -417,7 +437,7 @@ impl ConversationMut { /// that owns the lock, where no `ConversationLock` remains to drain. #[must_use] pub fn take_persist_failure(&self) -> Option { - self.persist.lock().failure.take() + self.persist.lock().take() } /// The write projection this scope persists with. diff --git a/crates/jp_workspace/src/conversation_lock_tests.rs b/crates/jp_workspace/src/conversation_lock_tests.rs index 3ed7e7d84..8fab1bf7e 100644 --- a/crates/jp_workspace/src/conversation_lock_tests.rs +++ b/crates/jp_workspace/src/conversation_lock_tests.rs @@ -107,7 +107,20 @@ impl PersistBackend for FailingPersistBackend { } fn test_lock_with_failing_backend() -> (ConversationLock, Arc) { + let (lock, backend, _) = test_lock_sharing_persist_state(); + (lock, backend) +} + +/// A failing-backend lock plus the persist record the workspace would own. +/// +/// Handed out separately so a test can drain after the lock itself is gone. +fn test_lock_sharing_persist_state() -> ( + ConversationLock, + Arc, + PersistFailures, +) { let backend = Arc::new(FailingPersistBackend::default()); + let failures = PersistFailures::default(); let lock = ConversationLock::new( test_handle(), Arc::new(RwLock::new(Conversation::default())), @@ -115,8 +128,9 @@ fn test_lock_with_failing_backend() -> (ConversationLock, Arc ConversationId { @@ -136,6 +150,7 @@ fn test_lock_with_mock() -> (ConversationLock, Arc) { Arc::clone(&mock) as _, Box::new(NoopLockGuard), Projection::Projected, + PersistFailures::default(), ); (lock, mock) } @@ -148,6 +163,7 @@ fn test_lock_no_writer() -> ConversationLock { Arc::new(NullPersistBackend), Box::new(NoopLockGuard), Projection::Projected, + PersistFailures::default(), ) } @@ -494,6 +510,31 @@ fn a_swallowed_flush_failure_still_reaches_a_drain() { ); } +#[test] +fn a_recorded_failure_outlives_the_lock_that_produced_it() { + // What a cancelled command future leaves behind: the scope and the lock are + // both dropped during the unwind, so the only reachable drain is the record + // the workspace owns. If the record died with the lock, an interrupted + // out-of-space run would report nothing but `Interrupted`. + let (lock, backend, failures) = test_lock_sharing_persist_state(); + + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("unsaved".into())); + } + drop(lock); + + assert_eq!(backend.attempts(), 1); + let error = failures + .lock() + .take() + .expect("the failure is still reachable after the lock is gone"); + assert_eq!( + error.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); +} + #[test] fn a_scope_drains_a_failure_recorded_before_it_existed() { // The lock-owning scope (`into_mut`) is the only drain point once the lock diff --git a/crates/jp_workspace/src/lib.rs b/crates/jp_workspace/src/lib.rs index 05800061d..d06f69db3 100644 --- a/crates/jp_workspace/src/lib.rs +++ b/crates/jp_workspace/src/lib.rs @@ -19,6 +19,7 @@ use std::{ }; use camino::{FromPathBufError, Utf8Path, Utf8PathBuf}; +use conversation_lock::PersistFailures; pub use conversation_lock::{ConversationLock, ConversationMut, LockResult}; pub use error::Error; use error::Result; @@ -65,6 +66,13 @@ pub struct Workspace { /// The in-memory state of the workspace. state: State, + + /// Drop-time persistence failures recorded by conversation scopes. + /// + /// Lives here rather than on a [`ConversationLock`] so a failure recorded + /// while a cancelled future unwinds is still reachable after the lock is + /// gone. + persist_failures: PersistFailures, } impl Workspace { @@ -116,6 +124,7 @@ impl Workspace { locker: backend.clone(), sessions: backend, state: State::default(), + persist_failures: PersistFailures::default(), } } @@ -125,6 +134,18 @@ impl Workspace { &self.root } + /// Take the drop-time persist failure recorded by any conversation scope. + /// + /// The drain of last resort: a scope that dropped while a cancelled future + /// unwound recorded its failure here, and the lock it came from is already + /// gone. + /// Yields each failure once, so draining here after a + /// [`ConversationLock::take_persist_failure`] does not report it twice. + #[must_use] + pub fn take_persist_failure(&self) -> Option { + self.persist_failures.lock().take() + } + /// Set the persist backend. #[must_use] pub fn with_persist(mut self, persist: Arc) -> Self { @@ -453,6 +474,7 @@ impl Workspace { Arc::clone(&self.persist), lock_guard, projection, + Arc::clone(&self.persist_failures), )) } @@ -580,6 +602,7 @@ impl Workspace { Arc::clone(&self.persist), lock_guard, projection, + Arc::clone(&self.persist_failures), ))) } @@ -776,6 +799,7 @@ impl Workspace { Arc::clone(&self.persist), Box::new(NoopLockGuard), projection, + Arc::clone(&self.persist_failures), ) } } diff --git a/crates/jp_workspace/src/lib_tests.rs b/crates/jp_workspace/src/lib_tests.rs index 44bf3085f..ddb891784 100644 --- a/crates/jp_workspace/src/lib_tests.rs +++ b/crates/jp_workspace/src/lib_tests.rs @@ -24,6 +24,67 @@ fn workspace_with_fs(root: impl Into, fs: &FsStorageBackend) -> Wor Workspace::new(root).with_backend(Arc::new(fs.clone())) } +/// Persistence backend whose every write fails with a full disk. +#[derive(Debug)] +struct AlwaysFullBackend; + +impl jp_storage::backend::PersistBackend for AlwaysFullBackend { + fn write( + &self, + _id: &ConversationId, + _metadata: &Conversation, + _events: &ConversationStream, + _projection: jp_storage::backend::Projection, + ) -> std::result::Result<(), jp_storage::Error> { + Err(jp_storage::Error::write_failed( + Utf8Path::new("/data/conv/events.json"), + std::io::Error::from(std::io::ErrorKind::StorageFull), + )) + } + + fn remove(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } + + fn archive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } + + fn unarchive(&self, _id: &ConversationId) -> std::result::Result<(), jp_storage::Error> { + Ok(()) + } +} + +#[test] +fn workspace_drains_a_persist_failure_left_by_a_dropped_lock() { + // The teardown path for a cancelled command future: everything the run held + // is dropped without any drain running, so the workspace is the only place + // left that can still report the conversation was not saved. + let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend)); + let config = Arc::new(AppConfig::new_test()); + + let lock = workspace + .create_and_lock_conversation(Conversation::default(), config, None) + .unwrap(); + { + let conv = lock.as_mut(); + conv.update_metadata(|m| m.title = Some("unsaved".into())); + } + drop(lock); + + let error = workspace + .take_persist_failure() + .expect("the workspace still holds the failure"); + assert_eq!( + error.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); + assert!( + workspace.take_persist_failure().is_none(), + "the failure is reported once" + ); +} + #[test] fn conversation_presence_reflects_creation_intent() { let mut ws = Workspace::new("root"); diff --git a/docs/rfd/069-guard-scoped-persistence-for-conversations.md b/docs/rfd/069-guard-scoped-persistence-for-conversations.md index 7d2fe2b01..f2bf7aab2 100644 --- a/docs/rfd/069-guard-scoped-persistence-for-conversations.md +++ b/docs/rfd/069-guard-scoped-persistence-for-conversations.md @@ -208,6 +208,13 @@ reporting. A caller that swallows that error leaves the scope dirty, so the drop retries and records the outcome — a swallowed flush failure still reaches a drain. +The record is owned by the `Workspace`, not by the lock. +A cancelled command future (Ctrl-C, SIGTERM) is dropped mid-flight: the scope +persists from its `Drop` and the lock goes with it, taking any drain inside the +future along. +`Workspace::take_persist_failure` runs after the command future has resolved or +been dropped, and is the drain of last resort for that path. + `AtomicBool` is used for the dirty flag instead of `Cell`. `Cell` is `!Sync`, which would make `ConversationMut` `!Sync` and cause async futures holding `&ConversationMut` across `.await` points to become @@ -376,11 +383,11 @@ This is slightly more verbose but structurally prevents `?` composes naturally since the callback's return type is forwarded. **Errors in `Drop` cannot be propagated.** A persist failure during -`ConversationMut`'s drop is recorded on the shared persist state and surfaced by -the next `flush()` or by a `take_persist_failure()` drain at teardown, rather -than returned from `drop`. -A scope that owns the lock (`into_mut`) and drops without either is the one -remaining hole: its failure reaches the log file but no caller. +`ConversationMut`'s drop is recorded on the workspace-owned persist state and +surfaced by the next `flush()`, by a `take_persist_failure()` drain in the +command, or by the CLI's teardown drain, rather than returned from `drop`. +The record outliving both the scope and the lock is what makes the teardown +drain reachable after a cancelled future. Long-running loops must call `flush()?` at checkpoints so that I/O failures halt immediately. From fe0c8a638cc5f4775b3107dbc357e4a4ff498eb5 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 28 Jul 2026 11:33:38 +0200 Subject: [PATCH 4/5] review feedback Signed-off-by: Jean Mertz --- crates/jp_cli/src/lib.rs | 15 ++++-- crates/jp_cli/src/lib_tests.rs | 57 +++++++++++++++++++++ crates/jp_storage/src/lib.rs | 15 ------ crates/jp_storage/src/lib_tests.rs | 15 ------ crates/jp_storage/src/validate.rs | 52 ++++++++++++++++--- crates/jp_storage/src/validate_tests.rs | 67 +++++++++++++++++++++++++ docs/.vitepress/rfd-summaries.json | 2 +- 7 files changed, 181 insertions(+), 42 deletions(-) diff --git a/crates/jp_cli/src/lib.rs b/crates/jp_cli/src/lib.rs index 7a1cab533..3ff7f145b 100644 --- a/crates/jp_cli/src/lib.rs +++ b/crates/jp_cli/src/lib.rs @@ -468,8 +468,9 @@ fn run_inner(cli: Cli, format: OutputFormat) -> Result<()> { // The shutdown arm above drops the command future, so a conversation scope // that was dirty persists from its `Drop` and records any failure on the - // workspace — with the command's own drain gone along with the future. This - // is the last place that can still tell the user nothing was saved. + // workspace — with the command's own drain gone along with the future. + // Draining here, before the `disable_persistence` check below, is what lets + // an interrupted unsaved run still say so. // Commands that reported already left nothing behind: the record yields // each failure once. let output = cmd::fold_persist_failure(output, ctx.workspace.take_persist_failure()); @@ -499,8 +500,14 @@ fn run_inner(cli: Cli, format: OutputFormat) -> Result<()> { // shutdown request (Ctrl-C, an interrupt earlier in the run, or SIGTERM) // switches to a 2s cancellation countdown; any further Ctrl-C exits the // process immediately via the signal router's escalation ladder. - rt.block_on(drain_background_tasks(&mut ctx)) - .map_err(Error::Task)?; + let drained = rt.block_on(drain_background_tasks(&mut ctx)); + + // Task sync takes conversation locks of its own — the title generator + // persists its result — so this is the run's last write, after the drain + // above. Held separately from the drain's own error so a failing task and an + // unsaved conversation are not reported as the same thing. + let output = cmd::fold_persist_failure(output, ctx.workspace.take_persist_failure()); + drained.map_err(Error::Task)?; // Remove ephemeral conversations that are no longer needed, but protect // any conversation that is active in a terminal session. diff --git a/crates/jp_cli/src/lib_tests.rs b/crates/jp_cli/src/lib_tests.rs index e677a4328..a8499e6d3 100644 --- a/crates/jp_cli/src/lib_tests.rs +++ b/crates/jp_cli/src/lib_tests.rs @@ -144,6 +144,63 @@ fn a_cancelled_command_future_still_reports_its_persist_failure() { ); } +#[test] +fn a_background_task_persist_failure_is_recorded_after_the_command_finished() { + // `TitleGeneratorTask::sync` takes a conversation lock of its own and + // swallows a failed `flush()` into a `warn!`. The still-dirty scope's drop + // then records on the workspace, which happens while background tasks are + // draining — after the command's own drain has already run. That is why + // `run_inner` folds a second time once the drain returns. + let rt = tokio::runtime::Builder::new_current_thread() + .build() + .unwrap(); + let mut workspace = Workspace::new("root").with_persist(Arc::new(AlwaysFullBackend)); + + let lock = workspace + .create_and_lock_conversation( + Conversation::default(), + Arc::new(AppConfig::new_test()), + None, + ) + .unwrap(); + let id = lock.id(); + // `sync` acquires the lock itself and skips when it is already held, which + // would make every assertion below pass vacuously. + drop(lock); + assert!( + workspace.take_persist_failure().is_none(), + "setup must record nothing, or the failure asserted below proves nothing" + ); + + let cfg = AppConfig::new_test(); + let task = Box::new(jp_task::task::TitleGeneratorTask { + conversation_id: id, + model_id: cfg.assistant.model.id.resolved().clone(), + providers: cfg.providers.llm.clone(), + events: jp_conversation::ConversationStream::new_test(), + title: Some("generated".into()), + is_tty: false, + }); + + // `sync` reports success: it only logs the write failure. + rt.block_on(jp_task::Task::sync(task, &mut workspace)) + .expect("sync swallows the write failure"); + + let recorded = workspace + .take_persist_failure() + .expect("the title task's failed write is recorded on the workspace"); + assert_eq!( + recorded.to_string(), + "Storage error: no space left on device while writing /data/conv/events.json" + ); + + // What the post-drain fold makes of it: a run that would otherwise have + // exited zero now carries the failure. + let error = cmd::fold_persist_failure(Ok(()), Some(recorded)) + .expect_err("an unsaved conversation must not exit zero"); + assert_eq!(error.message.as_deref(), Some("No space left on device")); +} + #[test] fn tracing_guard_persist_returns_explicit_log_file_path() { // `--log-file `: the file lives wherever the caller put it; persist diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs index 52b449613..6891de50e 100644 --- a/crates/jp_storage/src/lib.rs +++ b/crates/jp_storage/src/lib.rs @@ -875,21 +875,6 @@ fn import_external_copy( Ok(()) } -/// Remove leftover import staging directories from a `conversations/` -/// directory. -/// -/// One only survives a crash between the copy and the rename, and holds nothing -/// the workspace copy does not still have. -pub(crate) fn cleanup_import_staging(conversations_dir: &Utf8Path) { - for entry in dir_entries(conversations_dir) { - if entry.file_name().starts_with(IMPORT_STAGING_PREFIX) && entry.path().is_dir() { - let path = entry.into_path(); - trace!(%path, "Removing leftover import staging directory."); - let _err = fs::remove_dir_all(path); - } - } -} - /// Recursively copy directory `src` into `dst`. /// /// A failure names the destination path it was writing, and reports a full diff --git a/crates/jp_storage/src/lib_tests.rs b/crates/jp_storage/src/lib_tests.rs index b70908b7b..73a8af85d 100644 --- a/crates/jp_storage/src/lib_tests.rs +++ b/crates/jp_storage/src/lib_tests.rs @@ -82,21 +82,6 @@ fn a_failed_import_leaves_nothing_that_looks_like_a_conversation() { ); } -#[test] -fn cleanup_import_staging_removes_only_staging_dirs() { - let tmp = tempdir().unwrap(); - let conversations = tmp.path(); - fs::create_dir_all(conversations.join(".import-17457886043-title")).unwrap(); - fs::create_dir_all(conversations.join("17457886043-title")).unwrap(); - fs::create_dir_all(conversations.join(".archive")).unwrap(); - - cleanup_import_staging(conversations); - - assert!(!conversations.join(".import-17457886043-title").exists()); - assert!(conversations.join("17457886043-title").is_dir()); - assert!(conversations.join(".archive").is_dir()); -} - #[test] fn copy_dir_all_failure_names_the_destination_file() { // The import leg of a persist copies whole conversation directories, so a diff --git a/crates/jp_storage/src/validate.rs b/crates/jp_storage/src/validate.rs index d3b1be68a..64e68a251 100644 --- a/crates/jp_storage/src/validate.rs +++ b/crates/jp_storage/src/validate.rs @@ -6,8 +6,8 @@ use rayon::prelude::*; use tracing::{debug, trace}; use crate::{ - BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, METADATA_FILE, Storage, - cleanup_import_staging, dir_entries, trash::trash_conversation, value::cleanup_tmp_files, + BASE_CONFIG_FILE, CONVERSATIONS_DIR, EVENTS_FILE, IMPORT_STAGING_PREFIX, METADATA_FILE, + Storage, dir_entries, trash::trash_conversation, value::cleanup_tmp_files, }; /// Result of validating all conversation directories across storage roots. @@ -128,6 +128,7 @@ impl Storage { continue; } + self.cleanup_import_staging(&conversations_dir); validate_root(&conversations_dir, &mut result); } @@ -139,6 +140,48 @@ impl Storage { result } + + /// Remove leftover import staging directories from one `conversations/` + /// directory. + /// + /// A staging directory only survives a crash between an import's copy and + /// its rename, and holds nothing the workspace copy does not still have, so + /// removing one is pure disk hygiene. + /// + /// Each removal is made while holding the conversation's own lock. + /// Without it, this would race an import running under that lock in another + /// process: `copy_dir_all` recreates directories as it recurses, so a + /// mid-copy removal would leave the importer to rename an incomplete tree + /// into place. + /// A conversation whose lock is held is skipped entirely — an import that + /// is genuinely abandoned holds no lock, and the next locked import + /// pre-cleans its own staging directory anyway. + fn cleanup_import_staging(&self, conversations_dir: &Utf8Path) { + for entry in dir_entries(conversations_dir) { + let Some(dirname) = entry.file_name().strip_prefix(IMPORT_STAGING_PREFIX) else { + continue; + }; + if !entry.path().is_dir() { + continue; + } + + // An unparseable name cannot be tied to a lock, so leave it be + // rather than remove a directory whose owner is unknown. + let Ok(id) = ConversationId::try_from_dirname(dirname) else { + trace!(path = %entry.path(), "Skipping unparseable staging directory."); + continue; + }; + + let Ok(Some(_guard)) = self.try_lock_conversation(&id.to_string(), None) else { + trace!(%id, "Skipping staging directory: conversation is locked."); + continue; + }; + + let path = entry.into_path(); + trace!(%path, "Removing leftover import staging directory."); + let _err = std::fs::remove_dir_all(path); + } + } } /// Trash a conversation that failed validation. @@ -153,11 +196,6 @@ pub(crate) fn trash_invalid_conversation(entry: &InvalidConversation) -> crate:: fn validate_root(conversations_dir: &Utf8Path, result: &mut ValidationResult) { trace!(root = %conversations_dir, "Validating conversation root."); - // Debris from an import killed between its copy and its rename. Swept here - // rather than left in place so a large abandoned copy doesn't sit on the - // user's disk indefinitely. - cleanup_import_staging(conversations_dir); - let entries: Vec<_> = dir_entries(conversations_dir) .filter(|entry| { entry.file_type().ok().is_some_and(|ft| ft.is_dir()) diff --git a/crates/jp_storage/src/validate_tests.rs b/crates/jp_storage/src/validate_tests.rs index 71ee2a38f..831725a85 100644 --- a/crates/jp_storage/src/validate_tests.rs +++ b/crates/jp_storage/src/validate_tests.rs @@ -20,6 +20,73 @@ fn write_valid(storage: &camino::Utf8Path, id: &ConversationId) { write_json(&dir.join(EVENTS_FILE), &events).unwrap(); } +#[test] +fn staging_dirs_are_swept_only_when_the_conversation_is_unlocked() { + // The sweep runs at startup in every process, while another process may be + // importing that same conversation under its lock. Reaping mid-copy would + // let the importer rename an incomplete tree into place, so a held lock has + // to make the sweep skip. Driven by taking the lock in-process rather than + // racing two imports, so the outcome is deterministic. + let tmp = tempdir().unwrap(); + let storage = Storage::new(tmp.path()).unwrap(); + let conversations = tmp.path().join(CONVERSATIONS_DIR); + + let id = ConversationId::try_from_deciseconds_str("17636257526").unwrap(); + let staging = conversations.join(format!(".import-{}", id.to_dirname(Some("title")))); + fs::create_dir_all(staging.join("partial")).unwrap(); + + let guard = storage + .try_lock_conversation(&id.to_string(), None) + .unwrap() + .expect("the lock is free in this test"); + + drop(storage.validate_conversations()); + assert!( + staging.is_dir(), + "an in-flight import's staging directory must survive the sweep" + ); + + drop(guard); + + drop(storage.validate_conversations()); + assert!( + !staging.exists(), + "an abandoned staging directory is reaped once nobody holds the lock" + ); +} + +#[test] +fn the_sweep_leaves_real_conversations_and_dot_dirs_alone() { + let tmp = tempdir().unwrap(); + let storage = Storage::new(tmp.path()).unwrap(); + let conversations = tmp.path().join(CONVERSATIONS_DIR); + + let id = ConversationId::try_from_deciseconds_str("17636257526").unwrap(); + write_valid(tmp.path(), &id); + fs::create_dir_all(conversations.join(".archive")).unwrap(); + + drop(storage.validate_conversations()); + + assert!(conversations.join(id.to_dirname(None)).is_dir()); + assert!(conversations.join(".archive").is_dir()); +} + +#[test] +fn a_staging_dir_with_an_unparseable_name_is_left_in_place() { + // Without a parseable id there is no lock to check, so removing it could + // race an owner this code cannot identify. + let tmp = tempdir().unwrap(); + let storage = Storage::new(tmp.path()).unwrap(); + let conversations = tmp.path().join(CONVERSATIONS_DIR); + + let staging = conversations.join(".import-not-an-id"); + fs::create_dir_all(&staging).unwrap(); + + drop(storage.validate_conversations()); + + assert!(staging.is_dir()); +} + #[test] fn test_valid_conversations_are_collected() { let tmp = tempdir().unwrap(); diff --git a/docs/.vitepress/rfd-summaries.json b/docs/.vitepress/rfd-summaries.json index 96e8f3b2e..ce4841191 100644 --- a/docs/.vitepress/rfd-summaries.json +++ b/docs/.vitepress/rfd-summaries.json @@ -268,7 +268,7 @@ "summary": "Automatically retry forced tool calls with reasoning disabled if the model ignores soft-force directives." }, "069-guard-scoped-persistence-for-conversations.md": { - "hash": "8dc9e83e1b0f57e5ac28a4e16e3ef1feef092e0e33466c6f773a6446a2f528b9", + "hash": "15ef78674300a43b4ba7eeac9ab29415314b560933e8b5d84949e8b7a4c49871", "summary": "Guard-scoped persistence: ConversationMut auto-persists on drop while holding cross-process file lock, with callback-based writes and explicit flush for checkpoints." }, "072-command-plugin-system.md": { From a5ab5174b1045dbc376af6786b804fc8f3418376 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Tue, 28 Jul 2026 14:33:39 +0200 Subject: [PATCH 5/5] review feedback Signed-off-by: Jean Mertz --- crates/jp_storage/src/lib.rs | 40 +++++++++++++++++++++---- crates/jp_storage/src/lib_tests.rs | 34 +++++++++++++++++++++ crates/jp_storage/src/validate.rs | 9 +++--- crates/jp_storage/src/validate_tests.rs | 4 ++- 4 files changed, 77 insertions(+), 10 deletions(-) diff --git a/crates/jp_storage/src/lib.rs b/crates/jp_storage/src/lib.rs index 6891de50e..a2ae350ff 100644 --- a/crates/jp_storage/src/lib.rs +++ b/crates/jp_storage/src/lib.rs @@ -35,6 +35,31 @@ pub(crate) const CONVERSATIONS_DIR: &str = "conversations"; const IMPORT_STAGING_PREFIX: &str = ".import-"; pub(crate) const ARCHIVE_DIR: &str = ".archive"; +/// Build the staging directory name an import copies into. +/// +/// The name is unique per attempt, so a copy can never land in a tree left by +/// an earlier one. +/// That matters because a merge would publish entries the source no longer has: +/// `copy_dir_all` overwrites the files it copies but removes nothing, so +/// anything surviving from an older generation would be renamed into place +/// alongside the fresh copy. +/// +/// The uniqueness suffix goes *after* the conversation dirname so the leading +/// timestamp segment stays intact. +/// The sanitize sweep identifies a leftover by parsing that segment, and can +/// only take the conversation's lock (and so only reap the directory) while it +/// stays parseable. +fn import_staging_dirname(dirname: &str) -> String { + let nanos = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos()); + + format!( + "{IMPORT_STAGING_PREFIX}{dirname}-{}-{nanos}", + std::process::id() + ) +} + #[derive(Debug, Clone)] struct Storage { /// The path to the original storage directory. @@ -854,14 +879,19 @@ fn import_external_copy( // and skips it forever, while its fresh mtime makes the loader prefer it // over the intact workspace copy. Staging plus a rename makes the visible // result atomic. + // + // The staging name is unique per attempt, so this copy cannot land in a tree + // an earlier attempt left behind. Reusing one name and clearing it first + // would make correctness depend on that removal succeeding: a partial + // failure leaves entries the source no longer has, and the rename publishes + // them. let dirname = id.to_dirname(title); - let staging = user_conversations.join(format!("{IMPORT_STAGING_PREFIX}{dirname}")); + let staging = user_conversations.join(import_staging_dirname(&dirname)); let target = user_conversations.join(&dirname); - // A leftover staging directory is debris from an earlier crash; copying - // into it would carry stale entries across. - let _err = fs::remove_dir_all(&staging); - + // Both cleanups are best-effort: a leftover under a unique name is inert. + // It is never copied into, `find_normal_conversation_dir_path` cannot match + // it, and the sanitize sweep reaps it under the conversation's lock. if let Err(error) = copy_dir_all(&workspace_conv, &staging) { let _err = fs::remove_dir_all(&staging); return Err(error); diff --git a/crates/jp_storage/src/lib_tests.rs b/crates/jp_storage/src/lib_tests.rs index 73a8af85d..e47b791d7 100644 --- a/crates/jp_storage/src/lib_tests.rs +++ b/crates/jp_storage/src/lib_tests.rs @@ -82,6 +82,40 @@ fn a_failed_import_leaves_nothing_that_looks_like_a_conversation() { ); } +#[test] +fn an_import_never_merges_into_a_leftover_staging_tree() { + // Debris from a crashed import can survive: the error-path cleanup is + // best-effort, and `remove_dir_all` can partially fail (a read-only file on + // Windows is enough). `copy_dir_all` overwrites the files it copies but + // removes nothing, so a merge would publish entries the workspace copy no + // longer has. Staging under a fresh name per attempt is what rules that out. + let tmp = tempdir().unwrap(); + let workspace = tmp.path().join("ws").join("conversations"); + let user = tmp.path().join("user").join("conversations"); + let id = ConversationId::from_str("jp-c17457886043-otvo8").unwrap(); + let prefix = id.to_dirname(None); + + let src = workspace.join(&prefix); + fs::create_dir_all(&src).unwrap(); + fs::write(src.join("events.json"), "[]").unwrap(); + + // A stale tree holding a file the source does not have, under the name a + // single fixed staging directory would have used. + let stale = user.join(format!(".import-{}", id.to_dirname(Some("title")))); + fs::create_dir_all(&stale).unwrap(); + fs::write(stale.join("stale.md"), "from an older generation").unwrap(); + + import_external_copy(&id, Some("title"), &workspace, &user).expect("the import succeeds"); + + let imported = + find_normal_conversation_dir_path(&user, &prefix).expect("the conversation is imported"); + assert!( + !imported.join("stale.md").exists(), + "the published conversation must hold only what the source has" + ); + assert!(imported.join("events.json").is_file()); +} + #[test] fn copy_dir_all_failure_names_the_destination_file() { // The import leg of a persist copies whole conversation directories, so a diff --git a/crates/jp_storage/src/validate.rs b/crates/jp_storage/src/validate.rs index 64e68a251..47364dd11 100644 --- a/crates/jp_storage/src/validate.rs +++ b/crates/jp_storage/src/validate.rs @@ -146,16 +146,17 @@ impl Storage { /// /// A staging directory only survives a crash between an import's copy and /// its rename, and holds nothing the workspace copy does not still have, so - /// removing one is pure disk hygiene. + /// removing one is pure disk hygiene: each import stages under a fresh name + /// and never reads a leftover. /// /// Each removal is made while holding the conversation's own lock. /// Without it, this would race an import running under that lock in another /// process: `copy_dir_all` recreates directories as it recurses, so a /// mid-copy removal would leave the importer to rename an incomplete tree /// into place. - /// A conversation whose lock is held is skipped entirely — an import that - /// is genuinely abandoned holds no lock, and the next locked import - /// pre-cleans its own staging directory anyway. + /// A conversation whose lock is held is skipped entirely, and because the + /// sweep is only reclaiming disk space, skipping costs nothing but a delay + /// until the next run. fn cleanup_import_staging(&self, conversations_dir: &Utf8Path) { for entry in dir_entries(conversations_dir) { let Some(dirname) = entry.file_name().strip_prefix(IMPORT_STAGING_PREFIX) else { diff --git a/crates/jp_storage/src/validate_tests.rs b/crates/jp_storage/src/validate_tests.rs index 831725a85..d7d0bddf0 100644 --- a/crates/jp_storage/src/validate_tests.rs +++ b/crates/jp_storage/src/validate_tests.rs @@ -32,7 +32,9 @@ fn staging_dirs_are_swept_only_when_the_conversation_is_unlocked() { let conversations = tmp.path().join(CONVERSATIONS_DIR); let id = ConversationId::try_from_deciseconds_str("17636257526").unwrap(); - let staging = conversations.join(format!(".import-{}", id.to_dirname(Some("title")))); + // Built through the production helper: its per-attempt uniqueness suffix is + // exactly what the sweep has to still recognise and be able to lock. + let staging = conversations.join(crate::import_staging_dirname(&id.to_dirname(Some("title")))); fs::create_dir_all(staging.join("partial")).unwrap(); let guard = storage