fix(ui): report session persistence failures on shutdown

changelog.d/storage-shutdown-persistence.fixed.md

@@ -0,0 +1 @@
+Report storage writer shutdown failures instead of silently accepting unpersisted session snapshots.

n00n-ui/src/app/mod.rs

@@ -218,7 +218,7 @@ 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) => writer.wait_for_shutdown(TEST_WRITER_DRAIN_TIMEOUT).is_ok(),
             Err(writer) => {
                 drop(writer);
                 false

n00n-ui/src/app/tests.rs

@@ -2551,7 +2551,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]
@@ -2730,7 +2731,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));
@@ -2769,7 +2771,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]
@@ -2831,7 +2834,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);
@@ -2865,7 +2869,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]

n00n-ui/src/event_loop.rs

@@ -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";
@@ -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
@@ -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 {
@@ -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)
     }
 }
 

n00n-ui/src/storage_writer.rs

@@ -27,9 +27,20 @@ struct PendingSnapshot {
 type Pending = Arc<Mutex<HashMap<n00nId, PendingSnapshot>>>;
 type FailedRevisions = HashMap<n00nId, u64>;
 
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum StorageWriterShutdownError {
+    #[error("storage writer stopped with {count} unpersisted snapshot(s)")]
+    UnpersistedSnapshots { count: usize },
+    #[error("storage writer did not drain within {timeout:?}")]
+    Timeout { timeout: Duration },
+    #[error("storage writer completion channel disconnected")]
+    Disconnected,
+}
+
 #[derive(Default)]
 struct RetryState {
     attempts: HashMap<(n00nId, u64), u32>,
+    exhausted: HashMap<n00nId, u64>,
 }
 
 type DeleteCallback = Box<dyn FnOnce(Result<(), SessionError>) + Send>;
@@ -53,15 +64,15 @@ enum Op {
 pub struct StorageWriter {
     pending: Pending,
     ops: flume::Sender<Op>,
-    done_rx: flume::Receiver<()>,
+    done_rx: flume::Receiver<Result<(), usize>>,
 }
 
 impl StorageWriter {
     pub fn new(dir: StateDir) -> std::io::Result<Self> {
         let pending: Pending = Arc::default();
         let writer_pending = Arc::clone(&pending);
         let (ops, ops_rx) = flume::unbounded::<Op>();
-        let (done_tx, done_rx) = flume::bounded::<()>(1);
+        let (done_tx, done_rx) = flume::bounded::<Result<(), usize>>(1);
 
         std::thread::Builder::new()
             .name("storage-writer".into())
@@ -103,9 +114,15 @@ impl StorageWriter {
                         }
                     }
                 }
-                flush(&writer_pending, &mut logs, &mut durable_revisions, &dir);
+                let failed = flush(&writer_pending, &mut logs, &mut durable_revisions, &dir);
                 drop(logs);
-                let _ = done_tx.send(());
+                let unpersisted = retries.unpersisted_count(&failed, &durable_revisions);
+                let completion = if unpersisted == 0 {
+                    Ok(())
+                } else {
+                    Err(unpersisted)
+                };
+                let _ = done_tx.send(completion);
             })?;
 
         Ok(Self {
@@ -159,16 +176,26 @@ impl StorageWriter {
         }
     }
 
-    pub fn shutdown(self, timeout: Duration) {
-        if !self.wait_for_shutdown(timeout) {
-            warn!("storage writer did not drain within {timeout:?}");
-        }
+    pub(crate) fn shutdown(self, timeout: Duration) -> Result<(), StorageWriterShutdownError> {
+        self.wait_for_shutdown(timeout)
     }
 
-    pub(crate) fn wait_for_shutdown(self, timeout: Duration) -> bool {
+    pub(crate) fn wait_for_shutdown(
+        self,
+        timeout: Duration,
+    ) -> Result<(), StorageWriterShutdownError> {
         let Self { ops, done_rx, .. } = self;
         drop(ops);
-        done_rx.recv_timeout(timeout).is_ok()
+        match done_rx.recv_timeout(timeout) {
+            Ok(Ok(())) => Ok(()),
+            Ok(Err(count)) => Err(StorageWriterShutdownError::UnpersistedSnapshots { count }),
+            Err(flume::RecvTimeoutError::Timeout) => {
+                Err(StorageWriterShutdownError::Timeout { timeout })
+            }
+            Err(flume::RecvTimeoutError::Disconnected) => {
+                Err(StorageWriterShutdownError::Disconnected)
+            }
+        }
     }
 }
 
@@ -202,6 +229,10 @@ impl RetryState {
                 .is_some_and(|snapshot| snapshot.revision == revision)
             {
                 pending.remove(&id);
+                self.exhausted
+                    .entry(id)
+                    .and_modify(|current| *current = (*current).max(revision))
+                    .or_insert(revision);
             }
         }
         self.attempts.retain(|(id, revision), _| {
@@ -210,6 +241,24 @@ impl RetryState {
                 .is_some_and(|snapshot| snapshot.revision == *revision)
         });
     }
+
+    fn unpersisted_count(
+        &self,
+        failed: &FailedRevisions,
+        durable_revisions: &HashMap<n00nId, u64>,
+    ) -> usize {
+        failed.len()
+            + self
+                .exhausted
+                .iter()
+                .filter(|(id, revision)| {
+                    !failed.contains_key(id)
+                        && durable_revisions
+                            .get(id)
+                            .is_none_or(|durable| durable < revision)
+                })
+                .count()
+    }
 }
 
 fn flush_and_persist(
@@ -402,7 +451,7 @@ mod tests {
         writer.send(Box::new(b.clone()));
         b.title = "renamed".into();
         writer.send(Box::new(b));
-        writer.shutdown(DRAIN_TIMEOUT);
+        writer.shutdown(DRAIN_TIMEOUT).unwrap();
 
         assert!(AppSession::load(a_id, &dir).is_ok());
         assert_eq!(AppSession::load(b_id, &dir).unwrap().title, "renamed");
@@ -432,6 +481,21 @@ mod tests {
         assert!(lock(&pending).contains_key(&id));
     }
 
+    #[test]
+    fn shutdown_does_not_report_success_with_unpersisted_snapshot() {
+        let (tmp, dir) = state_dir();
+        let writer = StorageWriter::new(dir).unwrap();
+        let session = AppSession::new("test-model", "/tmp/shutdown-failure");
+        let id = session.id;
+        fs::create_dir_all(tmp.path().join(SESSIONS_DIR).join(format!("{id}.jsonl"))).unwrap();
+        writer.send(Box::new(session));
+
+        assert!(matches!(
+            writer.wait_for_shutdown(DRAIN_TIMEOUT),
+            Err(StorageWriterShutdownError::UnpersistedSnapshots { count: 1 })
+        ));
+    }
+
     #[test]
     fn retry_exhaustion_removes_only_exact_failed_revisions() {
         let pending: Pending = Arc::default();
@@ -489,6 +553,36 @@ mod tests {
         assert!(!pending.contains_key(&exact_id));
     }
 
+    #[test]
+    fn exhausted_retry_remains_a_shutdown_failure_until_durable() {
+        let pending: Pending = Arc::default();
+        let mut retries = RetryState::default();
+        let mut session = AppSession::new("test-model", "/tmp/exhausted");
+        session.meta.revision = 4;
+        let id = session.id;
+        lock(&pending).insert(
+            id,
+            PendingSnapshot {
+                revision: 4,
+                session: Box::new(session),
+            },
+        );
+        let failures = HashMap::from([(id, 4)]);
+        for _ in 0..MAX_RETRY_ATTEMPTS {
+            retries.record_failures(&pending, failures.clone());
+        }
+
+        assert!(lock(&pending).is_empty());
+        assert_eq!(
+            retries.unpersisted_count(&FailedRevisions::new(), &HashMap::new()),
+            1
+        );
+        assert_eq!(
+            retries.unpersisted_count(&FailedRevisions::new(), &HashMap::from([(id, 4)]),),
+            0
+        );
+    }
+
     #[test]
     fn exhausted_retry_for_one_session_does_not_block_explicit_persist() {
         let (tmp, dir) = state_dir();
@@ -545,7 +639,7 @@ mod tests {
 
         assert!(done_rx.recv_timeout(DRAIN_TIMEOUT).unwrap().is_ok());
         assert!(AppSession::load(id, &dir).is_ok());
-        writer.shutdown(DRAIN_TIMEOUT);
+        writer.shutdown(DRAIN_TIMEOUT).unwrap();
     }
 
     #[test]
@@ -563,7 +657,7 @@ mod tests {
         let mut second = AppSession::load(id, &dir).unwrap();
         second.title = "same revision, new snapshot".into();
         writer.send(Box::new(second));
-        writer.shutdown(DRAIN_TIMEOUT);
+        writer.shutdown(DRAIN_TIMEOUT).unwrap();
 
         assert_eq!(
             AppSession::load(id, &dir).unwrap().title,
@@ -591,7 +685,7 @@ mod tests {
 
         assert!(done_rx.recv_timeout(DRAIN_TIMEOUT).unwrap().is_ok());
         assert_eq!(AppSession::load(id, &dir).unwrap().title, "periodic save");
-        writer.shutdown(DRAIN_TIMEOUT);
+        writer.shutdown(DRAIN_TIMEOUT).unwrap();
     }
 
     #[test]
@@ -605,7 +699,7 @@ mod tests {
         writer.delete(id, move |res| {
             let _ = done_tx.send(res);
         });
-        writer.shutdown(DRAIN_TIMEOUT);
+        writer.shutdown(DRAIN_TIMEOUT).unwrap();
 
         assert!(done_rx.recv().unwrap().is_ok());
         assert!(AppSession::load(id, &dir).is_err());