diff --git a/cmux-tui/crates/acpmux/src/config.rs b/cmux-tui/crates/acpmux/src/config.rs index 0bb1bc69d93e..d5439f1b891b 100644 --- a/cmux-tui/crates/acpmux/src/config.rs +++ b/cmux-tui/crates/acpmux/src/config.rs @@ -288,7 +288,7 @@ pub(crate) fn private_state_file(path: &Path) -> Result { } #[cfg(unix)] -fn validate_private_state_file(file: &std::fs::File, path: &Path) -> Result<()> { +pub(crate) fn validate_private_state_file(file: &std::fs::File, path: &Path) -> Result<()> { use std::os::unix::fs::{MetadataExt, PermissionsExt}; let metadata = file.metadata()?; diff --git a/cmux-tui/crates/acpmux/src/owner_only.rs b/cmux-tui/crates/acpmux/src/owner_only.rs index c88760197f60..ae5ec508dfab 100644 --- a/cmux-tui/crates/acpmux/src/owner_only.rs +++ b/cmux-tui/crates/acpmux/src/owner_only.rs @@ -198,11 +198,21 @@ pub(crate) fn open_read(path: &Path) -> io::Result { /// Open or create a state file for read/write without following a reparse /// point, narrowing a wider DACL before it can be used as daemon.lock. pub(crate) fn open_rw(path: &Path) -> io::Result { + open_rw_with(path, OPEN_ALWAYS) +} + +/// [`open_rw`] of a state file that must exist already, never made here (a +/// log segment the journal store writes in place). +pub(crate) fn open_rw_existing(path: &Path) -> io::Result { + open_rw_with(path, OPEN_EXISTING) +} + +fn open_rw_with(path: &Path, disposition: u32) -> io::Result { use std::os::windows::fs::MetadataExt; let file = open_owner_only( path, - OPEN_ALWAYS, + disposition, GENERIC_READ | GENERIC_WRITE | 0x80 | WRITE_SECURITY_ACCESS, FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, )?; diff --git a/cmux-tui/crates/acpmux/src/store.rs b/cmux-tui/crates/acpmux/src/store.rs index 1f97118a1154..f4217a3ba838 100644 --- a/cmux-tui/crates/acpmux/src/store.rs +++ b/cmux-tui/crates/acpmux/src/store.rs @@ -1831,7 +1831,7 @@ fn sync_directory(path: &Path) -> Result<()> { .read(true) .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW) .open(path)?; - dir.sync_all().map_err(SyncFailed)?; + log_sync::sync_full(&dir).map_err(SyncFailed)?; #[cfg(test)] sync_model::folder_synced(path); Ok(()) @@ -2139,6 +2139,8 @@ mod journal_meta_tests; mod journal_store_tests; #[cfg(all(test, unix))] mod journal_sync_tests; +#[cfg(all(test, unix))] +mod journal_write_back_tests; impl LocalStore { /// `Store::scan`, and with `strict` `Store::scan_strict`. diff --git a/cmux-tui/crates/acpmux/src/store/journal_meta_tests.rs b/cmux-tui/crates/acpmux/src/store/journal_meta_tests.rs index af59cf607807..bebac3dd9d67 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_meta_tests.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_meta_tests.rs @@ -5,10 +5,12 @@ //! The journal runs on `cmux_journal::SimStorage`; the chat logs are real files whose OS crash //! follows the syncs that ran (`store/sync_model.rs`). +use super::log_sync::{FAIL_NEXT_LOG_SYNC, NOTE_BEFORE_SYNCS_WAIT}; use super::sync_model; use super::*; use cmux_journal::{Crash, SimStorage}; use serde_json::json; +use std::os::unix::fs::PermissionsExt; use std::sync::Arc; use std::time::Duration; @@ -325,6 +327,102 @@ fn the_checkpoint_never_makes_a_folder_removed_outside_the_store_again() { ); } +// ---------------------------------------------------------------- the open's syncs + +/// An open that replays many chats flushes the drive's cache a bounded number of times, not a +/// few times per chat: on macOS each flush (`F_FULLFSYNC`) costs milliseconds, and at 29 ms per +/// replayed chat (fleet ead05e82, aws-m4pro-5) an open after a kill with 300 active chats took +/// 8.8 s, past the daemon's 8 s start budget. Every file is still made durable (the order of the +/// writes is kept with a barrier sync, and one full flush ends the open). +#[test] +fn an_open_that_replays_many_chats_flushes_the_drive_a_bounded_number_of_times() { + use super::log_sync::FULL_SYNCS; + use std::sync::atomic::Ordering; + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(70); + let store = open(home.path(), &journal); + let chats = 40; + for chat in 0..chats { + let id = format!("chat-{chat}"); + store.save(&meta(&id, "chat", 0)).unwrap_or_else(|e| panic!("{e:#}")); + store.append(&id, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for chat in 0..chats { + let id = format!("chat-{chat}"); + store.append(&id, &record(2, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + store.save(&meta(&id, "renamed", 2)).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + let before = FULL_SYNCS.load(Ordering::SeqCst); + let store = open(home.path(), &killed); + let flushes = FULL_SYNCS.load(Ordering::SeqCst).saturating_sub(before); + for chat in 0..chats { + let id = format!("chat-{chat}"); + assert_eq!(seqs(store.as_ref(), &id), vec![1, 2], "{id} lost a line"); + } + assert!( + flushes <= 8, + "the open flushed the drive {flushes} times for {chats} replayed chats (one or two per open, not per chat)" + ); +} + +/// The open after a kill syncs nothing per replayed chat before it returns (the daemon binds its +/// socket after it): the log already holds the dead daemon's lines, only the checkpoint needs them +/// durable, and the checkpoint runs on its own thread after the start. At about 8 ordered syncs +/// per chat, an open with 300 chats took 6.1 s on an owned Mac (fleet 59175d6b, 2 ms a sync +/// under load). Every acknowledged line still survives an OS crash after the checkpoint. +#[test] +fn an_open_after_a_kill_syncs_nothing_per_chat_and_the_checkpoint_makes_it_durable() { + use super::log_sync::ALL_SYNCS; + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(71); + let store = open(home.path(), &journal); + let chats = 40; + for chat in 0..chats { + let id = format!("chat-{chat}"); + store.save(&meta(&id, "chat", 0)).unwrap_or_else(|e| panic!("{e:#}")); + store.append(&id, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for chat in 0..chats { + let id = format!("chat-{chat}"); + store.append(&id, &record(2, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + store.save(&meta(&id, "renamed", 2)).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + // This thread's syncs: the open's own (the daemon binds its socket on this thread after it). + let before = ALL_SYNCS.with(std::cell::Cell::get); + let store = open(home.path(), &killed); + let syncs = ALL_SYNCS.with(std::cell::Cell::get).saturating_sub(before); + for chat in 0..chats { + let id = format!("chat-{chat}"); + assert_eq!(seqs(store.as_ref(), &id), vec![1, 2], "{id} lost a line"); + let name = store.load(&id).unwrap_or_else(|e| panic!("{e:#}")).map(|meta| meta.name); + assert_eq!(name.as_deref(), Some("renamed"), "{id} lost its metadata"); + } + assert!( + syncs <= 4, + "the open synced {syncs} times for {chats} replayed chats before it returned" + ); + // The checkpoint (a timer on its own thread; here at once) makes it all durable. + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + for chat in 0..chats { + let id = format!("chat-{chat}"); + assert_eq!(seqs(store.as_ref(), &id), vec![1, 2], "{id} lost a line in the OS crash"); + } +} + // ---------------------------------------------------------------- review of #35483 (F1 to F4) /// F1 (review of #35483): the metadata twin of S2's finding 1a (lines written with the switch off). @@ -771,3 +869,530 @@ fn a_change_after_the_journal_is_made_again_reaches_session_json() { let on_disk = session_json(home.path(), CHAT).map(|value| value["name"].clone()); assert_eq!(on_disk, Some(json!("b1")), "session.json kept the old journal's image"); } + +// ---------------------------------------------------------------- review of #35866 (P1-1, P2-1, P2-2) + +/// Whether this user can still make a file in a folder of mode 0o500 (root can): the tests +/// below use a folder's mode to make one write fail. +fn mode_stops_writes(folder: &Path) -> bool { + std::fs::set_permissions(folder, std::fs::Permissions::from_mode(0o500)) + .unwrap_or_else(|e| panic!("chmod {}: {e}", folder.display())); + let probe = folder.join("rsync-review-probe"); + let made = std::fs::File::create(&probe).is_ok(); + let _ = std::fs::remove_file(&probe); + std::fs::set_permissions(folder, std::fs::Permissions::from_mode(0o700)) + .unwrap_or_else(|e| panic!("chmod {}: {e}", folder.display())); + !made +} + +/// Review of #35866, P1-1 (the reviewer's test): the open's repair leaves a whole log as it is +/// and puts its segments in `Order::read_segments`; the first checkpoint takes them. When that checkpoint fails before +/// `sync_read_segments` (here: a replayed metadata image whose `session.json` cannot be +/// written yet, so `project_metas` fails), the set is dropped, and the next checkpoint that +/// succeeds retires the records of lines that no sync made durable. An OS crash then loses +/// acknowledged lines. +#[test] +fn a_failed_checkpoint_after_the_open_keeps_the_segments_it_read() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + assert!( + mode_stops_writes(home.path()), + "this test needs a user that a folder's mode stops (not root)" + ); + let journal = SimStorage::new(91); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + // Another chat's metadata change past the checkpoint (its folder made by the save). + store.save(&meta("chat-b", "renamed", 0)).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + // chat-b's folder takes no new file for a while: a checkpoint's write of its + // `session.json` fails (not a failed sync: no freeze). + let folder = home.path().join("sessions").join("chat-b"); + std::fs::set_permissions(&folder, std::fs::Permissions::from_mode(0o500)) + .unwrap_or_else(|e| panic!("{e}")); + let store = open(home.path(), &killed); + // Every checkpoint fails while the folder is so (the checkpointer's first, at once, and + // this one): whichever ran first took the segments the open read. + let first = store.flush(); + std::fs::set_permissions(&folder, std::fs::Permissions::from_mode(0o700)) + .unwrap_or_else(|e| panic!("{e}")); + eprintln!( + "the checkpoint while chat-b's folder was 0o500: {:?}", + first.as_ref().err().map(|e| format!("{e:#}")) + ); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a checkpoint retired records whose log lines the open read and no sync made durable (a failed checkpoint dropped Order::read_segments)" + ); +} + +/// Review of #35866, P2-1 (the reviewer's test): a freeze while a checkpoint is in flight (held +/// right after its start, where it read its LSN: `HOLD_AFTER_START`). Against the marker design +/// of `f9c6a16d1b8`, the checkpoint went on and at its end removed the marker the freeze had just +/// written; the freeze is a journal record now, which no checkpoint in flight can take back, and +/// a fallback marker is never removed after a freeze of its own run. The restart must write the +/// line again: else its sync reports success for pages the failed sync dropped, and its +/// checkpoint retires the line's record. No timer: the checkpoint is let go after the freeze. +#[test] +fn a_freeze_during_a_checkpoint_still_makes_the_restart_write_the_lines_again() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(92); + let store: Arc = Arc::from(open(home.path(), &journal)); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + // A checkpoint in flight (chat-b's record gives it work): held right after its start. + store.append("chat-b", &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + let (started_tx, started) = std::sync::mpsc::channel(); + let (go, go_rx) = std::sync::mpsc::channel(); + let flusher = { + let store = Arc::clone(&store); + std::thread::spawn(move || { + super::journal_store::HOLD_AFTER_START + .with(|hold| *hold.borrow_mut() = Some((started_tx, go_rx))); + store.flush() + }) + }; + started.recv_timeout(WAIT).unwrap_or_else(|e| panic!("the checkpoint never started: {e}")); + // Meanwhile a line past that checkpoint's LSN, and a failed sync of its log on this + // thread (`flush_session`): the store freezes. + store.append(CHAT, &record(2, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let failed = store.flush_session(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + let _ = go.send(()); + let in_flight = flusher.join().unwrap_or_else(|_| panic!("the flush panicked")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // The daemon stops for a restart (no crash); the restart's first checkpoint runs. + drop(store); + let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + vec![1, 2], + "the restart after a failed sync trusted the log and lost an acked line (the checkpoint in flight returned {})", + if in_flight.is_ok() { "ok" } else { "an error" } + ); +} + +/// Review of #35866, P2-2 (the reviewer's test): against the marker design of `f9c6a16d1b8`, the +/// freeze marker was the only thing that made the restart write the lines again, and a failed +/// marker write was only logged. Here the home takes no new file when the log sync fails; on a +/// real disk the marker's own sync can fail with the log's. The freeze is a journal record now, +/// which needs no new file in the home: the restart must still write the lines again. +#[test] +fn a_freeze_in_a_home_that_takes_no_new_file_still_writes_the_lines_again() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + assert!( + mode_stops_writes(home.path()), + "this test needs a user that a folder's mode stops (not root)" + ); + let journal = SimStorage::new(94); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + std::fs::set_permissions(home.path(), std::fs::Permissions::from_mode(0o500)) + .unwrap_or_else(|e| panic!("{e}")); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let failed = store.flush(); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + std::fs::set_permissions(home.path(), std::fs::Permissions::from_mode(0o700)) + .unwrap_or_else(|e| panic!("{e}")); + assert!(failed.is_err(), "the checkpoint went on past a failed log sync"); + // The daemon stops for a restart (no crash); the restart's first checkpoint runs. + drop(store); + let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "the restart after a failed sync in a home that takes no new file trusted the log's pages and lost acked lines" + ); +} + +// ---------------------------------------------------------------- delta review of #35866: a refused frozen record + +// Delta review of #35866 at `f62feba6e3d` (`r-durability-rsync-reviewer`), finding D1: the +/// shutdown's last checkpoint runs after the journal's seal (`Hub::seal`: `Store::seal`, then +/// `Store::flush`; the daemon's exit past its budget does the same after `Hub::flush`). A log sync +/// that fails in it freezes the store, but the `frozen` record finds the journal closed +/// (`AppendError::Closed`) and only an error is logged. The restart in the same boot takes the +/// whole-log path, its checkpoint's sync of a new descriptor reports success for the pages the +/// failed sync dropped, and it retires their records: an OS crash then loses acknowledged lines. +/// The reviewer's test (branch `fix/r-durability-rsync-review-frozen-record-nomerge`, red at `2fd5a947e69` on macOS fleet 8be47d90 and Linux fleet 4e352aa0). +#[test] +fn a_log_sync_that_fails_in_the_shutdowns_last_checkpoint_keeps_the_rewrite() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(95); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // The shutdown's first checkpoint (`Hub::flush`). + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + // Lines that land before the seal (a request on an open connection, an agent past the + // budget), acknowledged: their records are durable. + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // `Hub::seal`: the seal, then a checkpoint; the shutdown's last log sync fails. It is armed + // before the seal: wherever the last checkpoint runs (after the seal, or inside it), its log + // sync on this thread fails. + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + store.seal(); + let failed = store.flush(); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + assert!(failed.is_err(), "the last checkpoint went on past a failed log sync"); + // The daemon exits and starts again in the same boot; the restart's first checkpoint runs. + drop(store); + let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a log sync that failed after the seal journaled no frozen record: the restart trusted the log's pages and lost acked lines" + ); +} + +/// The reviewer's test (delta review of #35866, finding D1, the queue-full form, with no test +/// seam in the store): the writer is behind (one sync of a +/// slow or failing disk takes 8 s here) and other chats' lines fill its queue. A request's +/// `flush_session` fails its log sync; its `frozen` record finds the queue full, the wait for the +/// writer (`SEAL_WAIT`, 5 s) ends first, and the retried append finds it full again. Only an error +/// is logged, and the restart in the same boot trusts the log's pages. +#[test] +fn a_frozen_record_while_the_journals_queue_stays_full_keeps_the_rewrite() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(97); + let small = cmux_journal::Config { max_pending_bytes: 8 * 1024, ..journal_config() }; + let open_small = |sim: &SimStorage| { + open_with_journal(&config(), home.path(), Arc::new(sim.clone()), small.clone()) + .unwrap_or_else(|e| panic!("open the store: {e:#}")) + }; + let store = open_small(&journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // The writer is behind: each sync takes 8 s, and another chat's lines fill its queue. + journal.set_sync_delay(Duration::from_secs(8)); + let full = (1..=100_000u64).any(|seq| store.append("chat-b", &record(seq, "queued")).is_err()); + assert!(full, "the journal's queue never filled"); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let started = std::time::Instant::now(); + let failed = store.flush_session(CHAT); + let took = started.elapsed(); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + journal.set_sync_delay(Duration::ZERO); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + eprintln!("the failed flush_session returned after {took:?}"); + // The daemon stops for a restart in the same boot (no crash); the restart's checkpoint runs. + drop(store); + let store = open_small(&journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open_small(&crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "the frozen record found the journal's queue full twice and was not journaled: the restart trusted the log's pages and lost acked lines" + ); +} + +/// Delta review of #35866, finding D1, a seal while a freeze is under way: a request's +/// `flush_session` fails its log sync and stops right before it freezes the store +/// (`HOLD_BEFORE_FREEZE`), and a shutdown seals the store meanwhile (begun by another cause: a +/// signal, an update's handoff). The freeze's `frozen` record must still reach the journal: the +/// journal closes only after the shutdown's last checkpoint, which waits for the sync under way. +/// The sealing thread says when it waits for that sync (`NOTE_BEFORE_SYNCS_WAIT`) or that the +/// seal returned, and only then is the freeze let go on: no timer. Red where the seal closes the +/// journal at once (the record is refused, `AppendError::Closed`). +#[test] +fn a_seal_while_a_freeze_is_under_way_keeps_the_rewrite() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(98); + let store: Arc = Arc::from(open(home.path(), &journal)); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // A request's flush_session: its log sync fails, and it stops right before it freezes. + let (arrived_tx, arrived) = std::sync::mpsc::channel(); + let (release, release_rx) = std::sync::mpsc::channel(); + let failing = { + let store = Arc::clone(&store); + std::thread::spawn(move || { + super::journal_store::HOLD_BEFORE_FREEZE + .with(|hold| *hold.borrow_mut() = Some((arrived_tx, release_rx))); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let result = store.flush_session(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + result + }) + }; + arrived.recv_timeout(WAIT).unwrap_or_else(|e| panic!("the failed sync never froze: {e}")); + // The shutdown's seal, on another thread: it says when it waits for the sync under way, or + // that it returned. + let (events_tx, events) = std::sync::mpsc::channel(); + let sealer = { + let store = Arc::clone(&store); + std::thread::spawn(move || { + NOTE_BEFORE_SYNCS_WAIT.with(|note| *note.borrow_mut() = Some(events_tx.clone())); + store.seal(); + let _ = events_tx.send("returned"); + }) + }; + let first = events.recv_timeout(WAIT).unwrap_or_else(|e| panic!("no word from the seal: {e}")); + let _ = release.send(()); + let failed = failing.join().unwrap_or_else(|_| panic!("the failing thread panicked")); + sealer.join().unwrap_or_else(|_| panic!("the seal panicked")); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + // The restart in the same boot; its first checkpoint runs. + drop(store); + let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a seal while a freeze was under way lost its frozen record (the seal {}): the restart trusted the log's pages and lost acked lines", + if first == "returned" { "closed the journal at once" } else { "waited for the sync" } + ); +} + +/// Delta review of #35866, D1, fail closed: a `frozen` record the journal still refuses (its +/// writer stopped: `AppendError::Closed`; test seam `FAIL_FROZEN_RECORD`) leaves a durable marker +/// (`journal-frozen`) instead, and the restart writes every replayed chat's lines again for it as +/// for the record. The first checkpoint after that restart removes the marker. +#[test] +fn a_refused_frozen_record_falls_back_to_a_durable_marker() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(99); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + super::journal_store::FAIL_FROZEN_RECORD.with(|fail| fail.set(true)); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let failed = store.flush_session(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + super::journal_store::FAIL_FROZEN_RECORD.with(|fail| fail.set(false)); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + let marker = home.path().join("journal-frozen"); + assert!(marker.exists(), "a refused frozen record left no marker"); + // The restart in the same boot: the marker makes it write the lines again, and its first + // checkpoint removes the marker. + drop(store); + let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + assert!(!marker.exists(), "the checkpoint after the rewrite kept the marker"); + let crashed = journal.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a refused frozen record left the restart nothing: it trusted the log's pages and lost acked lines" + ); +} + +/// Delta review of #35866, D1, fail closed, and the bound of #35895: a freeze's fallback marker is +/// written off the failed sync's count, so a checkpoint never waits for its disk writes. A +/// checkpoint in flight (held right after its start, `HOLD_AFTER_START`); another thread's +/// `flush_session` fails its log sync, its `frozen` record is refused (`FAIL_FROZEN_RECORD`), and +/// it stops right before it writes the marker (`HOLD_BEFORE_FROZEN_MARKER`). The checkpoint must +/// return, refused, before that thread goes on; it says if it waits for a sync under way +/// (`NOTE_BEFORE_SYNCS_WAIT`): no timer. +#[test] +fn a_checkpoint_does_not_wait_for_a_freezes_fallback_marker() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(100); + let store: Arc = Arc::from(open(home.path(), &journal)); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + store.append(CHAT, &record(2, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // A checkpoint in flight: past its start. It says when it waits for a sync, or that it + // returned. + let (started_tx, started) = std::sync::mpsc::channel(); + let (go, go_rx) = std::sync::mpsc::channel(); + let (events_tx, events) = std::sync::mpsc::channel(); + let flusher = { + let store = Arc::clone(&store); + std::thread::spawn(move || { + super::journal_store::HOLD_AFTER_START + .with(|hold| *hold.borrow_mut() = Some((started_tx, go_rx))); + NOTE_BEFORE_SYNCS_WAIT.with(|note| *note.borrow_mut() = Some(events_tx.clone())); + let result = store.flush(); + let _ = events_tx.send("returned"); + result + }) + }; + started.recv_timeout(WAIT).unwrap_or_else(|e| panic!("the checkpoint never started: {e}")); + // Another thread's flush_session: its log sync fails, its record is refused, and it stops + // right before its marker. + let (arrived_tx, arrived) = std::sync::mpsc::channel(); + let (release, release_rx) = std::sync::mpsc::channel(); + let failing = { + let store = Arc::clone(&store); + std::thread::spawn(move || { + super::journal_store::HOLD_BEFORE_FROZEN_MARKER + .with(|hold| *hold.borrow_mut() = Some((arrived_tx, release_rx))); + super::journal_store::FAIL_FROZEN_RECORD.with(|fail| fail.set(true)); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let result = store.flush_session(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + result + }) + }; + arrived.recv_timeout(WAIT).unwrap_or_else(|e| panic!("the failed sync never froze: {e}")); + let _ = go.send(()); + let first = + events.recv_timeout(WAIT).unwrap_or_else(|e| panic!("no word from the checkpoint: {e}")); + let _ = release.send(()); + let failed = failing.join().unwrap_or_else(|_| panic!("the failing thread panicked")); + let checkpointed = flusher.join().unwrap_or_else(|_| panic!("the flush panicked")); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + assert_eq!(first, "returned", "a checkpoint waited for a freeze's fallback marker"); + assert!(checkpointed.is_err(), "a checkpoint went past a freeze"); + assert!(home.path().join("journal-frozen").exists(), "the refused record left no marker"); +} + +// ---------------------------------------------------------------- delta review 2 of #35866: a kill before the freeze is durable + +/// The reviewer's test (delta review 2 of #35866 at `a2475c77387`, finding D2, window 1; its +/// repro `rsync_review3_a_kill_while_the_frozen_record_waits_in_the_queue_keeps_the_rewrite`, red +/// on macOS fleet e150f624 and Linux fleet e2ba4960), with a hold instead of a 2 s sync and a +/// 300 ms sleep: a kill while the `frozen` record waits in the journal's queue. The journal's +/// writer is inside a sync of an earlier batch (held, `SimStorage::hold_next_sync`); a request's +/// `flush_session` fails its log sync, and the store freezes: the record is taken into the queue, +/// so no marker is written. A kill before the writer writes it loses it. The restart in the same +/// boot must still not trust the pages the failed sync dropped: after its checkpoint and an OS +/// crash, every acknowledged line is there. +#[test] +fn a_kill_while_the_frozen_record_waits_in_the_queue_keeps_the_lines() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(101); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + // Acknowledged: their records are durable. + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // The writer is busy: it is inside the sync of chat-b's record, held. + let (in_sync, release) = journal.hold_next_sync(); + store.append("chat-b", &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + in_sync.recv_timeout(WAIT).unwrap_or_else(|e| panic!("the writer never began its sync: {e}")); + // A request's flush_session: its log sync fails, the store freezes; the frozen record waits + // in the queue behind the sync under way. + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let failed = store.flush_session(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + assert!(failed.is_err(), "flush_session's sync did not fail as the test asked"); + // A kill now, before the writer writes the record; the held sync then fails as crashed. + let killed = journal.restart(Crash::Process); + drop(release); + drop(store); + // The restart in the same boot; its first checkpoint runs. + let store = open(home.path(), &killed); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a kill while the frozen record waited in the journal's queue: the restart trusted the log's pages and lost acked lines" + ); +} + +/// Delta review 2 of #35866, finding D2, window 2 (found by reading; this is its test): a log's +/// close (`close_log`, an eviction, a segment's end) fails its sync and only notes it in the store's +/// `sync_failed` flag; the store freezes at the next checkpoint, which reads the flag (up to 30 s +/// later). A kill before that checkpoint leaves no frozen record and no marker, so the restart in +/// the same boot finds a whole log and must still not trust the pages the failed sync dropped: +/// after its checkpoint and an OS crash, every acknowledged line is there. +#[test] +fn a_kill_after_a_failed_log_close_before_any_checkpoint_keeps_the_lines() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(102); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + // Acknowledged: their records are durable. + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + // The chat's log closes and its sync fails: only the flag notes it. + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + store.close_log(CHAT); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + // A kill now, before any checkpoint reads the flag. + let killed = journal.restart(Crash::Process); + drop(store); + // The restart in the same boot; its first checkpoint runs. + let store = open(home.path(), &killed); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a kill after a failed log close and before any checkpoint: the restart trusted the log's pages and lost acked lines" + ); +} diff --git a/cmux-tui/crates/acpmux/src/store/journal_store.rs b/cmux-tui/crates/acpmux/src/store/journal_store.rs index c5018874a29d..b31b98334a5a 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_store.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_store.rs @@ -20,8 +20,12 @@ //! retires the journal segments before it. A failed log sync (in it, at a log's close, or in //! `flush_session`) freezes it: a later sync may report success for pages the kernel dropped, //! so the journal keeps every line and the store asks the daemon to stop -//! ([`Store::stop_reason`]); the next start's replay writes the logs again. Ten failed -//! checkpoints in a row ask the same. +//! ([`Store::stop_reason`]). The freeze is a `frozen` record in the journal (a control record +//! that a full queue never refuses; the journal closes only after the shutdown's last +//! checkpoint): the open that replays it writes every replayed chat's lines again. A record the +//! journal still refuses (its writer stopped) leaves a durable marker, `/journal-frozen`, +//! that the open reads the same way. Ten failed checkpoints in a row ask the daemon to stop +//! too. //! - Recovery (open) repairs each log and never deletes a whole line or a folder: see //! [`Inner::repair`]. It writes every replayed chat's lines again from the point the journal //! takes over, so the first checkpoint after it syncs bytes this process wrote (a killed @@ -63,22 +67,32 @@ mod meta; mod new_folders; mod repair; -use checkpoint::{read_checkpoint, remove_checkpoint_temporaries}; +use checkpoint::{frozen_marker, read_checkpoint, remove_checkpoint_temporaries}; pub(super) use fenced::FencedStorage; #[cfg(test)] pub(super) use meta::SETTLE_HOLD; use repair::Replay; #[cfg(test)] -pub(super) use repair::STOP_AFTER_CUT; +pub(super) use repair::{FAIL_NEXT_WRITE_BACK, HELD_CAP, STOP_AFTER_CUT}; /// The journal record kinds. (Kind 3, a `cancel` of a line the log refused, was used before the -/// review of #34901 and never shipped: a refused line is never journaled now.) +/// review of #34901 and never shipped: a refused line is never journaled now. Kind 4 is `meta`, +/// `journal_store/meta.rs`; kind 5 was planned for S3 and never used.) const KIND_LOG: u16 = 1; const KIND_FORGET: u16 = 2; +/// A freeze (a failed sync of the store's files; its reason as text): the open that replays it +/// writes every replayed chat's lines again (`Inner::freeze`). +const KIND_FROZEN: u16 = 6; const JOURNAL_DIR: &str = "journal"; const CHECKPOINT_NAME: &str = "journal-projected.json"; +/// The checkpoint file's schema. const CHECKPOINT_SCHEMA: &str = "acpmux.journal-projected.v1"; +/// A durable marker of a freeze whose `frozen` record may not have reached the journal (refused, +/// or not durable when the journal closed): the next open writes every replayed chat's lines +/// again, as for the record (fail closed; delta review of #35866, D1; +/// `journal_store/checkpoint.rs`). +const FROZEN_NAME: &str = "journal-frozen"; /// Where the journaled lines of a log that recovery leaves as it is go. const UNAPPLIED_DIR: &str = "journal-unapplied"; const CHECKPOINT_EVERY: Duration = Duration::from_secs(30); @@ -148,6 +162,20 @@ thread_local! { /// the LSN and the chats to sync. pub(crate) static HOLD_AFTER_START: log_sync::HoldSlot = const { std::cell::RefCell::new(None) }; + /// Tests: the next `frozen` record a freeze on this thread appends is refused, as when the + /// journal's writer stopped (`AppendError::Closed`). + pub(crate) static FAIL_FROZEN_RECORD: std::cell::Cell = + const { std::cell::Cell::new(false) }; + /// Tests: a freeze on this thread stops right before it writes its fallback marker (off any + /// sync's count). + pub(crate) static HOLD_BEFORE_FROZEN_MARKER: log_sync::HoldSlot = + const { std::cell::RefCell::new(None) }; +} + +/// Tests: the stop of [`HOLD_BEFORE_FROZEN_MARKER`]. +#[cfg(test)] +fn hold_before_frozen_marker() { + log_sync::hold(&HOLD_BEFORE_FROZEN_MARKER); } /// Tests: the stop of [`HOLD_AFTER_START`]. @@ -283,12 +311,40 @@ struct Order { /// Each chat's newest metadata image that its `session.json` does not hold yet (S3a, /// `journal_store/meta.rs`): reads see it first, the checkpoint writes it. metas: BTreeMap, + /// The log segments whose verified bytes the open's repair wrote back over themselves: the + /// next checkpoint syncs them before it retires their records. A killed daemon's writes may + /// be in the page cache only, and a previous run's failed sync may have dropped pages that + /// still read back; the write marked the range dirty again, so the sync puts the bytes on the + /// disk whatever happened before (delta review 2 of #35866, D2). The checkpoint only syncs: + /// a file put at the path since is never written (delta review 3, D3). A delete drops its + /// chat's. + read_segments: BTreeMap, + /// How many more written-back segments the open may keep open (`repair::held_cap`). + held_left: usize, + /// The open replayed a `frozen` record (a previous run's failed sync) or found the freeze + /// marker: its repair writes every replayed chat's lines again. + rewrite_all: bool, + /// The open found the freeze marker (`FROZEN_NAME`): the first checkpoint that succeeds + /// removes it. + marker: bool, + /// Chats whose lines the open took from a repair file (`journal_store/repair.rs`): written + /// again too, as their earlier repair could not write them. + rewrite: BTreeSet, + /// An upper bound of the LSN of each chat's newest log record in this process (an append, or + /// the open's replay and repair). A chat without one has no log record past the checkpoint + /// the open started from: a retention rewrite of a chat at or below the checkpoint needs no + /// checkpoint of its own (`Store::commit_compaction`). + log_lsn: HashMap, } impl Order { /// Whether a checkpoint has anything to do. fn has_work(&self) -> bool { - self.applied > self.checkpointed || !self.dirty.is_empty() || !self.metas.is_empty() + self.applied > self.checkpointed + || !self.dirty.is_empty() + || !self.metas.is_empty() + || !self.read_segments.is_empty() + || self.marker } } @@ -309,6 +365,7 @@ impl JournalStore { remove_checkpoint_temporaries(home); let local = LocalStore::new(home.join("sessions"), config.segment_bytes)?; let from = read_checkpoint(home)?; + let marker = frozen_marker(home); let (journal, report, mut replay) = open_journal(storage, &home.join(JOURNAL_DIR), journal_config, from)?; for damage in &report.damage { @@ -343,6 +400,9 @@ impl JournalStore { order: Mutex::new(Order { checkpointed: from, damage: !report.damage.is_empty(), + rewrite_all: replay.frozen || marker, + marker, + held_left: repair::held_cap(), ..Order::default() }), checkpointing: Mutex::new(()), @@ -360,8 +420,11 @@ impl JournalStore { // The lines of chats whose repair failed at an earlier open, journaled again first. inner.take_repairs(&mut replay); inner.apply(replay)?; - if let Err(error) = inner.checkpoint() { - tracing::warn!("event journal checkpoint after the replay: {error:#}"); + // The checkpoint after the replay runs on the checkpointer's thread at once, not here: the + // daemon binds its socket after this open returns, and the checkpoint's syncs are the + // open's cost per replayed chat (6.1 s for 300 chats on a loaded Mac, fleet 59175d6b). + if inner.order.lock_or_recover().has_work() { + inner.timer.lock_or_recover().wanted = true; } let store = Self { inner, @@ -398,6 +461,19 @@ impl JournalStore { pub(super) fn opened(&self) -> &cmux_journal::OpenReport { &self.opened } + + /// Tests: the written-back segments listed for the next checkpoint (`Order::read_segments`): + /// how many the open keeps open, and how many it closed. + #[cfg(test)] + pub(super) fn written_back(&self) -> (usize, usize) { + let order = self.inner.order.lock_or_recover(); + let held = order + .read_segments + .values() + .filter(|written| matches!(written, repair::WrittenBack::Held(_))) + .count(); + (held, order.read_segments.len().saturating_sub(held)) + } } /// [`super::open`] with the journal on `storage` (a simulated file system in the crash tests) and @@ -547,6 +623,13 @@ impl Store for JournalStore { inner.acked.store(lsn, std::sync::atomic::Ordering::SeqCst); } order.applied = lsn; + // A chat's first line in this process allocates its entry; every later one reuses it. + match order.log_lsn.get_mut(id) { + Some(newest) => *newest = lsn, + None => { + order.log_lsn.insert(id.to_owned(), lsn); + } + } order.bytes = order.bytes.saturating_add(u64::try_from(payload.len()).unwrap_or(u64::MAX)); let full = order.bytes > CHECKPOINT_BYTES && !order.frozen; drop(order); @@ -646,14 +729,31 @@ impl Store for JournalStore { }), }) } - /// The logs refuse writes, and the journal fully syncs every record (the shutdown's `flush` - /// then checkpoints). + /// No write lands after this returns; then the shutdown's last checkpoint, while the journal is + /// still open; then the journal closes, every record fully synced. The last checkpoint runs + /// here and not after (delta review of #35866, D1): a log sync that fails in it, or in a sync + /// under way that it waits for, freezes the store, and its `frozen` record must reach the + /// journal. A checkpoint after this (`Hub::seal`'s flush) finds no work. fn seal(&self) { - // Under the appends' lock: an append lands in the log and the journal, or in neither. - let _order = self.inner.order.lock_or_recover(); - Store::seal(&self.inner.local); + { + // Under the appends' lock: an append lands in the log and the journal, or in neither. + let _order = self.inner.order.lock_or_recover(); + Store::seal(&self.inner.local); + } + if let Err(error) = self.inner.checkpoint() { + tracing::warn!("the event journal's last checkpoint: {error:#}"); + } + let order = self.inner.order.lock_or_recover(); if let Err(error) = self.inner.journal.seal(SEAL_WAIT) { tracing::error!("seal the event journal: {error}"); + // Fail closed: a frozen store's record may not be durable. + let frozen = order.frozen; + drop(order); + if frozen { + self.inner.keep_frozen_marker( + "the event journal did not close with its frozen record durable", + ); + } } } fn close_log(&self, id: &str) { @@ -668,17 +768,28 @@ impl Store for JournalStore { fn stage_compaction(&self, id: &str, mode: CompactMode) -> Result> { Store::stage_compaction(&self.inner.local, id, mode) } - /// A checkpoint first, so no journal record of the chat is past it: a replay never writes - /// the lines a retention rewrite cut back into its log. A frozen or failed checkpoint stops - /// the rewrite. The commit (its record before and after the renames, the renames, the folder's - /// sync after them) runs in `counted_sync`, so a failed sync in it freezes the store, as a - /// failed log sync does (#35905): a later sync of the folder may report success for the + /// No journal record of the chat may be past the checkpoint: a replay must never write the + /// lines a retention rewrite cut back into its log. A chat whose newest log record is at or + /// below it (`Order::log_lsn`; retention takes chats untouched for days) needs no checkpoint; + /// any other runs one first, and a frozen or failed checkpoint stops the rewrite. The caller + /// holds the chat's append lock (no record of it lands meanwhile), so a checkpoint in flight + /// (the one after a restart's replay takes seconds with many chats) never holds that lock for + /// an old chat. The commit (its record before and after the renames, the renames, the + /// folder's sync after them) runs in `counted_sync`, so a failed sync in it freezes the store, + /// as a failed log sync does (#35905): a later sync of the folder may report success for the /// renames a failed one dropped, so no checkpoint may go past it. The rewritten copies are /// synced earlier, in `stage_compaction`, before any rename: a failure there only fails the /// pass and changes no log. fn commit_compaction(&self, id: &str, staged: Staged) -> Result { let inner = &self.inner; - inner.checkpoint()?; + let covered = { + let order = inner.order.lock_or_recover(); + !order.frozen + && order.log_lsn.get(id).is_none_or(|newest| *newest <= order.checkpointed) + }; + if !covered { + inner.checkpoint()?; + } inner.counted_sync("a rewritten event log failed to sync", || { Store::commit_compaction(&inner.local, id, staged) }) @@ -714,6 +825,16 @@ impl Store for JournalStore { } Store::delete(&inner.local, id)?; } + // Its folder is gone: the segments the open wrote back need no sync. A handle the open + // kept for one is closed here, off the appends' lock (a close can start disk work). + let dir = inner.local.dir(id); + let gone: Vec = { + let mut order = inner.order.lock_or_recover(); + let paths: Vec = + order.read_segments.keys().filter(|path| path.starts_with(&dir)).cloned().collect(); + paths.iter().filter_map(|path| order.read_segments.remove(path)).collect() + }; + drop(gone); if let Some(path) = repair { inner.remove_repair(&path)?; inner.order.lock_or_recover().repairs.remove(id); diff --git a/cmux-tui/crates/acpmux/src/store/journal_store/checkpoint.rs b/cmux-tui/crates/acpmux/src/store/journal_store/checkpoint.rs index ea3f5e391e69..d37bfe899a68 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_store/checkpoint.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_store/checkpoint.rs @@ -30,12 +30,21 @@ pub(super) fn read_checkpoint(home: &Path) -> Result { } } -/// Removes the temporary files of checkpoints a crash interrupted (`write_checkpoint`). +/// Whether the freeze marker (`/journal-frozen`) is there: a previous run's freeze whose +/// `frozen` record may not have reached the journal. +pub(super) fn frozen_marker(home: &Path) -> bool { + std::fs::symlink_metadata(home.join(FROZEN_NAME)).is_ok() +} + +/// Removes the temporary files of checkpoints and freeze markers a crash interrupted +/// (`write_home_file`). pub(super) fn remove_checkpoint_temporaries(home: &Path) { let Ok(entries) = std::fs::read_dir(home) else { return }; - let prefix = format!("{CHECKPOINT_NAME}.tmp-"); + let prefixes = [format!("{CHECKPOINT_NAME}.tmp-"), format!("{FROZEN_NAME}.tmp-")]; for entry in entries.flatten() { - if !entry.file_name().to_string_lossy().starts_with(&prefix) { + let name = entry.file_name(); + let name = name.to_string_lossy(); + if !prefixes.iter().any(|prefix| name.starts_with(prefix.as_str())) { continue; } let path = entry.path(); @@ -48,12 +57,21 @@ pub(super) fn remove_checkpoint_temporaries(home: &Path) { } /// Writes the checkpoint whole under a temporary name, syncs it, renames it into place and syncs -/// the home folder; an error at any step (the folder's sync too) is returned. +/// the home folder; an error at any step (the folder's sync too) is returned. Its syncs are full +/// (`F_FULLFSYNC` on macOS flushes the drive's whole cache): every log, folder and `session.json` +/// the checkpoint and recovery synced in write order before it is then durable against a power +/// loss too, once per checkpoint, before any journal record is retired. fn write_checkpoint(home: &Path, lsn: Lsn) -> Result<()> { - let path = home.join(CHECKPOINT_NAME); - crate::config::refuse_lost_home(&path)?; - let tmp = home.join(format!("{CHECKPOINT_NAME}.tmp-{}", std::process::id())); let bytes = serde_json::to_vec(&Checkpoint { schema: CHECKPOINT_SCHEMA.into(), lsn })?; + write_home_file(home, CHECKPOINT_NAME, &bytes) +} + +/// Writes `/` whole under a temporary name, fully synced, renamed into place, the home +/// folder synced; an error at any step is returned. +fn write_home_file(home: &Path, name: &str, bytes: &[u8]) -> Result<()> { + let path = home.join(name); + crate::config::refuse_lost_home(&path)?; + let tmp = home.join(format!("{name}.tmp-{}", std::process::id())); let mut options = OpenOptions::new(); options.write(true).create(true).truncate(true); #[cfg(unix)] @@ -62,8 +80,8 @@ fn write_checkpoint(home: &Path, lsn: Lsn) -> Result<()> { options.mode(0o600).custom_flags(libc::O_NOFOLLOW); } let mut file = options.open(&tmp).with_context(|| format!("write {}", tmp.display()))?; - file.write_all(&bytes)?; - file.sync_all()?; + file.write_all(bytes)?; + log_sync::sync_full(&file)?; drop(file); std::fs::rename(&tmp, &path).with_context(|| format!("rename to {}", path.display()))?; sync_directory(home) @@ -92,7 +110,7 @@ impl Inner { } fn checkpoint_once(&self) -> Result<()> { - let (lsn, dirty, metas) = { + let (lsn, dirty, metas, read) = { let mut order = self.order.lock_or_recover(); anyhow::ensure!( !order.frozen, @@ -106,7 +124,15 @@ impl Inner { } // A failed checkpoint waits for the timer or another `CHECKPOINT_BYTES`. order.bytes = 0; - (order.applied, std::mem::take(&mut order.dirty), order.metas.clone()) + // The segments the open read stay listed until their sync succeeds below: a + // checkpoint that fails before it (a journal sync slower than its wait, an image + // that cannot be written) leaves them for the next one (review of #35866, P1-1). + ( + order.applied, + std::mem::take(&mut order.dirty), + order.metas.clone(), + order.read_segments.clone(), + ) }; #[cfg(test)] super::hold_after_start(); @@ -125,6 +151,19 @@ impl Inner { self.order.lock_or_recover().dirty.extend(dirty); return Err(error); } + // The segments whose verified bytes the open's repair wrote back over themselves: synced + // before their records are retired. A killed daemon's writes may be in the page cache + // only (review 3 of #34901, finding 1a), and a previous run's failed sync may have dropped + // pages that still read back, which no durable note of it may say (delta review 2 of + // #35866, D2): the open's write marked the range dirty again, so this sync puts those + // bytes on the disk either way. + if let Err(error) = self.sync_read_segments(&read) { + self.order.lock_or_recover().dirty.extend(dirty); + return Err(error); + } + // Synced: no later checkpoint needs to sync them again (the list only shrinks after the + // open, so these are exactly the ones the sync covered). + self.order.lock_or_recover().read_segments.retain(|path, _| !read.contains_key(path)); if let Err(error) = self.sync_logs(&dirty) { self.order.lock_or_recover().dirty.extend(dirty); return Err(error); @@ -158,6 +197,7 @@ impl Inner { if damage && let Err(error) = self.journal.clear_damage() { tracing::warn!("clear the event journal's damage report: {error}"); } + self.clear_frozen_marker(); self.retire(lsn); Ok(()) } @@ -168,10 +208,37 @@ impl Inner { } } - /// Syncs the logs of `dirty` (their open segment and folders) and the sessions folder (a - /// delete's removal, a new chat's folder). A chat deleted meanwhile is passed. A failed sync - /// call freezes the checkpoint; another error (a folder that does not open) fails only this - /// checkpoint, and the next one tries again. + /// Syncs, in write order, each segment the open wrote its verified bytes back into + /// (`Order::read_segments`): through the handle the open kept, else by path. The checkpoint + /// writes nothing here, so a file that a delete, a folder replace or a retention rewrite put + /// at the path since is never changed (delta review 3 of #35866, D3), and its sync is + /// harmless; a closed segment that is gone is passed. (A retention rewrite of a replayed chat + /// checkpoints first: `Order::log_lsn`.) A failed sync freezes the store, as a failed log + /// sync does; a segment that does not open fails only this checkpoint, and the next one tries + /// again. + fn sync_read_segments(&self, read: &BTreeMap) -> Result<()> { + for (path, written) in read { + let opened; + let file = match written { + repair::WrittenBack::Held(file) => file.as_ref(), + repair::WrittenBack::Closed => match repair::open_segment_to_write(path) { + Ok(file) => { + opened = file; + &opened + } + Err(error) if is_not_found(&error) => continue, + Err(error) => return Err(error.context(format!("sync {}", path.display()))), + }, + }; + if let Err(error) = log_sync::sync_segment_ordered(file) { + self.freeze(format!("an event log failed to sync: {error}")); + return Err(anyhow::Error::from(SyncFailed(error)) + .context(format!("sync {}", path.display()))); + } + } + Ok(()) + } + fn sync_logs(&self, dirty: &BTreeSet) -> Result<()> { let failed = |error: anyhow::Error| { if error.chain().any(|cause| cause.is::()) { @@ -188,7 +255,7 @@ impl Inner { } } } - sync_directory(&self.local.root).map_err(failed) + log_sync::sync_directory_ordered(&self.local.root).map_err(failed) } /// Runs `sync`, a sync of the store's files that freezes the store when it fails (a request's @@ -197,9 +264,10 @@ impl Inner { /// rewrite's commit: its record, the renames of the rewritten segments under the writers' /// lock, the folder's sync after them). So a checkpoint never writes an LSN past its failure: /// it is under way until its failure, if any, froze the store, and a checkpoint waits for the - /// syncs under way before it writes (`checkpoint_once`). Never a wait for the journal (a - /// settle's wait for its record runs before it): the checkpoint's wait is bounded by disk work - /// on the store's own files. + /// syncs under way before it writes (`checkpoint_once`). Counted: the sync, then its freeze, + /// which never waits (one lock and a copy of the `frozen` record into the journal's queue). + /// Not counted: a wait before them (a settle's wait for its record) or a fallback marker after + /// them. So a checkpoint's wait is bounded by disk work on the store's own files. pub(super) fn counted_sync( &self, what: &str, @@ -209,24 +277,124 @@ impl Inner { #[cfg(test)] super::hold_in_sync(); let result = sync(); - if let Err(error) = &result - && error.chain().any(|cause| cause.is::()) - { + let why = match &result { + Err(error) if error.chain().any(|cause| cause.is::()) => { + Some(format!("{what}: {error:#}")) + } + _ => None, + }; + let refused = why.as_deref().is_some_and(|why| { #[cfg(test)] super::hold_before_freeze(); - self.freeze(format!("{what}: {error:#}")); - } - // Ended only now: a checkpoint that waited for it then reads the store frozen. + self.freeze_now(why) + }); + // Ended once the store is frozen: a checkpoint that waited for it reads `frozen`, and does + // not wait for a fallback marker's disk writes below. drop(under_way); + if let Some(why) = why { + self.freeze_later(refused, why); + } result } - /// A log sync failed: no checkpoint until a restart, and the daemon is asked to stop. + /// A sync of the store's files failed: no checkpoint until a restart, and the daemon is asked + /// to stop. The freeze is a `frozen` record in the journal, appended under the appends' lock + /// right after `frozen` is set, so it follows every line it covers: the open that replays it + /// writes every replayed chat's lines again, as a later sync may report success for the + /// pages the failed one dropped. A record and not a file of its own (review of #35866, P2-1 + /// and P2-2): no checkpoint in flight can take it back, as a checkpoint retires only what its + /// LSN covers and that LSN was read before this record. It is a control record + /// (`Journal::append_control`): a full queue never refuses it, and it never waits (delta + /// review of #35866, D1). The group commit makes it durable as any acknowledged record; the + /// journal closes only after the shutdown's last checkpoint (`Store::seal`). Fail closed: a + /// record the journal still refuses (its writer stopped) leaves a durable marker instead + /// ([`Inner::keep_frozen_marker`]). pub(super) fn freeze(&self, why: String) { - self.order.lock_or_recover().frozen = true; + let refused = self.freeze_now(&why); + self.freeze_later(refused, why); + } + + /// A freeze's step under the appends' lock: `frozen` set, and the `frozen` record appended, + /// which never waits. True when the journal refused the record (only the first freeze of the + /// store journals one). + fn freeze_now(&self, why: &str) -> bool { + let mut order = self.order.lock_or_recover(); + if !order.frozen { + order.frozen = true; + #[cfg(test)] + let refused = super::FAIL_FROZEN_RECORD.with(|fail| fail.replace(false)); + #[cfg(not(test))] + let refused = false; + let appended = if refused { + Err(AppendError::Closed) + } else { + self.journal.append_control(KIND_FROZEN, Class::Ack, frozen_reason(why)) + }; + match appended { + Ok(lsn) => order.applied = order.applied.max(lsn), + Err(error) => { + tracing::error!("journal the freeze: {error}: a marker takes its place"); + return true; + } + } + } + false + } + + /// A freeze's step off any sync's count: the fallback marker when the record was refused, + /// then the daemon asked to stop. + fn freeze_later(&self, refused: bool, why: String) { + if refused { + self.keep_frozen_marker(&why); + } self.ask_to_stop(why); } + /// Fail closed (delta review of #35866, D1): the freeze's `frozen` record may not reach the + /// journal (refused, or not durable when the journal closed), so a durable marker + /// (`/journal-frozen`) makes the next open write every replayed chat's lines again, as + /// the record would. The first checkpoint that succeeds after that open removes it + /// ([`Inner::clear_frozen_marker`]). A marker that cannot be written either is logged: the + /// disk fails both the logs and the home folder. + pub(super) fn keep_frozen_marker(&self, why: &str) { + #[cfg(test)] + super::hold_before_frozen_marker(); + if let Err(error) = write_home_file(&self.home, FROZEN_NAME, frozen_reason(why)) { + tracing::error!( + "write the event journal's freeze marker: {error:#}: the next start may trust pages the failed sync dropped" + ); + } + } + + /// After a checkpoint that succeeded: the marker an open found goes, as the lines that open + /// wrote again are durable now. Under the appends' lock and only while the store is not + /// frozen: a freeze of this run sets `frozen` under that lock before it writes its own marker, + /// so its marker is never removed here. The folder's sync after the removal is off the lock; + /// a removal a crash takes back only makes one more open write the lines again. + fn clear_frozen_marker(&self) { + let removed = { + let mut order = self.order.lock_or_recover(); + if !order.marker || order.frozen { + return; + } + order.marker = false; + let path = self.home.join(FROZEN_NAME); + crate::config::refuse_lost_home(&path) + .and_then(|()| std::fs::remove_file(&path).map_err(anyhow::Error::from)) + }; + match removed { + Ok(()) => { + if let Err(error) = sync_directory(&self.home) { + tracing::warn!( + "sync {} after the freeze marker: {error:#}", + self.home.display() + ); + } + } + Err(error) => tracing::warn!("remove the event journal's freeze marker: {error:#}"), + } + } + pub(super) fn ask_to_stop(&self, why: String) { tracing::error!( "{why}: the event journal keeps every line; the daemon stops so that its restart writes the logs again" @@ -405,9 +573,26 @@ impl SyncsUnderWay { impl Drop for UnderWay<'_> { fn drop(&mut self) { - if std::thread::panicking() { - self.inner.freeze("a sync of the store's files panicked".into()); - } + // As a failed sync: the store frozen before the ticket ends, a fallback marker after. + const WHY: &str = "a sync of the store's files panicked"; + let panicked = std::thread::panicking(); + let refused = panicked && self.inner.freeze_now(WHY); self.inner.syncs.end(self.ticket); + if panicked { + self.inner.freeze_later(refused, WHY.into()); + } + } +} + +/// The longest reason a `frozen` record keeps: a long error chain is cut at a character +/// boundary, so the record is never refused as too large. +const FROZEN_REASON_MAX: usize = 4096; + +/// `why` as a `frozen` record's payload, at most [`FROZEN_REASON_MAX`] bytes. +fn frozen_reason(why: &str) -> &[u8] { + let mut end = why.len().min(FROZEN_REASON_MAX); + while !why.is_char_boundary(end) { + end = end.saturating_sub(1); } + why.get(..end).unwrap_or_default().as_bytes() } diff --git a/cmux-tui/crates/acpmux/src/store/journal_store/meta.rs b/cmux-tui/crates/acpmux/src/store/journal_store/meta.rs index 9c08589a6635..a14baa6d7812 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_store/meta.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_store/meta.rs @@ -265,10 +265,14 @@ fn session_json(meta: &SessionMeta, lsn: Lsn, journal: u64) -> Result> { Ok(text.into_bytes()) } -/// Writes `meta` (the image of LSN `lsn`) to `dir`'s `session.json` durably: a new file, synced, -/// renamed into place, then the folder synced. A failed sync of either is a [`SyncFailed`] -/// (review of #35483, F3): what it was to make durable may be lost, and a later sync of the same -/// file may report success without it. +/// Writes `meta` (the image of LSN `lsn`) to `dir`'s `session.json`: whole under a temporary +/// name (owner-only, never through a symlink), synced in write order, renamed into place, the +/// folder synced in write order. The checkpoint that writes it (or recovery's, which ends the +/// open) flushes the drive once afterward (`write_checkpoint`), so a power loss keeps it too: +/// one drive flush per checkpoint, not two per chat (29 ms per replayed chat on macOS before, +/// fleet ead05e82). A failed sync of either is a [`SyncFailed`] (review of #35483, F3): what it +/// was to make durable may be lost, and a later sync of the same file may report success +/// without it. #[cfg(unix)] fn write_session_json(dir: &Path, meta: &SessionMeta, lsn: Lsn, journal: u64) -> Result<()> { use std::os::unix::fs::OpenOptionsExt; @@ -288,10 +292,12 @@ fn write_session_json(dir: &Path, meta: &SessionMeta, lsn: Lsn, journal: u64) -> .with_context(|| format!("write {}", tmp.display()))?; file.write_all(&session_json(meta, lsn, journal)?) .with_context(|| format!("write {}", tmp.display()))?; - file.sync_all().map_err(SyncFailed).with_context(|| format!("sync {}", tmp.display()))?; + log_sync::sync_segment_ordered(&file) + .map_err(SyncFailed) + .with_context(|| format!("sync {}", tmp.display()))?; drop(file); std::fs::rename(&tmp, &path).with_context(|| format!("rename to {}", path.display()))?; - sync_directory(dir).with_context(|| format!("sync {}", dir.display())) + log_sync::sync_directory_ordered(dir).with_context(|| format!("sync {}", dir.display())) })(); if written.is_err() { let _ = std::fs::remove_file(&tmp); @@ -433,10 +439,11 @@ impl Inner { #[cfg(test)] hold_settle(id); // Durable first (E13: an image names only lines the journal holds). The journal syncs in - // write order without flushing the drive's cache (macOS `F_BARRIERFSYNC`); the image's - // own `sync_all` (`F_FULLFSYNC`) flushes it, so after a power loss an image never holds - // a `journalLsn` the journal lost (review of #35483, refuted item). A weaker sync of - // `session.json` must first wait for `Journal::full_durable_lsn`. + // write order (macOS `F_BARRIERFSYNC`, a barrier on the drive), and so does the image, + // after the record's sync: a power loss that keeps the image keeps the record, so an + // image never holds a `journalLsn` the journal lost (review of #35483, refuted item). A + // sync of `session.json` that orders nothing must first wait for + // `Journal::full_durable_lsn`. self.journal .wait_durable(pending.lsn, SEAL_WAIT) .map_err(|error| anyhow::anyhow!("the event journal: {error}"))?; @@ -497,11 +504,12 @@ impl Inner { Ok(()) } - /// The replay's part: each chat's newest image to its `session.json` when its folder is the - /// one the image was saved for and the file holds an older image the journal wrote, a file - /// one of its records saw, or none. The image goes aside when the folder is gone (removed: - /// never made again, [module docs](self)) or replaced, or when its `session.json` moved on - /// without the journal (F1 of the review of #35483). + /// The replay's part: each chat's newest image into the overlay (the checkpoint, on its own + /// thread, writes it to `session.json`: no sync per replayed chat before the daemon serves) + /// when its folder is the one the image was saved for and the file holds an older image the + /// journal wrote, a file one of its records saw, or none. The image goes aside when the + /// folder is gone (removed: never made again, [module docs](self)) or replaced, or when its + /// `session.json` moved on without the journal (F1 of the review of #35483). pub(super) fn apply_metas(&self, metas: BTreeMap) -> Result<()> { for (chat, replayed) in metas { let dir = self.local.dir(&chat); @@ -534,12 +542,10 @@ impl Inner { tracing::error!(session = %chat, kept = %kept.display(), "journaled metadata of a chat not written again: {why}"); continue; } - if let Err(error) = - write_session_json(&dir, &meta, replayed.lsn, self.journal.journal_id()) - { - let kept = self.keep_aside_meta(&chat, replayed.lsn, &replayed.image)?; - tracing::error!(session = %chat, kept = %kept.display(), "the chat's metadata could not be written: {error:#}"); - } + self.order + .lock_or_recover() + .metas + .insert(chat, Pending { lsn: replayed.lsn, meta, folder: replayed.folder }); } Ok(()) } diff --git a/cmux-tui/crates/acpmux/src/store/journal_store/repair.rs b/cmux-tui/crates/acpmux/src/store/journal_store/repair.rs index c3d5abde6d11..0c4590cc430e 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_store/repair.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_store/repair.rs @@ -9,12 +9,70 @@ thread_local! { /// Tests: a repair on this thread stops the open with an error right after it cut a log, as a /// crash there would. pub(crate) static STOP_AFTER_CUT: std::cell::Cell = const { std::cell::Cell::new(false) }; + /// Tests: the next write-back of a verified range on this thread fails, as a write error + /// would (not a failed sync). + pub(crate) static FAIL_NEXT_WRITE_BACK: std::cell::Cell = + const { std::cell::Cell::new(false) }; + /// Tests: the open on this thread keeps at most this many written-back segments open + /// ([`held_cap`]). + pub(crate) static HELD_CAP: std::cell::Cell> = + const { std::cell::Cell::new(None) }; +} + +/// The most written-back segments an open keeps open until its checkpoint syncs them +/// ([`held_cap`]): far above any real count of replayed chats, and far below any raised limit. +#[cfg(unix)] +const HELD_MAX: usize = 4096; + +/// [`held_cap`] when the open-file limit cannot be read. +#[cfg(unix)] +const HELD_WHEN_UNKNOWN: usize = 64; + +/// A log segment the open wrote its verified bytes back into (`Order::read_segments`): the next +/// checkpoint syncs it before it retires the records of those bytes. +#[derive(Clone, Debug)] +pub(super) enum WrittenBack { + /// Through the handle the open wrote with, kept open until that sync, after which the + /// checkpoint's thread closes it: the file the open read, wherever a delete, a folder move + /// or a retention rewrite has put it since. Within [`held_cap`] the open closes nothing it + /// wrote: on macOS a close right after the write cost the open about a quarter of a + /// millisecond per replayed chat and made the first acknowledged change after it about ten + /// times slower (the open-time A/B of delta review 3 of #35866, on three Macs). + Held(Arc), + /// By path: the open closed the file after the write (Windows, where an open file can block + /// its folder's rename or removal, and the segments past [`held_cap`]). + Closed, +} + +/// How many written-back segments the open keeps open (`WrittenBack::Held`): a quarter of the +/// daemon's open-file limit, which it raised before the store opened, at most [`HELD_MAX`]. So a +/// limit the daemon could not raise (256 under launchd) leaves room for everything else. None +/// on Windows. +#[cfg(unix)] +pub(super) fn held_cap() -> usize { + #[cfg(test)] + if let Some(cap) = HELD_CAP.with(std::cell::Cell::get) { + return cap; + } + let mut limit = libc::rlimit { rlim_cur: 0, rlim_max: 0 }; + // SAFETY: getrlimit writes the struct it is given and reads nothing else. + if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut limit) } != 0 { + return HELD_WHEN_UNKNOWN; + } + (usize::try_from(limit.rlim_cur).unwrap_or(usize::MAX) / 4).min(HELD_MAX) +} + +#[cfg(not(unix))] +pub(super) fn held_cap() -> usize { + 0 } /// The records after the checkpoint, by chat. #[derive(Default)] pub(super) struct Replay { chats: BTreeMap)>>, + /// A `frozen` record was replayed: a run after the checkpoint had a sync fail. + pub(super) frozen: bool, /// Each chat's newest metadata image after its last `forget` (S3a, `journal_store/meta.rs`). pub(super) metas: BTreeMap, } @@ -36,6 +94,7 @@ impl Replay { self.metas.remove(chat); } super::meta::KIND_META => self.take_meta(&record)?, + KIND_FROZEN => self.frozen = true, other => return Err(format!("unknown event journal record kind {other}")), } Ok(()) @@ -76,7 +135,14 @@ impl Inner { /// keeps its lines in `journal-unapplied/` and the open goes on; only a failure to keep them /// fails it. pub(super) fn apply(&self, replay: Replay) -> Result<()> { - let Replay { chats, metas } = replay; + // `frozen` was read at the open into `Order::rewrite_all`, which `repair` reads for every + // replayed chat: a replayed freeze makes the open write every chat's lines again. + let Replay { chats, metas, frozen: _ } = replay; + let replayed: Vec = chats + .iter() + .filter(|(_, lines)| !lines.is_empty()) + .map(|(chat, _)| chat.clone()) + .collect(); // The metadata, then the logs. Neither makes a folder that is gone (`meta.rs`). self.apply_metas(metas)?; for (chat, journaled) in chats { @@ -106,6 +172,12 @@ impl Inner { } let mut order = self.order.lock_or_recover(); order.applied = self.journal.last_lsn(); + // Every replayed chat has records past the checkpoint (the replay's, and the lines its + // repair journaled again): an upper bound for each. + let newest = order.applied; + for chat in replayed { + order.log_lsn.insert(chat, newest); + } Ok(()) } @@ -118,11 +190,16 @@ impl Inner { /// A line is a whole JSON line: bytes that do not parse (a torn tail, a hole of zeros, a line /// a hole ran into a later one) are not a line and go. /// - /// Every merged line is written again from the repair start, even when the log has them all: - /// a killed daemon's log writes may be in the page cache only, and after a failed sync the - /// kernel may have dropped pages it still shows and that a later sync reports as written. So - /// the first checkpoint after this open syncs bytes this process wrote before it retires their - /// records (review 2 of #34901, finding 1). The lines that only the log has from the repair + /// A log that already holds every merged line in order, with nothing torn, keeps its bytes: + /// the bytes read are written back over themselves ([`Inner::write_back`]), and the next + /// checkpoint syncs its segments before it retires their records. + /// + /// Otherwise every merged line is written again from the repair start; also when the + /// previous run froze (a failed sync: the kernel may have dropped pages it still shows and + /// that a later sync reports as written), which a replayed `frozen` record says, and for a + /// chat whose lines come from a repair file. Either way the first + /// checkpoint after this open syncs the bytes before it retires their records (review 2 of + /// #34901, finding 1). The lines that only the log has from the repair /// start are journaled and made durable first; then the log is cut there and the merged lines /// written, so a crash at any step loses no line: the next open repairs the same way. fn repair( @@ -164,6 +241,36 @@ impl Inner { } let merged: Vec<(u64, &[u8])> = merged.into_iter().collect(); let keep_merged = || merged.iter().map(|(_, line)| line.to_vec()).collect::>(); + // The log already holds every merged line, in order, with nothing torn among them: it + // keeps its bytes, written back over themselves, and the next checkpoint syncs them + // before it retires their records (a killed daemon's writes may be in the page cache + // only). After a frozen run (`rewrite_all`), every replayed chat's lines are written + // again, as are those of a chat whose earlier repair failed (its repair file's). + let rewrite = { + let order = self.order.lock_or_recover(); + order.rewrite_all || order.rewrite.contains(id) + }; + let whole = pieces.len() == merged.len() + && pieces.iter().zip(&merged).all(|(piece, (seq, line))| { + piece + .line + .as_ref() + .is_some_and(|(at, bytes)| at == seq && bytes.as_slice() == *line) + }); + if whole && !rewrite { + match self.write_back(id, &pieces) { + Ok(written) => { + self.order.lock_or_recover().read_segments.extend(written); + return Ok(Repaired::Done); + } + // A write error, not a failed sync: the lines are written again from the repair + // start below, whose cut also removes a range written back in part (delta + // review 3 of #35866). + Err(error) => { + tracing::warn!(session = %id, "write the log's verified bytes back: {error:#}; its lines are written again"); + } + } + } if let Some(start) = pieces.first() { // The lines only the log holds from the repair start: into the journal first. let only_logged: Vec<&[u8]> = merged @@ -187,6 +294,51 @@ impl Inner { Ok(Repaired::Done) } + /// Writes each segment's verified range of `id`'s log (the pieces of a whole read) back over + /// itself, and returns the segments written, for the next checkpoint to sync + /// (`Order::read_segments`). The bytes are the ones the read checked against the journal, + /// never read again: a page the kernel dropped after a previous run's failed sync may read + /// back stale. The write marks the range dirty again, so the checkpoint's sync puts these + /// bytes on the disk even when an earlier sync failed and left no note of it (delta review 2 + /// of #35866, D2). It runs here, before any other thread uses the logs, so it writes the file + /// it read; the checkpoint only syncs, never writes, so a file that a delete, a folder + /// replace or a retention rewrite put at the path since is never changed (delta review 3, + /// D3). Each segment keeps its handle open for that sync while the open's budget + /// (`Order::held_left`, [`held_cap`]) lasts, else it is closed and synced by path. + fn write_back(&self, id: &str, pieces: &[Piece]) -> Result> { + let mut written = Vec::new(); + for (index, path) in self.local.segments(id)? { + // The segment's verified range: from its first piece to its last piece's end, the + // pieces' bytes in order (contiguous: `read_pieces` reads every byte). + let mut range: Option<(u64, Vec)> = None; + for piece in pieces.iter().filter(|piece| piece.segment == index) { + let Some((_, line)) = &piece.line else { continue }; + range.get_or_insert_with(|| (piece.offset, Vec::new())).1.extend_from_slice(line); + } + let Some((offset, bytes)) = range else { continue }; + #[cfg(test)] + if FAIL_NEXT_WRITE_BACK.with(|fail| fail.replace(false)) { + anyhow::bail!("a failed write-back of {} (a test)", path.display()); + } + let file = open_segment_to_write(&path)?; + write_at(&file, &bytes, offset) + .with_context(|| format!("write back {}", path.display()))?; + #[cfg(all(test, unix))] + sync_model::rewritten(&file, offset); + let held = { + let mut order = self.order.lock_or_recover(); + let left = order.held_left; + order.held_left = left.saturating_sub(1); + left > 0 + }; + written.push(( + path, + if held { WrittenBack::Held(Arc::new(file)) } else { WrittenBack::Closed }, + )); + } + Ok(written) + } + /// Appends `lines` of chat `id` to the journal and waits until they are durable. fn journal_lines(&self, id: &str, lines: &[&[u8]]) -> Result<()> { let mut last = None; @@ -242,7 +394,7 @@ impl Inner { truncate_private(path, offset)?; } } - sync_directory(&self.local.dir(id).join("events")) + log_sync::sync_directory_ordered(&self.local.dir(id).join("events")) } fn forget_index_copy(&self, id: &str, index: u32) -> Result<()> { @@ -387,6 +539,7 @@ impl Inner { } let mut removed = false; for (path, chat, lines) in taken { + self.order.lock_or_recover().rewrite.insert(chat.clone()); replay.chats.entry(chat).or_default().extend(lines); match crate::config::refuse_lost_home(&path) .and_then(|()| std::fs::remove_file(&path).map_err(anyhow::Error::from)) @@ -533,7 +686,7 @@ fn piece_seq(line: &[u8]) -> Option { /// Syncs the folder at `path`; nothing when it is not there. fn sync_if_there(path: &Path) -> Result<()> { match std::fs::symlink_metadata(path) { - Ok(_) => sync_directory(path), + Ok(_) => log_sync::sync_directory_ordered(path), Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), Err(error) => Err(anyhow::Error::from(error).context(format!("stat {}", path.display()))), } @@ -575,6 +728,45 @@ fn truncate_private(path: &Path, len: u64) -> Result<()> { file.set_len(len).with_context(|| format!("shorten {}", path.display()))?; #[cfg(all(test, unix))] sync_model::cut(&file, len); - log_sync::sync_segment(&file, true)?; + log_sync::sync_segment_ordered(&file)?; Ok(()) } + +/// The log segment at `path` opened to write in place (the open's write-back) or to sync (the +/// checkpoint's: Windows flushes only a handle that may write): never made, never through a final +/// link, never under a lost home, and only an owner-only file of this user (the checks of +/// `config::private_state_file`, which every read of a segment makes). +#[cfg(unix)] +pub(super) fn open_segment_to_write(path: &Path) -> Result { + use std::os::unix::fs::OpenOptionsExt; + crate::config::refuse_lost_home(path)?; + let file = OpenOptions::new() + .write(true) + .custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK) + .open(path) + .with_context(|| format!("open {} to write", path.display()))?; + crate::config::validate_private_state_file(&file, path)?; + Ok(file) +} + +#[cfg(windows)] +pub(super) fn open_segment_to_write(path: &Path) -> Result { + crate::config::refuse_lost_home(path)?; + Ok(crate::owner_only::open_rw_existing(path)?) +} + +/// Writes `bytes` at `offset` of `file`, whole. +fn write_at(file: &File, bytes: &[u8], offset: u64) -> std::io::Result<()> { + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.write_all_at(bytes, offset) + } + #[cfg(not(unix))] + { + use std::io::{Seek, SeekFrom, Write}; + let mut file = file; + file.seek(SeekFrom::Start(offset))?; + file.write_all(bytes) + } +} diff --git a/cmux-tui/crates/acpmux/src/store/journal_store_tests.rs b/cmux-tui/crates/acpmux/src/store/journal_store_tests.rs index f981a11045b5..87aebcadb16d 100644 --- a/cmux-tui/crates/acpmux/src/store/journal_store_tests.rs +++ b/cmux-tui/crates/acpmux/src/store/journal_store_tests.rs @@ -810,8 +810,10 @@ fn an_os_crash_after_a_restart_from_a_kill_keeps_every_acked_line() { assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); let killed = journal.restart(Crash::Process); drop(store); - // The restart replays 2..5 and its first checkpoint runs. + // The restart replays 2..5 and its first checkpoint runs (on the checkpointer's thread after + // the open returns; `flush` waits for one here, so the crash comes after it). let store = open(home.path(), &killed); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); let crashed = killed.restart(Crash::Os); drop(store); sync_model::os_crash(&home.path().join("sessions")); @@ -842,9 +844,11 @@ fn an_os_crash_after_a_restart_from_a_failed_sync_keeps_every_acked_line() { let failed = store.flush(); FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); assert!(failed.is_err(), "the checkpoint went on past a failed log sync"); - // The daemon stops for a restart (no crash), and the restart's first checkpoint runs. + // The daemon stops for a restart (no crash), and the restart's first checkpoint runs (on the + // checkpointer's thread after the open returns; `flush` waits for one here). drop(store); let store = open(home.path(), &journal); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); let crashed = journal.restart(Crash::Os); drop(store); sync_model::os_crash(&home.path().join("sessions")); @@ -1016,6 +1020,57 @@ fn an_idle_store_does_not_wake_its_checkpointer() { ); } +/// A freeze is a `frozen` record in the journal: the next open writes every replayed chat's +/// lines again even when its log reads whole (the kernel may have dropped pages that still read +/// back), and its first checkpoint retires the record, so the open after that takes the whole log +/// as it is (no write on the opening thread). +#[test] +fn a_freeze_makes_the_next_open_write_the_lines_again_and_only_that_open() { + use super::log_sync::ALL_SYNCS; + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(48); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + store.append(CHAT, &record(2, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(true)); + let failed = store.flush(); + FAIL_NEXT_LOG_SYNC.with(|fail| fail.set(false)); + assert!(failed.is_err(), "the checkpoint went on past a failed log sync"); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + drop(store); + let log = home.path().join("sessions").join(CHAT).join("events").join("000001.ndjson"); + let before = std::fs::read(&log).unwrap_or_default(); + // The restart after the freeze writes the chat's lines again (its repair cuts and writes the + // log, with syncs on the opening thread). + let syncs = ALL_SYNCS.with(std::cell::Cell::get); + let store = open(home.path(), &journal); + let rewritten = ALL_SYNCS.with(std::cell::Cell::get).saturating_sub(syncs); + assert!(rewritten > 0, "the open after a freeze did not write the chat's lines again"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + assert_eq!(seqs(store.as_ref(), CHAT), vec![1, 2]); + assert_eq!( + std::fs::read(&log).unwrap_or_default(), + before, + "the rewrite changed the log's bytes" + ); + // A kill after that checkpoint (which retired the record): the next open trusts the log. + store.append(CHAT, &record(3, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + let syncs = ALL_SYNCS.with(std::cell::Cell::get); + let store = open(home.path(), &killed); + let again = ALL_SYNCS.with(std::cell::Cell::get).saturating_sub(syncs); + assert_eq!(seqs(store.as_ref(), CHAT), vec![1, 2, 3]); + assert!( + again < rewritten, + "an open long after a freeze still wrote the lines again ({again} syncs on its thread, {rewritten} at the open after the freeze)" + ); +} + /// The idle checkpointer (follow-up 4) and its period. A change whose `touched` takes the idle /// flag while the checkpointer holds the timer lock (between marking itself idle and its second /// look) notifies once it gets that lock. The second look saw the change, so the checkpointer is @@ -1692,3 +1747,147 @@ fn a_checkpoint_never_goes_past_a_log_close_whose_sync_is_failing_on_another_thr ); assert!(waited, "the checkpoint did not wait for the close's note"); } + +// ---------------------------------------------------------------- the open's checkpoint in flight + +/// A record whose long text a retention rewrite cuts (`store/compact.rs`). +fn long_line(seq: u64) -> EventRecord { + EventRecord { + seq, + at: 1_000 + seq, + dir: "in".into(), + kind: "claude.assistant".into(), + msg: json!({ "text": "x".repeat(8 * 1024) }), + host_seq: None, + } +} + +/// The names in `journal-unapplied/` that start with `id`. +fn kept_aside(home: &Path, id: &str) -> Vec { + std::fs::read_dir(home.join("journal-unapplied")) + .map(|entries| { + entries + .flatten() + .map(|entry| entry.file_name().to_string_lossy().into_owned()) + .filter(|name| name.starts_with(id)) + .collect() + }) + .unwrap_or_default() +} + +/// A retention rewrite of a chat whose records are all checkpointed (retention takes a chat +/// untouched for days) runs no checkpoint of its own. Its caller holds the chat's append lock for +/// the commit (`hub/retention.rs`), so a checkpoint in flight, such as the one the checkpointer +/// runs right after a restart's replay (seconds with many chats), must not hold that lock: a +/// person's first message to that chat would wait for it. +#[test] +fn a_retention_rewrite_of_a_checkpointed_chat_does_not_wait_for_a_checkpoint() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(83); + let store: Arc = Arc::from(open(home.path(), &journal)); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + store.append(CHAT, &long_line(2)).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let staged = store + .stage_compaction(CHAT, CompactMode::Write { archive: None }) + .unwrap_or_else(|e| panic!("{e:#}")) + .unwrap_or_else(|| panic!("nothing to compact")); + // A checkpoint in flight: chat-b's record waits in a 2 s sync, and the checkpoint for it. + journal.set_sync_delay(Duration::from_secs(2)); + store.append("chat-b", &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + let flusher = { + let store = Arc::clone(&store); + std::thread::spawn(move || store.flush()) + }; + std::thread::sleep(Duration::from_millis(300)); + let started = std::time::Instant::now(); + let committed = store.commit_compaction(CHAT, staged).unwrap_or_else(|e| panic!("{e:#}")); + let took = started.elapsed(); + journal.set_sync_delay(Duration::ZERO); + flusher + .join() + .unwrap_or_else(|_| panic!("the flush panicked")) + .unwrap_or_else(|e| panic!("{e:#}")); + assert!(committed, "the rewrite was refused"); + assert!( + took < Duration::from_millis(500), + "the retention rewrite of a checkpointed chat waited {took:?} for a checkpoint in flight" + ); +} + +/// The other way: a retention rewrite of a chat with a record past the checkpoint checkpoints +/// first, so a kill right after it replays no line the rewrite cut, and keeps no copy aside. +#[test] +fn a_retention_rewrite_of_a_chat_with_records_past_the_checkpoint_checkpoints_first() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(84); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + store.append(CHAT, &long_line(2)).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let staged = store + .stage_compaction(CHAT, CompactMode::Write { archive: None }) + .unwrap_or_else(|e| panic!("{e:#}")) + .unwrap_or_else(|| panic!("nothing to compact")); + assert!(store.commit_compaction(CHAT, staged).unwrap_or_else(|e| panic!("{e:#}"))); + let log = home.path().join("sessions").join(CHAT).join("events/000001.ndjson"); + let cut = std::fs::read(&log).unwrap_or_else(|e| panic!("{e}")); + let killed = journal.restart(Crash::Process); + drop(store); + let store = open(home.path(), &killed); + assert_eq!(seqs(store.as_ref(), CHAT), vec![1, 2]); + assert!( + std::fs::read(&log).unwrap_or_else(|e| panic!("{e}")) == cut, + "a replay wrote lines a retention rewrite cut back into the log" + ); + assert!( + kept_aside(home.path(), CHAT).is_empty(), + "a replay kept a rewritten chat's journaled lines aside: {:?}", + kept_aside(home.path(), CHAT) + ); +} + +/// The same after a restart: a chat the replay wrote has records past the checkpoint until the +/// checkpoint after the open, so its retention rewrite checkpoints first too. (A guard of the +/// outcome: the checkpointer may finish the open's checkpoint before the rewrite, and then the +/// rewrite needs none.) +#[test] +fn a_retention_rewrite_of_a_replayed_chat_checkpoints_first() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(85); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + store.append(CHAT, &long_line(2)).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + // The checkpoint after the open waits in a 2 s sync of chat-b's line: the replayed records + // stay past the checkpoint while the rewrite runs. + let store = open(home.path(), &killed); + killed.set_sync_delay(Duration::from_secs(2)); + store.append("chat-b", &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + let staged = store + .stage_compaction(CHAT, CompactMode::Write { archive: None }) + .unwrap_or_else(|e| panic!("{e:#}")) + .unwrap_or_else(|| panic!("nothing to compact")); + assert!(store.commit_compaction(CHAT, staged).unwrap_or_else(|e| panic!("{e:#}"))); + killed.set_sync_delay(Duration::ZERO); + let log = home.path().join("sessions").join(CHAT).join("events/000001.ndjson"); + let cut = std::fs::read(&log).unwrap_or_else(|e| panic!("{e}")); + let again = killed.restart(Crash::Process); + drop(store); + let store = open(home.path(), &again); + assert_eq!(seqs(store.as_ref(), CHAT), vec![1, 2]); + assert!( + std::fs::read(&log).unwrap_or_else(|e| panic!("{e}")) == cut, + "a replay wrote lines a retention rewrite cut back into the log" + ); + assert!( + kept_aside(home.path(), CHAT).is_empty(), + "a replay kept a rewritten chat's journaled lines aside: {:?}", + kept_aside(home.path(), CHAT) + ); +} diff --git a/cmux-tui/crates/acpmux/src/store/journal_write_back_tests.rs b/cmux-tui/crates/acpmux/src/store/journal_write_back_tests.rs new file mode 100644 index 000000000000..44d6e28abe4a --- /dev/null +++ b/cmux-tui/crates/acpmux/src/store/journal_write_back_tests.rs @@ -0,0 +1,336 @@ +//! The write-back of the log ranges the open's repair verified (delta reviews 2 and 3 of #35866, +//! D2 and D3): the verified bytes go back into the file the open read, and never into another +//! file that a delete or a folder replace put at the same path since. An inode number is not a +//! file's identity: a file system that hands a freed inode out again (ext4, XFS) gives a new file +//! at the same path the old number, and Windows has no number to compare. +//! +//! The journal runs on `cmux_journal::SimStorage`; the chat logs are real files. + +use super::journal_store::{FAIL_NEXT_WRITE_BACK, HELD_CAP, JournalStore}; +use super::sync_model; +use super::*; +use cmux_journal::{Crash, SimStorage}; +use serde_json::json; +use std::os::unix::fs::{DirBuilderExt, MetadataExt, PermissionsExt}; +use std::sync::Arc; +use std::time::Duration; + +const CHAT: &str = "chat-a"; +const WAIT: Duration = Duration::from_secs(10); + +fn config() -> StoreConfig { + StoreConfig { journal: true, ..StoreConfig::default() } +} + +fn journal_config() -> cmux_journal::Config { + cmux_journal::Config { + retry_min: Duration::from_millis(1), + retry_max: Duration::from_millis(5), + ..cmux_journal::Config::default() + } +} + +fn open(home: &Path, journal: &SimStorage) -> Box { + open_with_journal(&config(), home, Arc::new(journal.clone()), journal_config()) + .unwrap_or_else(|e| panic!("open the store: {e:#}")) +} + +fn record(seq: u64, kind: &str) -> EventRecord { + EventRecord { + seq, + at: 1_000 + seq, + dir: "mux".into(), + kind: kind.into(), + msg: json!({ "n": seq }), + host_seq: None, + } +} + +fn meta(id: &str, name: &str) -> SessionMeta { + serde_json::from_value(json!({ + "schema": META_SCHEMA, + "id": id, + "name": name, + "harness": "claude", + "family": "claude", + "cwd": "/tmp", + "status": "idle", + "createdAt": 1, + "updatedAt": 1, + "lastSeq": 0, + })) + .unwrap_or_else(|e| panic!("a test meta: {e}")) +} + +fn seqs(store: &dyn Store, id: &str) -> Vec { + let events = store.events(id, 0, 10_000).unwrap_or_else(|e| panic!("read {id}: {e:#}")); + events.iter().map(|record| record.seq).collect() +} + +fn set_mode(path: &Path, mode: u32) { + std::fs::set_permissions(path, std::fs::Permissions::from_mode(mode)) + .unwrap_or_else(|e| panic!("chmod {}: {e}", path.display())); +} + +/// Gives a folder its mode 0o700 back when dropped, also when the test panics (so the temporary +/// home can be removed). +struct Writable(PathBuf); + +impl Drop for Writable { + fn drop(&mut self) { + let _ = std::fs::set_permissions(&self.0, std::fs::Permissions::from_mode(0o700)); + } +} + +/// Whether this user can still make a file in a folder of mode 0o500 (root can). +fn mode_stops_writes(folder: &Path) -> bool { + set_mode(folder, 0o500); + let probe = folder.join("write-back-probe"); + let made = std::fs::File::create(&probe).is_ok(); + let _ = std::fs::remove_file(&probe); + set_mode(folder, 0o700); + !made +} + +/// chat-a replayed after a kill: its log holds lines 1 to 5, and lines 2 to 5 are past the +/// checkpoint, so the open's repair reads them back, finds them equal to the journal's and keeps +/// the log as it is (`Order::read_segments`). Every checkpoint is then held off before it reaches +/// the logs: chat-b's folder takes no new file, so its replayed metadata image cannot be written +/// and the checkpoint fails in `project_metas` (no freeze: the segments stay listed). Returns the +/// store, chat-b's folder guard, chat-a's first segment and that segment's file id. +fn replayed_chat_with_its_checkpoints_held( + home: &Path, + seed: u64, +) -> (Box, Writable, PathBuf, u64) { + let journal = SimStorage::new(seed); + let store = open(home, &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + store.save(&meta("chat-b", "renamed")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + let folder_b = home.join("sessions").join("chat-b"); + set_mode(&folder_b, 0o500); + let guard = Writable(folder_b); + let store = open(home, &killed); + let held = store.flush(); + assert!(held.is_err(), "the checkpoint went on although chat-b's image cannot be written"); + let segment = home.join("sessions").join(CHAT).join("events").join("000001.ndjson"); + let ino = std::fs::metadata(&segment).unwrap_or_else(|e| panic!("{e}")).ino(); + (store, guard, segment, ino) +} + +/// Makes chat-a's folder and its `events/` again (owner-only), as a new chat or an import does. +fn make_events_folder(home: &Path) { + let events = home.join("sessions").join(CHAT).join("events"); + std::fs::DirBuilder::new() + .mode(0o700) + .recursive(true) + .create(&events) + .unwrap_or_else(|e| panic!("make {}: {e}", events.display())); +} + +/// The test's message when the checkpoint changed a log it must not write. +fn changed(why: &str, before: &[u8], after: &[u8]) -> String { + format!( + "the checkpoint wrote the old chat's verified bytes into {why}:\nbefore: {}\nafter: {}", + String::from_utf8_lossy(before), + String::from_utf8_lossy(after) + ) +} + +/// Delta review 3 of #35866, D3 (the reviewer's test, `r-durability-rsync-reviewer` at +/// `7f3e036d277`; its repro +/// `rsync_review5_a_deleted_chats_write_back_never_lands_in_a_new_chats_log`, red on macOS +/// fleet 857c0030 and Linux fleet 878ccf7a). chat-a is deleted while the open's +/// write-back of its log is pending, and made again under the same id (the trigger bridge's fixed +/// ids allow it once the folder is gone). The new chat's first segment gets the old file id: a +/// stand-in for a reused inode, the old file kept by a second link and linked back, emptied +/// (ext4 handed the freed inode out again in the reviewer's Linux repro). The checkpoint must +/// not write the deleted chat's bytes over the new chat's lines. +#[test] +fn a_chat_deleted_and_made_again_never_takes_the_old_chats_verified_bytes() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + assert!( + mode_stops_writes(home.path()), + "this test needs a user that a folder's mode stops (not root)" + ); + let (store, guard, segment, ino) = replayed_chat_with_its_checkpoints_held(home.path(), 111); + let kept = home.path().join("kept-old-segment"); + std::fs::hard_link(&segment, &kept).unwrap_or_else(|e| panic!("{e}")); + store.delete(CHAT).unwrap_or_else(|e| panic!("{e:#}")); + make_events_folder(home.path()); + std::fs::hard_link(&kept, &segment).unwrap_or_else(|e| panic!("{e}")); + std::fs::remove_file(&kept).unwrap_or_else(|e| panic!("{e}")); + std::fs::OpenOptions::new() + .write(true) + .open(&segment) + .and_then(|file| file.set_len(0)) + .unwrap_or_else(|e| panic!("{e}")); + assert_eq!(std::fs::metadata(&segment).unwrap_or_else(|e| panic!("{e}")).ino(), ino); + for seq in 1..=8 { + store.append(CHAT, &record(seq, "made-again")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let before = std::fs::read(&segment).unwrap_or_else(|e| panic!("{e}")); + // chat-b's folder takes files again: the next checkpoint goes on. + drop(guard); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let after = std::fs::read(&segment).unwrap_or_else(|e| panic!("{e}")); + assert!( + after == before, + "{}", + changed( + "a new chat's log (a delete, then another file with the same file id)", + &before, + &after + ) + ); +} + +/// D3 with no delete: code outside the store replaces chat-a's folder, as a transfer-in that +/// replaces a durable chat does (`hub/transfer_in.rs`: the old folder is renamed aside, the +/// import puts a new one in its place, and the old one is removed). The new folder's first +/// segment has the old segment's file id (the old file renamed in, a stand-in for a reused +/// inode) and the new chat's own lines. The store never hears of the replace, so only a +/// write-back that never runs after the open keeps the checkpoint from writing the old chat's +/// bytes over these lines. +#[test] +fn a_folder_replaced_outside_the_store_never_takes_the_old_chats_verified_bytes() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + assert!( + mode_stops_writes(home.path()), + "this test needs a user that a folder's mode stops (not root)" + ); + let (store, guard, segment, ino) = replayed_chat_with_its_checkpoints_held(home.path(), 113); + let sessions = home.path().join("sessions"); + let aside = sessions.join(format!(".{CHAT}.replacement-backup")); + std::fs::rename(sessions.join(CHAT), &aside).unwrap_or_else(|e| panic!("{e}")); + make_events_folder(home.path()); + std::fs::rename(aside.join("events").join("000001.ndjson"), &segment) + .unwrap_or_else(|e| panic!("{e}")); + let mut before = Vec::new(); + for seq in 1..=8 { + let line = + serde_json::to_vec(&record(seq, "transferred-in")).unwrap_or_else(|e| panic!("{e}")); + before.extend_from_slice(&line); + before.push(b'\n'); + } + std::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(&segment) + .and_then(|mut file| file.write_all(&before)) + .unwrap_or_else(|e| panic!("{e}")); + std::fs::remove_dir_all(&aside).unwrap_or_else(|e| panic!("{e}")); + assert_eq!(std::fs::metadata(&segment).unwrap_or_else(|e| panic!("{e}")).ino(), ino); + drop(guard); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let after = std::fs::read(&segment).unwrap_or_else(|e| panic!("{e}")); + assert!( + after == before, + "{}", + changed("a folder put in its place outside the store (the same file id)", &before, &after) + ); +} + +/// The open's write-back fails (a write error, not a failed sync; the reviewer's answer in delta +/// review 3 of #35866): the open neither fails nor freezes the store. The chat's log is written +/// again from the repair start, as for a log that does not read whole, and the first checkpoint +/// syncs it: after an OS crash, every acknowledged line is there. +#[test] +fn a_failed_write_back_at_the_open_writes_the_lines_again() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + let journal = SimStorage::new(114); + let store = open(home.path(), &journal); + store.append(CHAT, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for seq in 2..=5 { + store.append(CHAT, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + FAIL_NEXT_WRITE_BACK.with(|fail| fail.set(true)); + let store = open(home.path(), &killed); + let unused = FAIL_NEXT_WRITE_BACK.with(|fail| fail.replace(false)); + assert!(!unused, "the open never wrote a verified range back (no fast path taken)"); + store.flush().unwrap_or_else(|e| panic!("the checkpoint after a failed write-back: {e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + assert_eq!( + seqs(store.as_ref(), CHAT), + (1..=5).collect::>(), + "a failed write-back at the open: the lines were not written again and synced" + ); +} + +/// The open keeps at most its cap of written-back segments open for the checkpoint's sync +/// (`repair::held_cap`, a quarter of the open-file limit) and closes the rest, which the +/// checkpoint syncs by path: a limit the daemon could not raise is never used up by them. Either +/// way, after the checkpoint and an OS crash, every acknowledged line is there. +#[test] +fn the_written_back_segments_past_the_cap_are_synced_by_path() { + let home = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}")); + assert!( + mode_stops_writes(home.path()), + "this test needs a user that a folder's mode stops (not root)" + ); + let chats = ["chat-a", "chat-c", "chat-d"]; + let journal = SimStorage::new(115); + let store = open(home.path(), &journal); + for chat in chats { + store.append(chat, &record(1, "created")).unwrap_or_else(|e| panic!("{e:#}")); + } + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + for chat in chats { + for seq in 2..=5 { + store.append(chat, &record(seq, "queued")).unwrap_or_else(|e| panic!("{e:#}")); + } + } + store.save(&meta("chat-b", "renamed")).unwrap_or_else(|e| panic!("{e:#}")); + assert!(store.wait_durable(WAIT).unwrap_or_else(|e| panic!("{e:#}")), "never durable"); + let killed = journal.restart(Crash::Process); + drop(store); + // chat-b's folder takes no new file: every checkpoint fails before it syncs the segments + // the open wrote back, so the list stays as the open made it. + let folder_b = home.path().join("sessions").join("chat-b"); + set_mode(&folder_b, 0o500); + let guard = Writable(folder_b); + HELD_CAP.with(|cap| cap.set(Some(1))); + let opened = + JournalStore::open(&config(), home.path(), Arc::new(killed.clone()), journal_config()); + HELD_CAP.with(|cap| cap.set(None)); + let store = opened.unwrap_or_else(|e| panic!("open the store: {e:#}")); + assert!( + store.flush().is_err(), + "the checkpoint went on although chat-b's image cannot be written" + ); + assert_eq!( + store.written_back(), + (1, 2), + "the open kept a written-back segment open past its cap, or closed one within it" + ); + drop(guard); + store.flush().unwrap_or_else(|e| panic!("{e:#}")); + let crashed = killed.restart(Crash::Os); + drop(store); + sync_model::os_crash(&home.path().join("sessions")); + let store = open(home.path(), &crashed); + for chat in chats { + assert_eq!( + seqs(store.as_ref(), chat), + (1..=5).collect::>(), + "{chat}: a written-back segment was not synced before its records were retired" + ); + } +} diff --git a/cmux-tui/crates/acpmux/src/store/log_sync.rs b/cmux-tui/crates/acpmux/src/store/log_sync.rs index 80612815cd75..61acfcc8b393 100644 --- a/cmux-tui/crates/acpmux/src/store/log_sync.rs +++ b/cmux-tui/crates/acpmux/src/store/log_sync.rs @@ -82,6 +82,24 @@ pub(super) fn fail_dir_sync(path: &Path) -> bool { }) } +/// Tests: the syncs that flush the drive's cache (on macOS every `sync_all`/`sync_data` is an +/// `F_FULLFSYNC`), counted. +#[cfg(test)] +pub(crate) static FULL_SYNCS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); +#[cfg(test)] +thread_local! { + /// Tests: every sync of a log segment, a folder or a store file, full or ordered, that this + /// thread made (the open's own syncs, not those of the checkpointer's thread). + pub(crate) static ALL_SYNCS: std::cell::Cell = const { std::cell::Cell::new(0) }; +} + +#[cfg(test)] +fn counted(result: &std::io::Result<()>) { + if result.is_ok() { + ALL_SYNCS.with(|syncs| syncs.set(syncs.get().saturating_add(1))); + } +} + /// Syncs one log segment: its data, or with `all` its metadata too. The tests' crash model /// follows each sync (`store/sync_model.rs`). pub(super) fn sync_segment(file: &File, all: bool) -> std::io::Result<()> { @@ -92,11 +110,76 @@ pub(super) fn sync_segment(file: &File, all: bool) -> std::io::Result<()> { return Err(std::io::Error::other("a failed sync (a test)")); } let result = if all { file.sync_all() } else { file.sync_data() }; + #[cfg(test)] + counted(&result); + #[cfg(test)] + if result.is_ok() { + FULL_SYNCS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + #[cfg(all(test, unix))] + super::sync_model::synced(file, result.is_ok()); + result +} + +/// Syncs one log segment so that it survives a crash of any process and a kernel panic, in +/// write order with every other sync (macOS `F_BARRIERFSYNC`, which does not flush the drive's +/// cache; Linux `fdatasync`): the journal store's checkpoint and recovery, which then flush the +/// drive once with the checkpoint file ([`sync_full`]) before they retire any journal record. +/// The same test seams as [`sync_segment`]. +pub(super) fn sync_segment_ordered(file: &File) -> std::io::Result<()> { + #[cfg(test)] + if FAIL_NEXT_LOG_SYNC.with(|fail| fail.replace(false)) { + #[cfg(unix)] + super::sync_model::synced(file, false); + return Err(std::io::Error::other("a failed sync (a test)")); + } + let result = cmux_journal::sync_file(file, cmux_journal::SyncLevel::Barrier); + #[cfg(test)] + counted(&result); #[cfg(all(test, unix))] super::sync_model::synced(file, result.is_ok()); result } +/// [`sync_segment_ordered`] for a folder's entries; a failed sync is a [`SyncFailed`]. +#[cfg(unix)] +pub(super) fn sync_directory_ordered(path: &Path) -> Result<()> { + use std::os::unix::fs::OpenOptionsExt; + #[cfg(test)] + if fail_dir_sync(path) { + return Err(SyncFailed(std::io::Error::other("a failed folder sync (a test)")).into()); + } + let dir = OpenOptions::new() + .read(true) + .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW) + .open(path)?; + let result = cmux_journal::sync_file(&dir, cmux_journal::SyncLevel::Barrier); + #[cfg(test)] + counted(&result); + result.map_err(SyncFailed)?; + #[cfg(test)] + super::sync_model::folder_synced(path); + Ok(()) +} + +/// Windows has no folder sync: NTFS journals its folder changes. +#[cfg(not(unix))] +pub(super) fn sync_directory_ordered(_path: &Path) -> Result<()> { + Ok(()) +} + +/// `sync_all` of a file or folder (the drive's cache flushed on macOS), counted in tests. +pub(super) fn sync_full(file: &File) -> std::io::Result<()> { + let result = file.sync_all(); + #[cfg(test)] + counted(&result); + #[cfg(test)] + if result.is_ok() { + FULL_SYNCS.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + result +} + /// A sync call (`fsync`) that failed, as opposed to an error before it (a folder that does not /// open): what it was to make durable may be lost, and a later sync of the same file may report /// success without it (the journal store then keeps every line until a restart). @@ -126,8 +209,9 @@ impl LocalStore { self.sync_failed.load(std::sync::atomic::Ordering::SeqCst) } - /// Syncs `id`'s open log segment and its folders, without holding the writers' lock while - /// the disk works (the journal store's checkpoint, `store/journal_store.rs`). A segment that + /// Syncs `id`'s open log segment and its folders in write order ([`sync_segment_ordered`]), + /// without holding the writers' lock while the disk works (the journal store's checkpoint, + /// `store/journal_store.rs`, which flushes the drive once after every chat's). A segment that /// is not open was synced at its close (`SegmentWriter::close`, which notes a failure). pub(super) fn sync_log(&self, id: &str) -> Result<()> { // Debug builds only: `ACPMUX_TEST_FAIL_LOG_SYNC` fails every log sync, as a disk whose @@ -150,10 +234,10 @@ impl LocalStore { if let Some(file) = open { #[cfg(test)] hold(&HOLD_AFTER_LOG_CLONE); - sync_segment(&file, true).map_err(SyncFailed)?; + sync_segment_ordered(&file).map_err(SyncFailed)?; } let session = self.dir(id); - sync_directory(&session.join("events"))?; - sync_directory(&session) + sync_directory_ordered(&session.join("events"))?; + sync_directory_ordered(&session) } } diff --git a/cmux-tui/crates/acpmux/src/store/sync_model.rs b/cmux-tui/crates/acpmux/src/store/sync_model.rs index 37c63e532d79..945c32073e30 100644 --- a/cmux-tui/crates/acpmux/src/store/sync_model.rs +++ b/cmux-tui/crates/acpmux/src/store/sync_model.rs @@ -84,6 +84,18 @@ pub(crate) fn synced(file: &File, ok: bool) { }); } +/// The bytes of the log segment `file` from `offset` on were written again (the open's +/// write-back of a range it verified; the checkpoint's sync of the same handle follows): a loss +/// from at or after `offset` ends, as for a cut, since those pages are dirty again and the next +/// sync writes them. +pub(crate) fn rewritten(file: &File, offset: u64) { + update(file, |kept, _| { + if kept.lost_from.is_some_and(|from| from >= offset) { + kept.lost_from = None; + } + }); +} + /// The log segment `file` was cut to `len` (its sync follows). pub(crate) fn cut(file: &File, len: u64) { update(file, |kept, _| { diff --git a/cmux-tui/crates/cmux-journal/src/cmux_journal_tests.rs b/cmux-tui/crates/cmux-journal/src/cmux_journal_tests.rs index 4bd1272effde..05469a3f10e1 100644 --- a/cmux-tui/crates/cmux-journal/src/cmux_journal_tests.rs +++ b/cmux-tui/crates/cmux-journal/src/cmux_journal_tests.rs @@ -273,6 +273,31 @@ fn check_append_says_what_append_would_do() { assert_eq!(journal.check_append(1), Err(AppendError::Closed)); } +/// A control record (`append_control`) is taken while the queue is full, and becomes durable as +/// any record; a sealed journal refuses it too. +#[test] +fn a_control_record_is_taken_while_the_queue_is_full() { + let storage = SimStorage::new(37); + let config = Config { max_pending_bytes: 4096, ..quick() }; + let (journal, _, _) = open_at(&storage, config, 0).unwrap_or_else(|e| panic!("{e}")); + storage.set_sync_delay(Duration::from_millis(500)); + let big = vec![b'a'; 1000]; + let full = (0..50).any(|_| journal.append(1, Class::Ack, &big) == Err(AppendError::Backlog)); + assert!(full, "the pending cap never refused an append"); + let control = journal + .append_control(2, Class::Ack, b"frozen") + .unwrap_or_else(|e| panic!("a full queue refused a control record: {e}")); + assert_eq!(journal.append(1, Class::Ack, &big), Err(AppendError::Backlog)); + storage.set_sync_delay(Duration::ZERO); + journal.wait_durable(control, WAIT).unwrap_or_else(|e| panic!("{e}")); + assert!(journal.seal(WAIT).is_ok()); + assert_eq!(journal.append_control(2, Class::Ack, b"frozen"), Err(AppendError::Closed)); + assert_eq!( + journal.append_control(2, Class::Ack, &vec![0u8; crate::MAX_PAYLOAD_LEN + 1]), + Err(AppendError::TooLarge) + ); +} + #[test] fn a_sealed_journal_refuses_appends() { let storage = SimStorage::new(10); diff --git a/cmux-tui/crates/cmux-journal/src/journal.rs b/cmux-tui/crates/cmux-journal/src/journal.rs index f26c8bb944c5..52cb8421d6b3 100644 --- a/cmux-tui/crates/cmux-journal/src/journal.rs +++ b/cmux-tui/crates/cmux-journal/src/journal.rs @@ -76,13 +76,14 @@ impl Shared { impl State { /// Whether a record of `size` bytes (header included) may join the queue now: refused when - /// the journal is closed, or when the queue holds records and would grow past - /// [`Config::max_pending_bytes`] (one record is always taken, however large). - fn admits(&self, config: &Config, size: usize) -> Result<(), AppendError> { + /// the journal is closed, or, for a `limited` record, when the queue holds records and would + /// grow past [`Config::max_pending_bytes`] (one record is always taken, however large). + fn admits(&self, config: &Config, size: usize, limited: bool) -> Result<(), AppendError> { if self.closed { return Err(AppendError::Closed); } - if !self.pending.is_empty() + if limited + && !self.pending.is_empty() && self.pending.len().saturating_add(size) > config.max_pending_bytes { return Err(AppendError::Backlog); @@ -264,6 +265,29 @@ impl Journal { /// Appends one record and returns its LSN. A memory copy: no I/O, no wait. The record is /// durable once [`Journal::durable_lsn`] reaches the LSN. pub fn append(&self, kind: u16, class: Class, payload: &[u8]) -> Result { + self.append_admitted(kind, class, payload, true) + } + + /// Appends one control record as [`Journal::append`] does, except that a full queue never + /// refuses it ([`AppendError::Backlog`]): for a record its caller must not lose to a slow or + /// failing disk, and appends a bounded number of times (acpmux's `frozen` record, once per + /// store). Still refused when the journal is closed or the payload is too large. + pub fn append_control( + &self, + kind: u16, + class: Class, + payload: &[u8], + ) -> Result { + self.append_admitted(kind, class, payload, false) + } + + fn append_admitted( + &self, + kind: u16, + class: Class, + payload: &[u8], + limited: bool, + ) -> Result { if payload.len() > MAX_PAYLOAD_LEN { return Err(AppendError::TooLarge); } @@ -271,7 +295,7 @@ impl Journal { let crc = format::payload_crc(payload); let mut state = self.shared.lock(); let size = format::record_len(payload.len()); - state.admits(&self.shared.config, size)?; + state.admits(&self.shared.config, size, limited)?; let lsn = state.stats.next_lsn; state.stats.next_lsn = lsn.saturating_add(1); let was_empty = state.pending.is_empty(); @@ -294,7 +318,7 @@ impl Journal { if payload_len > MAX_PAYLOAD_LEN { return Err(AppendError::TooLarge); } - self.shared.lock().admits(&self.shared.config, format::record_len(payload_len)) + self.shared.lock().admits(&self.shared.config, format::record_len(payload_len), true) } /// This journal's id (never 0). LSNs are unique only within one id: a journal made again from diff --git a/cmux-tui/crates/cmux-journal/src/lib.rs b/cmux-tui/crates/cmux-journal/src/lib.rs index 9426b7b07c75..7829b5e7b3f8 100644 --- a/cmux-tui/crates/cmux-journal/src/lib.rs +++ b/cmux-tui/crates/cmux-journal/src/lib.rs @@ -71,7 +71,7 @@ pub use journal::{AppendError, Journal, JournalStats, OpenError, SealError, Wait pub use recovery::{Damage, OpenReport}; #[cfg(any(test, feature = "sim"))] pub use sim::{Crash, SIM_PAGE, SimRng, SimStorage}; -pub use storage::{FsStorage, OpenMode, Storage, StorageFile, StorageLock, SyncLevel}; +pub use storage::{FsStorage, OpenMode, Storage, StorageFile, StorageLock, SyncLevel, sync_file}; use std::time::Duration; diff --git a/cmux-tui/crates/cmux-journal/src/sim.rs b/cmux-tui/crates/cmux-journal/src/sim.rs index 6d09b396983c..28288fd1cedd 100644 --- a/cmux-tui/crates/cmux-journal/src/sim.rs +++ b/cmux-tui/crates/cmux-journal/src/sim.rs @@ -124,6 +124,8 @@ struct Fs { crash_after_failure: Option, /// Each read evicts the dropped pages it read (`SimStorage::evict_dropped_pages_after_read`). evict_after_read: bool, + /// The next sync stops once under way (`SimStorage::hold_next_sync`). + sync_hold: Option<(std::sync::mpsc::Sender<()>, std::sync::mpsc::Receiver<()>)>, } /// What an operation does: run, fail once, or crash the process. @@ -150,6 +152,7 @@ impl Fs { fail_sync: None, crash_after_failure: None, evict_after_read: false, + sync_hold: None, } } @@ -199,8 +202,18 @@ impl Fs { } } -/// Sleeps the sync delay with the lock released; `None` when the process crashed meanwhile. +/// Sleeps the sync delay with the lock released, after a hold if one was asked for +/// (`SimStorage::hold_next_sync`); `None` when the process crashed meanwhile. fn slow_sync<'a>(mut guard: MutexGuard<'a, Fs>, fs: &'a Mutex) -> Option> { + if let Some((arrived, release)) = guard.sync_hold.take() { + drop(guard); + let _ = arrived.send(()); + let _ = release.recv(); + guard = lock(fs); + if guard.crashed { + return None; + } + } let delay = guard.sync_delay; if !delay.is_zero() { drop(guard); @@ -248,6 +261,17 @@ impl SimStorage { lock(&self.fs).sync_delay = delay; } + /// The next sync stops once under way, with the process's lock released: it says so on the + /// returned receiver, then waits until the returned sender sends or is dropped. A test knows + /// from it that the writer is inside a sync, with no timer. A crash meanwhile + /// ([`SimStorage::restart`]) fails that sync when it goes on. + pub fn hold_next_sync(&self) -> (std::sync::mpsc::Receiver<()>, std::sync::mpsc::Sender<()>) { + let (arrived_tx, arrived) = std::sync::mpsc::channel(); + let (release, release_rx) = std::sync::mpsc::channel(); + lock(&self.fs).sync_hold = Some((arrived_tx, release_rx)); + (arrived, release) + } + pub fn is_crashed(&self) -> bool { lock(&self.fs).crashed } diff --git a/cmux-tui/crates/cmux-journal/src/storage.rs b/cmux-tui/crates/cmux-journal/src/storage.rs index df3fd381ec1b..a8107e019c6d 100644 --- a/cmux-tui/crates/cmux-journal/src/storage.rs +++ b/cmux-tui/crates/cmux-journal/src/storage.rs @@ -216,7 +216,7 @@ fn write_all_at(file: &std::fs::File, mut offset: u64, mut bytes: &[u8]) -> io:: /// durable. A call a signal interrupts (EINTR) is made again, as std does. std's `sync_data` and /// `sync_all` are both `F_FULLFSYNC` here, so they are not used. #[cfg(target_os = "macos")] -fn sync_file(file: &std::fs::File, level: SyncLevel) -> io::Result<()> { +pub fn sync_file(file: &std::fs::File, level: SyncLevel) -> io::Result<()> { use std::os::fd::AsRawFd; let fd = file.as_raw_fd(); let command = match level { @@ -256,7 +256,7 @@ fn retry_interrupted(call: impl Fn() -> libc::c_int) -> io::Result<()> { /// [`SyncLevel`] elsewhere: `fdatasync` (Linux; it also flushes a volatile drive cache) or `fsync`; /// `FlushFileBuffers` on Windows for both. #[cfg(not(target_os = "macos"))] -fn sync_file(file: &std::fs::File, level: SyncLevel) -> io::Result<()> { +pub fn sync_file(file: &std::fs::File, level: SyncLevel) -> io::Result<()> { match level { SyncLevel::Barrier => file.sync_data(), SyncLevel::Full => file.sync_all(),