Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/storage-shutdown-persistence.fixed.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Report storage writer shutdown failures instead of silently accepting unpersisted session snapshots.
37 changes: 32 additions & 5 deletions n00n-storage/src/sessions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3283,15 +3283,18 @@ where
}

/// # Errors
/// Returns `SessionError` if the session file cannot be found or removed.
/// Returns `SessionError` if the session file cannot be found or cleanup fails.
pub fn delete_from(id: n00nId, dir: &Path) -> Result<(), SessionError> {
let _lock = lock_openai_response_chain_in(dir, id)?;
let Some(path) = locate_session_file(dir, id) else {
return Err(StorageError::NotFound(id.to_string()).into());
};
try_remove(&path)?;
let path = locate_session_file(dir, id);
if let Some(path) = &path {
try_remove(path)?;
}
try_remove(&openai_response_chain_path(dir, id))?;
remove_from_cwd_index(dir, id)?;
if path.is_none() {
return Err(StorageError::NotFound(id.to_string()).into());
}
Ok(())
}

Expand Down Expand Up @@ -4740,6 +4743,30 @@ mod tests {
));
}

#[test]
fn delete_missing_primary_cleans_sidecar_and_cwd_index() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path();
let mut session: TestSession = Session::new("m", "/orphaned");
session.save_to(dir).unwrap();
let sidecar = openai_response_chain_path(dir, session.id);
fs::write(&sidecar, b"orphaned").unwrap();
fs::remove_file(jsonl_path(dir, session.id)).unwrap();

let error = TestSession::delete_from(session.id, dir).unwrap_err();

assert!(matches!(
error,
SessionError::Storage(StorageError::NotFound(_))
));
assert!(!sidecar.exists());
assert!(
!load_cwd_index(dir)
.values()
.any(|value| *value == session.id.to_string())
);
}

#[test]
fn title_unicode_safe() {
let input = "あ".repeat(100);
Expand Down
8 changes: 7 additions & 1 deletion n00n-ui/src/app/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -218,7 +218,13 @@ impl Drop for TestStateDir {
return;
};
let writer_stopped = match Arc::try_unwrap(writer) {
Ok(writer) => writer.wait_for_shutdown(TEST_WRITER_DRAIN_TIMEOUT),
Ok(writer) => match writer.wait_for_shutdown(TEST_WRITER_DRAIN_TIMEOUT) {
Ok(()) => true,
Err(error) => {
tracing::warn!(%error, "test storage writer shutdown failed");
false
}
},
Err(writer) => {
drop(writer);
false
Expand Down
15 changes: 10 additions & 5 deletions n00n-ui/src/app/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2581,7 +2581,8 @@ fn drain_writer(app: App, writer: Arc<StorageWriter>) {
Arc::try_unwrap(writer)
.ok()
.expect("app must hold the only other writer reference")
.shutdown(WRITER_DRAIN_TIMEOUT);
.shutdown(WRITER_DRAIN_TIMEOUT)
.unwrap();
}

#[test]
Expand Down Expand Up @@ -2760,7 +2761,8 @@ fn draw_failure_pending_submission_restores_fifo_images_and_control_after_restar
Arc::try_unwrap(writer)
.ok()
.expect("test owns the storage writer")
.shutdown(WRITER_DRAIN_TIMEOUT);
.shutdown(WRITER_DRAIN_TIMEOUT)
.unwrap();

let writer = Arc::new(StorageWriter::new(dir.clone()).unwrap());
let mut restarted = build_app(dir.clone(), Arc::clone(&writer));
Expand Down Expand Up @@ -2799,7 +2801,8 @@ fn draw_failure_pending_submission_restores_fifo_images_and_control_after_restar
Arc::try_unwrap(writer)
.ok()
.expect("test owns the restarted storage writer")
.shutdown(WRITER_DRAIN_TIMEOUT);
.shutdown(WRITER_DRAIN_TIMEOUT)
.unwrap();
}

#[test]
Expand Down Expand Up @@ -2861,7 +2864,8 @@ fn mcp_prompt_draw_failure_survives_restart_without_text_fallback() {
Arc::try_unwrap(writer)
.ok()
.expect("test owns the storage writer")
.shutdown(WRITER_DRAIN_TIMEOUT);
.shutdown(WRITER_DRAIN_TIMEOUT)
.unwrap();

let writer = Arc::new(StorageWriter::new(dir.clone()).unwrap());
let mut restarted = build_app_with_mcp(dir.clone(), Arc::clone(&writer), mcp_reader);
Expand Down Expand Up @@ -2895,7 +2899,8 @@ fn mcp_prompt_draw_failure_survives_restart_without_text_fallback() {
Arc::try_unwrap(writer)
.ok()
.expect("test owns the restarted storage writer")
.shutdown(WRITER_DRAIN_TIMEOUT);
.shutdown(WRITER_DRAIN_TIMEOUT)
.unwrap();
}

#[test]
Expand Down
34 changes: 23 additions & 11 deletions n00n-ui/src/event_loop.rs
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,8 @@ const PERIODIC_SAVE_INTERVAL: Duration = Duration::from_secs(1);
const DRAIN_BUDGET: usize = 256;
const AGENT_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(3);
const STORAGE_WRITER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
const STORAGE_WRITER_REFS_ERR: &str =
"storage writer has outstanding references, skipping graceful shutdown";
const DELETE_FOCUSED_ERR: &str = "cannot delete the focused session";
const NOT_LIVE_ERR: &str = "session not live";
const TEAM_TOOL_NAME: &str = "team";
Expand Down Expand Up @@ -606,8 +608,16 @@ impl<'t> EventLoop<'t> {
// Fatal errors still save every session, shut down MCP transports
// (terminating and reaping their child processes), and drain the
// storage writer before the process exits.
let report = self.shutdown();
result.map(|()| report)
let shutdown = self.shutdown();
match result {
Ok(()) => shutdown,
Err(error) => {
if let Err(shutdown_error) = shutdown {
warn!(error = %shutdown_error, "shutdown after fatal error was incomplete");
}
Err(error)
}
}
}

/// Wait for the next event from any source, or time out so animations
Expand Down Expand Up @@ -1490,7 +1500,7 @@ impl<'t> EventLoop<'t> {
}
}

fn shutdown(mut self) -> ShutdownReport {
fn shutdown(mut self) -> Result<ShutdownReport> {
self.preserve_post_draw_submissions();
let exit = self.sessions[self.focused].app.exit_request;
for rt in &self.sessions {
Expand All @@ -1512,17 +1522,19 @@ impl<'t> EventLoop<'t> {
smol::block_on(h.shutdown());
}
crate::agent::join_all(agent_tasks, AGENT_SHUTDOWN_TIMEOUT);
match Arc::try_unwrap(self.ctx.storage_writer) {
Ok(writer) => writer.shutdown(STORAGE_WRITER_SHUTDOWN_TIMEOUT),
Err(_) => {
warn!("storage writer has outstanding references, skipping graceful shutdown");
}
}
ShutdownReport {
let storage_result = match Arc::try_unwrap(self.ctx.storage_writer) {
Ok(writer) => writer
.shutdown(STORAGE_WRITER_SHUTDOWN_TIMEOUT)
.map_err(Into::into),
Err(_) => Err(eyre!(STORAGE_WRITER_REFS_ERR)),
};
let report = ShutdownReport {
exit,
tabs,
focused: self.focused,
}
};
storage_result?;
Ok(report)
}
}

Expand Down
Loading
Loading