diff --git a/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs b/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs index df2d507..9c3c6ca 100644 --- a/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs +++ b/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs @@ -1365,7 +1365,8 @@ impl MvpAgent { /// `FlushAndAck` on the persistence actor — a true sync barrier that only /// resolves after all queued writes (chat messages, updates) hit disk. /// - /// Returns `Ok(())` on success, `Err(reason)` on timeout or channel failure. + /// Returns `Ok(())` once the writes are on disk, `Err(reason)` on a failed + /// flush, timeout or channel failure. pub(crate) async fn flush_session( &self, session_id: &acp::SessionId, @@ -1384,7 +1385,11 @@ impl MvpAgent { return Err("send failed"); } match tokio::time::timeout(std::time::Duration::from_secs(5), rx).await { - Ok(Ok(_)) => Ok(()), + Ok(Ok(Ok(()))) => Ok(()), + Ok(Ok(Err(error))) => { + tracing::error!(session_id = %session_id.0, %error, "persistence flush failed"); + Err("persistence flush failed") + } Ok(Err(_)) => Err("channel closed"), Err(_) => Err("timeout"), } diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/rewind.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/rewind.rs index 0a15b7e..a9c9e88 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/rewind.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/rewind.rs @@ -548,17 +548,10 @@ impl SessionActor { if report.changed() && !dry_run { // Flush barrier: success must mean the rewrite is on disk. - let (flush_tx, flush_rx) = tokio::sync::oneshot::channel(); - if self - .notifications - .persistence_tx - .send(PersistenceMsg::FlushAndAck { - respond_to: flush_tx, - }) - .is_err() - || flush_rx.await.is_err() - { - anyhow::bail!("history repaired in memory but the persistence flush failed"); + if let Err(error) = self.persistence_barrier().await { + anyhow::bail!( + "history repaired in memory but the persistence flush failed: {error}" + ); } tracing::warn!( session_id = %self.session_info.id.0, diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/turn.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/turn.rs index 746837f..1cf44da 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/turn.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/turn.rs @@ -681,22 +681,14 @@ impl SessionActor { .await .is_some() { - let (flush_tx, flush_rx) = oneshot::channel(); - if self - .notifications - .persistence_tx - .send(PersistenceMsg::FlushAndAck { - respond_to: flush_tx, - }) - .is_ok() - && flush_rx.await.is_ok() - { - let _ = ack.send(()); - } else { - tracing::error!( - session_id = % self.session_info.id.0, prompt_id = % - prompt_id, "persist_ack flush barrier failed" - ); + match self.persistence_barrier().await { + Ok(()) => { + let _ = ack.send(()); + } + Err(error) => tracing::error!( + session_id = % self.session_info.id.0, prompt_id = % prompt_id, + %error, "persist_ack flush barrier failed" + ), } } else { tracing::error!( diff --git a/crates/codegen/kigi-shell/src/session/acp_session_impl/updates.rs b/crates/codegen/kigi-shell/src/session/acp_session_impl/updates.rs index 7342fe1..0270f91 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_impl/updates.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_impl/updates.rs @@ -333,15 +333,24 @@ impl SessionActor { if let Err(e) = crate::session::replay_events::flush_replay_actor(&self.event_tx).await { tracing::warn!(?e, "flush_replay_actor failed"); } + if let Err(error) = self.persistence_barrier().await { + tracing::error!(%error, "persistence flush failed"); + } + } + /// The `FlushAndAck` durability barrier: `Ok` only once every prior + /// persistence write is on stable media. + pub(super) async fn persistence_barrier(&self) -> std::io::Result<()> { let (tx, rx) = tokio::sync::oneshot::channel(); if self .notifications .persistence_tx .send(PersistenceMsg::FlushAndAck { respond_to: tx }) - .is_ok() + .is_err() { - let _ = rx.await; + return Err(std::io::Error::other("persistence actor gone")); } + rx.await + .unwrap_or_else(|_| Err(std::io::Error::other("persistence actor gone"))) } /// Extracts the update type name and relevant parameters for logging fn extract_update_info( diff --git a/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs b/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs index be6e267..d251d97 100644 --- a/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs +++ b/crates/codegen/kigi-shell/src/session/acp_session_tests/cancel_running_task_tests.rs @@ -435,7 +435,7 @@ async fn first_turn_memory_injection_persists_to_chat_history() { respond_to: flush_tx, }) .unwrap(); - flush_rx.await.unwrap(); + flush_rx.await.unwrap().unwrap(); let loaded = storage .load_session_without_updates(&session_info) .await @@ -757,7 +757,7 @@ async fn first_turn_memory_injection_disabled_does_not_persist_to_chat_history() respond_to: flush_tx, }) .unwrap(); - flush_rx.await.unwrap(); + flush_rx.await.unwrap().unwrap(); let storage = crate::session::storage::JsonlStorageAdapter::with_explicit_session_dir( session_dir.path().to_path_buf(), ); @@ -1310,7 +1310,7 @@ async fn actor_with_persistence_drain() -> std::sync::Arc { tokio::task::spawn_local(async move { while let Some(msg) = persistence_rx.recv().await { if let PersistenceMsg::FlushAndAck { respond_to } = msg { - let _ = respond_to.send(()); + let _ = respond_to.send(Ok(())); } } }); diff --git a/crates/codegen/kigi-shell/src/session/commands.rs b/crates/codegen/kigi-shell/src/session/commands.rs index 247b0f3..114d6bd 100644 --- a/crates/codegen/kigi-shell/src/session/commands.rs +++ b/crates/codegen/kigi-shell/src/session/commands.rs @@ -334,10 +334,10 @@ pub enum SessionCommand { CopyFile { respond_to: oneshot::Sender>, }, - /// Flush the replay buffer and persistence, then signal completion. - /// Used during reconnect to ensure all buffered content is persisted before replay. + /// Flush the replay buffer and persistence, then report the durability + /// barrier result. Used during reconnect so replay follows persisted content. FlushComplete { - respond_to: oneshot::Sender<()>, + respond_to: oneshot::Sender>, }, /// Update MCP servers for an existing session (used during reconnect or /// mid-session via the `kigi/session/update_mcp_servers` extension method). diff --git a/crates/codegen/kigi-shell/src/session/persistence.rs b/crates/codegen/kigi-shell/src/session/persistence.rs index 3cd8860..bb71b8d 100644 --- a/crates/codegen/kigi-shell/src/session/persistence.rs +++ b/crates/codegen/kigi-shell/src/session/persistence.rs @@ -359,11 +359,13 @@ pub enum PersistenceMsg { /// stays sequential with other summary.json mutations. GeneratedTitle(String), Flush, - /// Flush all pending writes, then signal the caller once the flush is complete. - /// Unlike `Flush` (fire-and-forget), this is a **sync barrier**: the caller's - /// oneshot only resolves after `flush_pending()` finishes writing to disk. + /// Flush all pending writes AND fsync the session files, then signal the + /// caller. Unlike `Flush` (fire-and-forget, page cache only), this is a + /// **durability barrier**: `Ok` means every prior write is on stable + /// media; `Err` reports a storage write that failed since the last + /// barrier or a failed sync. FlushAndAck { - respond_to: tokio::sync::oneshot::Sender<()>, + respond_to: tokio::sync::oneshot::Sender>, }, /// Flush all pending writes, then copy the current session directory contents and return /// the in-memory snapshot to the caller (who can tar.gz + upload to GCS, etc.). @@ -1392,6 +1394,9 @@ struct SessionPersistence { storage: Arc, /// Pending ACP notification for merging consecutive text chunks pending_notification: Option, + /// First storage failure since the last barrier: `FlushAndAck` must not + /// report `Ok` for state that never reached the file. + pending_write_error: Option, rx: mpsc::UnboundedReceiver, /// Session title generation lifecycle. summary: crate::session::summary::SummaryGenerator, @@ -1491,29 +1496,49 @@ impl SessionPersistence { } } - async fn write_update(&mut self, update: &SessionUpdate) { - if let Err(e) = self.storage.append_update(&self.info, update).await { - tracing::warn!(?e, "failed to write update"); + fn note_storage_failure(&mut self, error: &io::Error) { + if self.pending_write_error.is_none() { + self.pending_write_error = Some(io::Error::new(error.kind(), error.to_string())); } } - /// Flush any pending merged ACP notification to disk. - async fn flush_pending(&mut self) { - // Write any pending merged ACP notification - if let Some(notification) = self.pending_notification.take() { - self.write_update(&SessionUpdate::Acp(Box::new(notification))) - .await; + /// Append an update; a failure is logged and latched for the next barrier. + async fn write_update(&mut self, update: &SessionUpdate) -> io::Result<()> { + let result = self.storage.append_update(&self.info, update).await; + if let Err(error) = &result { + tracing::warn!(%error, "failed to write update"); + self.note_storage_failure(error); + } + result + } + + /// Write the pending merged ACP notification. + async fn flush_pending(&mut self) -> io::Result<()> { + match self.pending_notification.take() { + Some(notification) => { + self.write_update(&SessionUpdate::Acp(Box::new(notification))) + .await + } + None => Ok(()), } } - /// Flush pending writes and sync all session files to disk. - /// Called before CopyFile to ensure all data is persisted. - async fn flush_and_sync(&mut self) { - self.flush_pending().await; - // Sync all session files to disk to ensure they're actually written - if let Err(e) = self.storage.sync_session_files(&self.info).await { - tracing::warn!(?e, "Failed to sync session files to disk"); + /// Fsync every session file to stable media. + async fn sync_files(&self) -> io::Result<()> { + let synced = self.storage.sync_session_files(&self.info).await; + if let Err(error) = &synced { + tracing::warn!(%error, "failed to sync session files to disk"); } + synced + } + + /// The `FlushAndAck` barrier. First error wins: a write that failed since + /// the last barrier outranks a failed sync. + async fn flush_and_sync(&mut self) -> io::Result<()> { + let flushed = self.flush_pending().await; + let prior_write = self.pending_write_error.take().map_or(Ok(()), Err); + let synced = self.sync_files().await; + flushed.and(prior_write).and(synced) } async fn run(mut self) { @@ -1530,34 +1555,38 @@ impl SessionPersistence { } match msg { PersistenceMsg::Flush => { - self.flush_pending().await; + // Logged and latched by write_update; the next barrier reports it. + let _ = self.flush_pending().await; } PersistenceMsg::FlushAndAck { respond_to } => { - self.flush_pending().await; - let _ = respond_to.send(()); + let result = self.flush_and_sync().await; + let _ = respond_to.send(result); } PersistenceMsg::Update(update) => { match update { SessionUpdate::Acp(notification) => { // ACP notifications use merging to coalesce consecutive text chunks if let Some(to_write) = self.maybe_merge_notification(¬ification) { - self.write_update(&SessionUpdate::Acp(Box::new(to_write))) + // Logged and latched by write_update. + let _ = self + .write_update(&SessionUpdate::Acp(Box::new(to_write))) .await; } } SessionUpdate::Xai(_) => { - // xAI notifications are written directly without merging - self.write_update(&update).await; + // Written directly without merging; logged and latched by write_update. + let _ = self.write_update(&update).await; } } } PersistenceMsg::Chat(chat_msg) => { - if let Err(e) = self + if let Err(error) = self .storage .append_chat_message(&self.info, &chat_msg) .await { - tracing::warn!(?e, "failed to write chat message"); + tracing::warn!(%error, "failed to write chat message"); + self.note_storage_failure(&error); } } PersistenceMsg::ReplaceChatHistory(messages) => { @@ -1571,6 +1600,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to replace chat history"); + self.note_storage_failure(&e); } } PersistenceMsg::CurrentModel { @@ -1589,21 +1619,25 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to update current model"); + self.note_storage_failure(&e); } } PersistenceMsg::PlanState(state) => { if let Err(e) = self.storage.write_plan_state(&self.info, &state).await { tracing::warn!(?e, "failed to write plan state"); + self.note_storage_failure(&e); } } PersistenceMsg::PlanModeState(state) => { if let Err(e) = self.storage.write_plan_mode_state(&self.info, &state).await { tracing::warn!(?e, "failed to write plan mode state"); + self.note_storage_failure(&e); } } PersistenceMsg::GoalModeState(state) => { if let Err(e) = self.storage.write_goal_mode_state(&self.info, &state).await { tracing::warn!(?e, "failed to write goal mode state"); + self.note_storage_failure(&e); } } PersistenceMsg::GraphModeState(state) => { @@ -1613,6 +1647,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write graph mode state"); + self.note_storage_failure(&e); } } PersistenceMsg::ContentChunk(content_chunks) => { @@ -1686,12 +1721,14 @@ impl SessionPersistence { } Err(e) => { tracing::warn!(?e, "failed to persist generated session title"); + self.note_storage_failure(&e); } } } PersistenceMsg::RewindPoint(point) => { - if let Err(e) = self.storage.append_rewind_point(&self.info, &point).await { - tracing::warn!(?e, "failed to write rewind point"); + if let Err(error) = self.storage.append_rewind_point(&self.info, &point).await { + tracing::warn!(%error, "failed to write rewind point"); + self.note_storage_failure(&error); } } PersistenceMsg::TruncateRewindPoints { from_index } => { @@ -1701,6 +1738,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, from_index, "failed to truncate rewind points"); + self.note_storage_failure(&e); } } PersistenceMsg::MergeRewindPointsFrom { target_index } => { @@ -1710,6 +1748,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, target_index, "failed to merge rewind points"); + self.note_storage_failure(&e); } } PersistenceMsg::CollectionId(collection_id) => { @@ -1719,6 +1758,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write collection id"); + self.note_storage_failure(&e); } } PersistenceMsg::NextTraceTurn { @@ -1731,11 +1771,13 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write next trace turn"); + self.note_storage_failure(&e); } } PersistenceMsg::Signals(signals) => { if let Err(e) = self.storage.write_signals(&self.info, &signals).await { tracing::warn!(?e, "failed to write session signals"); + self.note_storage_failure(&e); } } PersistenceMsg::AnnouncementState(state) => { @@ -1745,16 +1787,19 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write announcement state"); + self.note_storage_failure(&e); } } PersistenceMsg::Feedback(entry) => { if let Err(e) = self.storage.append_feedback(&self.info, &entry).await { tracing::warn!(?e, "failed to write feedback entry"); + self.note_storage_failure(&e); } } PersistenceMsg::Btw(entry) => { if let Err(e) = self.storage.append_btw(&self.info, &entry).await { tracing::warn!(?e, "failed to write btw entry"); + self.note_storage_failure(&e); } } PersistenceMsg::GitHead { commit, branch } => { @@ -1764,6 +1809,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to persist git HEAD"); + self.note_storage_failure(&e); } } PersistenceMsg::CompactionCheckpoint(checkpoint) => { @@ -1773,6 +1819,7 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write compaction checkpoint file"); + self.note_storage_failure(&e); } } PersistenceMsg::CompactionRequest(request) => { @@ -1782,11 +1829,13 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write compaction request artifact"); + self.note_storage_failure(&e); } } PersistenceMsg::RecapRequest(request) => { if let Err(e) = self.storage.write_recap_request(&self.info, &request).await { tracing::warn!(?e, "failed to write recap request artifact"); + self.note_storage_failure(&e); } } PersistenceMsg::CompactionSegment(segment) => { @@ -1796,20 +1845,25 @@ impl SessionPersistence { .await { tracing::warn!(?e, "failed to write compaction segment"); + self.note_storage_failure(&e); } } PersistenceMsg::CopyFile { one_shot } => { - // Flush pending writes and sync all session files to disk before copying. - self.flush_and_sync().await; - - let result = self.copy_session_dir_to_memory().await; + // Not the durability barrier: the write-failure latch stays for FlushAndAck. + let flushed = self.flush_pending().await; + let result = match flushed.and(self.sync_files().await) { + Ok(()) => self.copy_session_dir_to_memory().await, + Err(error) => { + Err(anyhow::Error::new(error).context("session files are not on disk")) + } + }; let _ = one_shot.send(result); } } } - // Drain the merge buffer on channel close. - self.flush_pending().await; + // Drain the merge buffer on channel close; a failure is logged by write_update. + let _ = self.flush_pending().await; } async fn copy_session_dir_to_memory(&self) -> anyhow::Result { @@ -2002,6 +2056,7 @@ pub(crate) async fn new( info: info_clone, storage: storage.clone(), pending_notification: None, + pending_write_error: None, rx, summary: crate::session::summary::SummaryGenerator::new( crate::session::summary::SummaryConfig { @@ -2070,6 +2125,7 @@ pub async fn new_with_explicit_dir( info: info_clone, storage: storage.clone(), pending_notification: None, + pending_write_error: None, rx, summary: crate::session::summary::SummaryGenerator::new( crate::session::summary::SummaryConfig { @@ -2181,6 +2237,7 @@ pub(crate) async fn load_light( info: loaded_info, storage: storage.clone(), pending_notification: None, + pending_write_error: None, rx, summary: summary_gen, registry_title_sync, @@ -3625,3 +3682,120 @@ mod repo_wide_resolution_tests { ); } } + +#[cfg(test)] +mod barrier_tests { + use super::*; + + fn sampling_client() -> crate::sampling::Client { + crate::sampling::Client::new(kigi_sampler::SamplerConfig { + api_key: Some("test-key".to_string()), + base_url: "http://localhost".to_string(), + model: "test".to_string(), + max_completion_tokens: None, + temperature: None, + top_p: None, + api_backend: Default::default(), + chat_compat: Default::default(), + auth_scheme: Default::default(), + anthropic_oauth: false, + github_copilot: false, + openai_codex: false, + extra_headers: Default::default(), + context_window: 100_000, + force_http1: false, + max_retries: None, + stream_tool_calls: false, + idle_timeout_secs: None, + reasoning_effort: None, + origin_client: None, + attribution_callback: None, + bearer_resolver: None, + supports_backend_search: false, + compactions_remaining: None, + compaction_at_tokens: None, + doom_loop_recovery: None, + header_injector: None, + }) + .expect("sampling client should build") + } + + async fn barrier(handle: &PersistenceHandle) -> io::Result<()> { + let (tx, rx) = tokio::sync::oneshot::channel(); + handle + .tx + .send(PersistenceMsg::FlushAndAck { respond_to: tx }) + .unwrap(); + rx.await.unwrap() + } + + async fn spawn(dir: &std::path::Path) -> (Info, PersistenceHandle) { + let info = Info { + id: acp::SessionId::new("barrier"), + cwd: dir.to_string_lossy().into_owned(), + }; + let handle = new_with_explicit_dir( + &info, + dir.to_path_buf(), + acp::ModelId::new("test-model"), + sampling_client(), + crate::test_support::TEST_MODEL.to_owned(), + ) + .await + .unwrap(); + (info, handle) + } + + /// A chat append that never reached the file fails the next barrier once; + /// the barrier after that is clean again. + #[tokio::test] + async fn flush_and_ack_reports_a_lost_chat_append_once() { + let dir = tempfile::tempdir().unwrap(); + let (info, handle) = spawn(dir.path()).await; + barrier(&handle).await.unwrap(); + + // A directory where the chat log should be makes the append fail. + let chat_file = JsonlStorageAdapter::with_explicit_session_dir(dir.path().to_path_buf()) + .chat_file(&info); + let _ = std::fs::remove_file(&chat_file); + std::fs::create_dir(&chat_file).unwrap(); + handle + .tx + .send(PersistenceMsg::Chat(ConversationItem::user("lost"))) + .unwrap(); + // CopyFile is an ordering fence that keeps the write-failure latch. + let (tx, rx) = tokio::sync::oneshot::channel(); + handle + .tx + .send(PersistenceMsg::CopyFile { one_shot: tx }) + .unwrap(); + let _ = rx.await.unwrap(); + // With the path restored only the lost append can fail the barrier. + std::fs::remove_dir(&chat_file).unwrap(); + let error = barrier(&handle).await.unwrap_err(); + assert_ne!(error.kind(), io::ErrorKind::NotFound, "got {error}"); + + barrier(&handle).await.unwrap(); + } + + /// A history rewrite that never landed fails the next barrier. + #[tokio::test] + async fn flush_and_ack_reports_a_failed_history_replacement() { + let dir = tempfile::tempdir().unwrap(); + let (info, handle) = spawn(dir.path()).await; + let chat_file = JsonlStorageAdapter::with_explicit_session_dir(dir.path().to_path_buf()) + .chat_file(&info); + // The rewrite stages a sibling temp file; a directory there fails it. + std::fs::create_dir(chat_file.with_extension("jsonl.tmp")).unwrap(); + handle + .tx + .send(PersistenceMsg::ReplaceChatHistory(vec![ + ConversationItem::user("rewritten"), + ])) + .unwrap(); + let error = barrier(&handle).await.unwrap_err(); + assert_ne!(error.kind(), io::ErrorKind::NotFound, "got {error}"); + + barrier(&handle).await.unwrap(); + } +} diff --git a/crates/codegen/kigi-shell/src/session/storage/jsonl/mod.rs b/crates/codegen/kigi-shell/src/session/storage/jsonl/mod.rs index 60b7300..ad98e92 100644 --- a/crates/codegen/kigi-shell/src/session/storage/jsonl/mod.rs +++ b/crates/codegen/kigi-shell/src/session/storage/jsonl/mod.rs @@ -86,7 +86,7 @@ impl JsonlStorageAdapter { fn updates_file(&self, info: &Info) -> PathBuf { self.session_dir(info).join("updates.jsonl") } - fn chat_file(&self, info: &Info) -> PathBuf { + pub(crate) fn chat_file(&self, info: &Info) -> PathBuf { self.session_dir(info).join("chat_history.jsonl") } fn summary_file(&self, info: &Info) -> PathBuf { @@ -700,7 +700,7 @@ impl JsonlStorageAdapter { options: super::CopySessionOptions, ) -> io::Result { let target_dir = self.session_dir(target_info); - std::fs::create_dir_all(&target_dir)?; + super::create_dir_all_durable(&target_dir)?; let source_summary = self.read_summary_sync(source_info)?; let chat_format_version = source_summary.chat_format_version; let mut chat_to_copy: Vec = @@ -920,7 +920,7 @@ async fn next_compaction_segment_index(compaction_dir: &std::path::Path) -> u64 impl StorageAdapter for JsonlStorageAdapter { async fn init_session(&self, info: &Info, model_id: acp::ModelId) -> io::Result { let dir = self.session_dir(info); - std::fs::create_dir_all(&dir)?; + super::create_dir_all_durable(&dir)?; let summary_path = self.summary_file(info); if Path::new(&summary_path).exists() { tracing::info!("Loading existing session from JSONL"); @@ -1283,29 +1283,10 @@ impl StorageAdapter for JsonlStorageAdapter { .await } async fn sync_session_files(&self, info: &Info) -> io::Result<()> { - let info_clone = info.clone(); - let adapter_clone = self.clone(); - tokio::task::spawn_blocking(move || -> io::Result<()> { - use std::fs::OpenOptions; - let adapter = adapter_clone; - let files_to_sync = [ - adapter.updates_file(&info_clone), - adapter.chat_file(&info_clone), - adapter.summary_file(&info_clone), - adapter.plan_file(&info_clone), - adapter.rewind_points_file(&info_clone), - ]; - for file_path in &files_to_sync { - if file_path.exists() - && let Ok(file) = OpenOptions::new().write(true).open(file_path) - { - let _ = file.sync_all(); - } - } - Ok(()) - }) - .await - .map_err(io::Error::other)? + let dir = self.session_dir(info); + tokio::task::spawn_blocking(move || super::sync_tree_durable(&dir)) + .await + .map_err(io::Error::other)? } async fn replace_chat_history( &self, diff --git a/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs b/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs index 40c792e..3174244 100644 --- a/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs +++ b/crates/codegen/kigi-shell/src/session/storage/jsonl/tests.rs @@ -2794,3 +2794,24 @@ async fn graph_mode_state_round_trips_and_tombstones() { let after = adapter.load_session_without_updates(&info).await.unwrap(); assert!(after.graph_mode_state.is_none(), "cleared graph must not resurrect"); } + +/// The barrier covers nested state files, and an unsyncable one fails it. +#[cfg(unix)] +#[tokio::test] +async fn sync_session_files_reports_a_nested_file_that_cannot_be_synced() { + use std::os::unix::fs::PermissionsExt; + // Root opens anything; the permission probe is meaningless there. + if unsafe { libc::geteuid() } == 0 { + return; + } + let dir = TempDir::new().unwrap(); + let adapter = JsonlStorageAdapter::with_explicit_session_dir(dir.path().to_path_buf()); + let info = create_test_info(); + adapter.init_session(&info, default_model_id()).await.unwrap(); + let nested = adapter.goal_mode_state_file(&info); + std::fs::create_dir_all(nested.parent().unwrap()).unwrap(); + std::fs::write(&nested, b"{}").unwrap(); + std::fs::set_permissions(&nested, std::fs::Permissions::from_mode(0o000)).unwrap(); + let error = adapter.sync_session_files(&info).await.unwrap_err(); + assert_eq!(error.kind(), io::ErrorKind::PermissionDenied, "got {error}"); +} diff --git a/crates/codegen/kigi-shell/src/session/storage/mod.rs b/crates/codegen/kigi-shell/src/session/storage/mod.rs index 48cca64..a6f6a2b 100644 --- a/crates/codegen/kigi-shell/src/session/storage/mod.rs +++ b/crates/codegen/kigi-shell/src/session/storage/mod.rs @@ -1,5 +1,5 @@ use async_trait::async_trait; -use std::io::{self, BufRead, BufReader, Seek, SeekFrom}; +use std::io::{self, BufRead, BufReader, Seek, SeekFrom, Write}; use std::path::Path; use crate::extensions::notification::SessionNotification; @@ -15,6 +15,137 @@ use agent_client_protocol as acp; use kigi_sampling_types::ReasoningEffort; use kigi_workspace::session::file_state::RewindPoint; +/// Fsync `file` to stable media; macOS `fsync` may stop at the drive cache. +#[cfg(target_os = "macos")] +pub(crate) fn sync_file_durable(file: &std::fs::File) -> io::Result<()> { + use std::os::fd::AsRawFd; + file.sync_all()?; + fullfsync_raw(file.as_raw_fd()) +} + +#[cfg(target_os = "macos")] +fn fullfsync_raw(fd: std::os::fd::RawFd) -> io::Result<()> { + // SAFETY: fcntl on a live descriptor with no pointer arguments. + if unsafe { libc::fcntl(fd, libc::F_FULLFSYNC) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +#[cfg(not(target_os = "macos"))] +pub(crate) fn sync_file_durable(file: &std::fs::File) -> io::Result<()> { + file.sync_all() +} + +/// Fsync `dir` so entries just created or renamed into it survive power loss. +#[cfg(target_os = "macos")] +pub(crate) fn sync_dir_durable(dir: &Path) -> io::Result<()> { + use std::os::fd::AsRawFd; + let file = std::fs::File::open(dir)?; + file.sync_all()?; + fullfsync_raw(file.as_raw_fd()) +} + +#[cfg(all(unix, not(target_os = "macos")))] +pub(crate) fn sync_dir_durable(dir: &Path) -> io::Result<()> { + std::fs::File::open(dir)?.sync_all() +} + +/// Windows has no directory-handle fsync; NTFS journals directory metadata. +#[cfg(windows)] +pub(crate) fn sync_dir_durable(_dir: &Path) -> io::Result<()> { + Ok(()) +} + +/// Write `bytes` to `tmp`, fsync them, rename over `dest`, fsync the parent: +/// a crash never leaves `dest` torn, and the rename never publishes +/// unsynced bytes over a synced file. +pub(crate) fn write_bytes_atomic(tmp: &Path, dest: &Path, bytes: &[u8]) -> io::Result<()> { + let written = std::fs::File::create(tmp).and_then(|mut file| { + file.write_all(bytes)?; + sync_file_durable(&file) + }); + if let Err(error) = written { + let _ = std::fs::remove_file(tmp); + return Err(error); + } + crate::util::fs::replace_file(tmp, dest)?; + sync_parent_dir_durable(dest) +} + +fn sync_parent_dir_durable(path: &Path) -> io::Result<()> { + let parent = path + .parent() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "path has no parent"))?; + sync_dir_durable(parent) +} + +fn open_for_sync(path: &Path) -> io::Result { + let mut options = std::fs::OpenOptions::new(); + options.read(true); + // FlushFileBuffers needs a writable handle; fsync does not. + options.write(cfg!(windows)); + options.open(path) +} + +/// Fsync every file and directory under `dir`, `dir` included, so all +/// session state reaches stable media. A path that vanished is not an error. +pub(crate) fn sync_tree_durable(dir: &Path) -> io::Result<()> { + let entries = match std::fs::read_dir(dir) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + }; + for entry in entries { + let entry = entry?; + let file_type = match entry.file_type() { + Ok(file_type) => file_type, + Err(error) if error.kind() == io::ErrorKind::NotFound => continue, + Err(error) => return Err(error), + }; + if file_type.is_dir() { + sync_tree_durable(&entry.path())?; + } else if file_type.is_file() { + match open_for_sync(&entry.path()) { + Ok(file) => sync_file_durable(&file)?, + Err(error) if error.kind() == io::ErrorKind::NotFound => {} + Err(error) => return Err(error), + } + } + } + match sync_dir_durable(dir) { + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + result => result, + } +} + +/// `create_dir_all`, then fsync every directory that gained an entry so the +/// chain itself survives power loss. A directory that exists but is still +/// empty re-syncs a few ancestors: an earlier create may have died before +/// its sync. +pub(crate) fn create_dir_all_durable(dir: &Path) -> io::Result<()> { + let mut gaining_an_entry = Vec::new(); + let mut cursor = dir; + while !cursor.exists() { + let Some(parent) = cursor.parent() else { break }; + gaining_an_entry.push(parent.to_path_buf()); + cursor = parent; + } + std::fs::create_dir_all(dir)?; + if gaining_an_entry.is_empty() && std::fs::read_dir(dir)?.next().is_none() { + gaining_an_entry.extend(dir.ancestors().skip(1).take(4).map(Path::to_path_buf)); + } + for parent in gaining_an_entry.iter().filter(|parent| !is_fs_root(parent)) { + sync_dir_durable(parent)?; + } + Ok(()) +} + +fn is_fs_root(path: &Path) -> bool { + path.parent() + .is_none_or(|parent| parent.as_os_str().is_empty()) +} + pub mod jsonl; pub mod search; pub mod search_fts; diff --git a/crates/codegen/kigi-shell/src/session/storage/summary_write.rs b/crates/codegen/kigi-shell/src/session/storage/summary_write.rs index 8295dfd..cd65629 100644 --- a/crates/codegen/kigi-shell/src/session/storage/summary_write.rs +++ b/crates/codegen/kigi-shell/src/session/storage/summary_write.rs @@ -228,12 +228,13 @@ fn read_summary(path: &Path) -> io::Result { fn write_summary_atomic(summary_path: &Path, summary: &Summary) -> io::Result<()> { let bytes = serde_json::to_vec_pretty(summary) .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - let tmp = summary_path.with_extension("json.tmp"); - std::fs::write(&tmp, &bytes)?; - // Windows-safe replace: this is the write that persists a session's - // CURRENT MODEL — a bare rename made a switched model silently revert - // on resume whenever AV/indexer held summary.json open on Windows. - crate::util::fs::replace_file(&tmp, summary_path) + // Durable and Windows-safe: this write persists a session's CURRENT MODEL, + // and other writers (`/rename`) race the persistence barrier on this path. + super::write_bytes_atomic( + &summary_path.with_extension("json.tmp"), + summary_path, + &bytes, + ) } #[cfg(test)]