diff --git a/Cargo.lock b/Cargo.lock index 5472c3d930d..ba849675b5d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1676,6 +1676,8 @@ dependencies = [ "toyos-elide", "toyos-inspect", "toyos-logstream", + "toyos-quiesce", + "toyos-tmpdir", "toyos-wallclock", ] diff --git a/issues/a-file-held-across-a-file-servers-restart-answers-gone.md b/issues/a-file-held-across-a-file-servers-restart-answers-gone.md index 66d8ce393e6..2fbb2d38cb2 100644 --- a/issues/a-file-held-across-a-file-servers-restart-answers-gone.md +++ b/issues/a-file-held-across-a-file-servers-restart-answers-gone.md @@ -7,12 +7,22 @@ opened: 2026-09-27 # A file held across a file server's restart answers Gone -No volume records an object id, so nothing tells the held file from another -put at its path since, and a file held across a restart answers `Gone` -(`toyos::files`, so std's `File` and a C descriptor alike); `logd` loses -`/log` for the boot if LOG's server restarts. +A file id is good only on the connection it was answered on, and a server's +volume id is its pid (`system/fileserver/src/main.rs`), so a file held across +a restart answers `Stale` or `Gone` (`toyos::fs`, so std's `File` and a C +descriptor alike), and a program that does not open its path again loses +what it writes after. logkeeper does: a part its volume refuses or loses is +opened again and given the rest from its replay +(`system/logkeeper/src/store.rs`). Reopening by path is the hazard: a stick +swapped for another carrying a file under the same name has that file +rewritten past what this boot last synced to it. -**Exit**: DATA's entry carries an object id and a generation that nothing -reuses, a held handle reopens only when both match, and a host test that -makes a same-length file in a held file's freed block within the second has -the reopen refused. +On DATA a held file can be told from another put at its path since: bcachefs +never reuses an inode number (`next_inode`, A8.9). Reopening it by that +number waits on a volume identity that outlives a restart, which the pid is +not; ruled so by the coordinator, 2026-11-05. A FAT role has no such +number and keeps `Gone`. + +**Exit**: a file held on DATA across its server's restart is reopened by its +volume's identity and its inode number, and a host test that makes a file at +a held file's path before the restart has the held one still name its own. diff --git a/issues/a-log-volume-full-of-logkeepers-own-parts-is-never-written-again.md b/issues/a-log-volume-full-of-logkeepers-own-parts-is-never-written-again.md new file mode 100644 index 00000000000..b6e4e859ade --- /dev/null +++ b/issues/a-log-volume-full-of-logkeepers-own-parts-is-never-written-again.md @@ -0,0 +1,42 @@ +--- +status: open +kind: finding +opened: 2026-11-10 +severity: [data-loss] +--- + +# A log volume full of logkeeper's own parts is never written again + +`logkeeper` deletes its old files in one place, `sweep` +(`system/logkeeper/src/store.rs`), which runs from `Volume::open`, before the +boot's first create, and from `Volume::rotate`. It keeps at most +`MAX_LOG_FILES` files and counts files, not bytes. A part the volume refused +is opened again by `Volume::reopen`, which does not sweep, and a refused part +never fills, so it never reaches a rotation. + +So where `/log` has no room and holds fewer than `MAX_LOG_FILES` of +`logkeeper`'s files, nothing frees any: the open's sweep deletes nothing, the +create or a later write is refused, and every retry is refused the same way +for the rest of the boot. A later boot whose first create is refused adds no +file, so it stands where this one did. The boot's log is then in the replay +and on the console only, which `logkeeper` says, and the stop is not marked. + +With `MAX_LOG_FILES` or more files the open's sweep deletes only as many as +the count is over by, one where there are sixteen, and a rotation one more. +That frees what the retired files held and no more, so on a volume otherwise +full the boot is refused once it has written past that, and from there stands +as above. + +Sixteen parts of `MAX_LOG_BYTES` are under half of the smallest log volume, so +this needs something else to have filled it: the bootloader's files, a +diagnostic dropped beside the logs, or a volume smaller than the build makes. + +Read from the code. The refusal and the doubling retry were seen on the guest +runner on a volume a filler file had filled (run 42133377735's stalled row); +no run has filled a volume with `logkeeper`'s own parts. + +**Exit**: a part refused for want of room frees room from `logkeeper`'s own +older files by `retire`'s order before it is asked again, or the record of a +ruling that a full log volume is the owner's to clear. + +Owner: `system/logkeeper`; unheld. diff --git a/issues/a-machine-without-an-iommu-refuses-every-claim.md b/issues/a-machine-without-an-iommu-refuses-every-claim.md index 2b197b91cfd..1131ce2c002 100644 --- a/issues/a-machine-without-an-iommu-refuses-every-claim.md +++ b/issues/a-machine-without-an-iommu-refuses-every-claim.md @@ -39,9 +39,12 @@ On such a machine whose image is on its NVMe disk, which it: `diskserver` is refused the controller (`diskserver: NOT SERVING — pci:1b36:0010 is on this machine and the kernel refused this service its claim`), so the three file servers say `Data serving … — absent`, `Log serving -/log — absent` and `Boot serving /boot — absent`, and `logkeeper` says `no -/log on this machine - this boot's kernel log is on the console only`. Read off the no-unit arm of `iommu_virtio_platform`, which is -green with it: the arm asserts the refusal and reads no volume. **The kernel +/log — absent` and `Boot serving /boot — absent`, and `logkeeper` says its +first part `is asked again - this boot's lines are kept until /log takes +them`, publishes `degraded`, and asks at a doubling wait to 64 s for the rest +of the boot. The file servers' lines are read off the no-unit arm of +`iommu_virtio_platform`, which is green with them; `logkeeper`'s is read from +its code, since the arm asserts the refusal and reads no volume. **The kernel drives no USB either** since usbd took the xHCI (the small-kernel track's step 10), so on such a machine usbd is refused its controller too, a USB stick's volumes are absent like the NVMe disk's, and no USB keyboard or pointer diff --git a/issues/logkeeper-ends-the-boots-log-on-one-refused-create-and-nothing-durable-says-so.md b/issues/logkeeper-ends-the-boots-log-on-one-refused-create-and-nothing-durable-says-so.md deleted file mode 100644 index 6876a061d0c..00000000000 --- a/issues/logkeeper-ends-the-boots-log-on-one-refused-create-and-nothing-durable-says-so.md +++ /dev/null @@ -1,62 +0,0 @@ ---- -status: open -kind: defect -severity: [data-loss] -opened: 2026-09-21 ---- - -# logkeeper ends the boot's log on one refused create, and nothing durable says so - -logkeeper was `logd` when this was measured, and the kernel still drove the -stick; the sections below keep the names of then. - -T14 run 69 (`usb-transport-break` on the boot stick, this tree at `a9adf0d7`) -ran its job, synced, said `Rebooting.` at 5.808 s and read back `DONE`. Its -sealed record says - -``` -log: /log holds this boot to 0 ms and 302 record(s) committed after that reached no volume -``` - -and the stick carries `loader.log` and `attempts` and **no kernel log file at -all**. The disk was online to the end: the same record carries `usb-storage: -disk 0 does not implement SYNCHRONIZE CACHE (sense 0x05/0x20/0x00)` at 5.808 s, -which is a whole Bulk-Only round trip and a REQUEST SENSE answered by the stick -at the shutdown — and on an ordinary boot of this machine that line is written -at `logd`'s first `fsync` (run 64: 1.900 s), so in run 69 no `fsync` of `logd`'s -reached the device before the shutdown's own. - -## What is known - -- **The staged break lands inside `logd`.** The arm abandons the boot's first - WRITE(10). Under QEMU (`tests/images/jobcase.toml` on `Profile::Metal` with the arm) one - such boot has `spawn: - /system/bin/logd` at 0.301 s, the break at 0.323 s and `logd: this boot's - kernel log is …` after it: the write is `Volume::open` creating the file. -- **One refused create is no log for the boot.** `Volume::open` - (`system/logkeeper/src/store.rs`) answers `None` on any error from `create`, - with no retry, and `main` then runs to the end of the boot with no volume. A - refused append or flush later in the boot ends the volume the same way - (`answered`, `system/logkeeper/src/main.rs`). -- **What it says reaches nothing on this machine.** `logkeeper: cannot create …` - is logkeeper's own record, which with no volume reaches only its console; a - laptop with no serial port has no channel there. -- Under QEMU the same boot's recovery takes in about a millisecond, the create - succeeds and `/log` is complete, so no QEMU boot shows any of this. - -## What is not known - -Which error the create got in run 69. The disk never went offline, so it was -not a transport that gave up. Two candidates fit and neither is measured: the -re-issued WRITE(10) answered with CSW status 1 (`Scsi::Refused`, a device error -to every caller — a device that has just taken a class reset may owe a UNIT -ATTENTION), or the call ran out of `block::OPERATION` or of the driver's budget -after a break (`Scsi::Budget`, `WouldBlock`). Either leaves a `usb-storage:` -record, which reaches no file when the volume got nothing and, since the -black-box page went, crosses no reset either. - -## Exit condition - -A boot whose first device write is refused once and whose `/log` still holds -that boot's records — or, where `logd` does give a volume up, a record of why -on a channel that survives the machine, which its console is not. diff --git a/issues/std-file-lock-answers-ok-and-locks-nothing.md b/issues/std-file-lock-answers-ok-and-locks-nothing.md index 8f333c7f2d2..2fd57c7e7fd 100644 --- a/issues/std-file-lock-answers-ok-and-locks-nothing.md +++ b/issues/std-file-lock-answers-ok-and-locks-nothing.md @@ -17,7 +17,8 @@ cargo is one such program. Every file lock it takes is one of these calls (its `src/util/flock.rs` at `7c83d4cc0953`, the cargo the fork pins), so two cargo processes in one ToyOS are kept apart by nothing. -Owner: `issues/toyos-builds-itself.md`, whose M4 runs cargo in the guest; +Owner: `issues/toyos-builds-itself.md`, whose M4 runs cargo in the guest +(where it stays, by the coordinator's ruling of 2026-11-05); the locks libc owes for the same milestone are `issues/libc-refuses-what-toyos-cannot-yet-answer.md`'s. diff --git a/issues/storage-is-layers-and-a-role-is-a-filesystem.md b/issues/storage-is-layers-and-a-role-is-a-filesystem.md index bdefce9f0a8..0f06e4da139 100644 --- a/issues/storage-is-layers-and-a-role-is-a-filesystem.md +++ b/issues/storage-is-layers-and-a-role-is-a-filesystem.md @@ -34,6 +34,15 @@ what is left here is built on them. and ROOT), and a link that fails is refused by name. 7. NTFS read-only, postponed here by the owner. +**A DATA commit's durability costs two Flushes per journal entry**, three for +an entry naming itself (`write_entries`, `lib/bcachefs/src/engine/mount/mod.rs`): +a flush, the entry, a flush. A write with Force Unit Access is durable when +it completes, with no Flush after it (NVM Express NVM Command Set, Write's +FUA bit; SCSI SBC, WRITE(10)'s FUA bit). FUA is missing end to end: `toyos-blockring`'s +Write carries no flag, the nvme driver never sets Write's FUA bit, usbd +never sets WRITE(10)'s, and bcachefs asks for none. Assigned with #1131, +whose APST wake cost the second Flush is part of. + **A file's pages are the kernel's paged objects'** (memory stage 7a, built by this track): one per file, supplied by its file server, read by its holders without a request (`system/fileserver/src/pages.rs`). The server's block cache diff --git a/issues/the-console-waits-on-logkeepers-file-writes.md b/issues/the-console-waits-on-logkeepers-file-writes.md index 08ea5dbb044..71f20a5c4e5 100644 --- a/issues/the-console-waits-on-logkeepers-file-writes.md +++ b/issues/the-console-waits-on-logkeepers-file-writes.md @@ -8,13 +8,22 @@ severity: [hang] # The console waits on logkeeper's file writes `logkeeper`'s loop is the only thing that reads the rings and grows the -replay the console's carriers and readers are fed from, and it writes and -syncs each round to `/log` itself (`Log::to_volume`, `Volume::write` and -`Volume::sync` in `system/logkeeper/src`), each a blocking call on the LOG -file server. `LOG_WRITE_BUDGET` judges a call only once it returns. So a LOG -server that takes a write and never answers, alive but wedged, leaves every -console channel silent for as long as it does, with no failure bound, and the -supervisor restarts only a server that ended. +replay the console's carriers and readers are fed from, and it makes every +call on the LOG file server itself, each one blocking (`system/logkeeper/src`): + +- a round's write and sync (`Log::to_volume`: `Volume::write`, `Volume::sync`); +- the catch-up of a file behind the replay (`Log::caught_up`, from a round and + from `Log::adopt`: `Volume::catch_up`'s reopen, writes, sync and rotation); +- a rotation's sync, retention sweep and create (`Log::rotate_if_full`); +- the stop's catch-up and sync (`Volume::finish`), whose time bounds how many + calls it starts and not how long one takes; +- the stop mark and its removal (`store::mark_stopped`, `unmark_stopped`). + +`LOG_WRITE_BUDGET` judges a call only once it returns. So a LOG server that +takes a call and never answers, alive but wedged, leaves every console channel +silent for as long as it does, with no failure bound, and the supervisor +restarts only a server that ended. The stop's wait on the volume's opener is +the one wait here with a bound (`STOP_CATCH_UP`). Read from the code, not reproduced. The volume's open waited the same way before it moved to a thread of its own, which is how the console stood silent @@ -24,8 +33,10 @@ for the supervisor's 30 s declaration bound on `acpi_press_is_one_sci`'s boot. server stops answering mid-boot still carries its later lines. The shape: the file is a consumer of the replay with its own offset, written on its own thread as the readers and carriers are, and the stop asks it to reach the -replay's end and sync within a bound. #1190's per-file offset and bounded -stop catch-up (`Volume::through`, `catch_up`, `finish`) are its vehicle, so -whoever reconciles #1190 with this takes it up. +replay's end and sync within a bound. The per-file offset and the stop's +bounded catch-up are in the tree (`Volume::through`, `catch_up`, `finish`). +What is left is the move: the loop still owns the `Volume` and makes the calls +above, so the volume goes to a thread of its own that reads the replay, and +the stop's bound becomes a wait on that thread, which the loop can give up on. Owner: `system/logkeeper`; unheld. diff --git a/lib/logstream/src/lib.rs b/lib/logstream/src/lib.rs index 5f487389059..d5198e92440 100644 --- a/lib/logstream/src/lib.rs +++ b/lib/logstream/src/lib.rs @@ -122,7 +122,8 @@ pub const REGISTER: u32 = 1; pub const SERVED: u32 = 2; /// The supervisor → `logkeeper`: the machine is about to be stopped. `logkeeper` reads every ring -/// and the kernel's records, writes them, makes the volume durable, and +/// and the kernel's records, writes them, makes the volume durable — or says, +/// in the file and on the console, what of the boot it could not — and /// answers [`FLUSHED`]; the supervisor asks for the stop once it has, or once its bound /// is spent. pub const FLUSH: u32 = 5; diff --git a/system/logkeeper/Cargo.toml b/system/logkeeper/Cargo.toml index 919bdcc2ba4..0560522eff5 100644 --- a/system/logkeeper/Cargo.toml +++ b/system/logkeeper/Cargo.toml @@ -13,6 +13,12 @@ toyos-logstream = { path = "../../lib/logstream" } toyos-elide = { path = "../../lib/elide" } toyos-inspect = { path = "../../lib/inspect" } toyos-wallclock = { path = "../../lib/wallclock" } +# The supervisor's bound on the stop's flush, which the stop's catch-up keeps inside. +toyos-quiesce = { path = "../../lib/quiesce" } + +[dev-dependencies] +# A scratch `/log` for the volume's own tests. +toyos-tmpdir = { path = "../../lib/tmpdir" } [package.metadata.toyos.host] exempt.owns = "the kernel's record ring, read under the `logread` SysCap" diff --git a/system/logkeeper/src/inspect.rs b/system/logkeeper/src/inspect.rs index 0929b37c761..3f185a41fae 100644 --- a/system/logkeeper/src/inspect.rs +++ b/system/logkeeper/src/inspect.rs @@ -46,7 +46,8 @@ const ACCEPT: u64 = u64::MAX; pub enum State { /// Every round written, flushed and published. Writing = 0, - /// Every round answered, and slower than a log is worth. + /// The volume answers slower than a log is worth, or refused the part + /// and is asked again: the replay keeps what the file is owed. Degraded = 1, /// No volume, or one given up on: the log is on the console only. ConsoleOnly = 2, diff --git a/system/logkeeper/src/main.rs b/system/logkeeper/src/main.rs index adee0b8eb3b..c4237b1953b 100644 --- a/system/logkeeper/src/main.rs +++ b/system/logkeeper/src/main.rs @@ -45,12 +45,14 @@ //! The console's carriers are the third ([`carrier`]): the console UART's //! driver, and a VM's console channel, which the UART goes quiet for while it //! carries the log. Neither of the two can slow the file: each is a thread -//! with an offset into the replay, which follows the file once it is open. +//! with an offset into the replay. The replay follows the file while the file +//! has all of it, and leads it while the part is refused or catching up. //! -//! **The volume is opened on a thread of its own, and nothing waits for it**: -//! /log's file server may start late. Until it opens, the readers and the -//! console have each round as it is read; once it opens, the file first takes -//! the boot so far from the replay. +//! **The volume is opened on a thread of its own**: /log's file server may +//! start late. Only the stop's flush waits for it, within [`STOP_CATCH_UP`]. +//! Until it opens, the readers and the console have each round as it is read; +//! once it opens, the file takes the boot so far from the replay, a part's +//! worth a round. //! //! # Its whole authority //! @@ -79,6 +81,10 @@ //! served log is handed what the file holds and no more. A stop the kernel //! refused is the supervisor's [`toyos_logstream::RESUME`], and what was held is written //! then. +//! +//! **A volume that refuses, or whose server ends, is asked again** +//! ([`policy::Retry`], and at the stop's flush whatever the wait), and its +//! file given back what it missed ([`store::Volume`]). mod carrier; mod inspect; @@ -87,7 +93,7 @@ mod policy; mod serve; mod store; -use std::sync::mpsc::{self, TryRecvError}; +use std::sync::mpsc::{self, RecvTimeoutError, TryRecvError}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -106,10 +112,10 @@ use toyos_logstream::{ use toyos_wallclock::Civil; use carrier::Carriers; -use serve::{Owed, Positions}; +use serve::Positions; use origin::{Origin, Said}; -use policy::{Step, LOG_WRITE_BUDGET}; -use store::{Volume, DIR, MAX_LOG_BYTES, MAX_LOG_FILES}; +use policy::{Retry, Step, LOG_WRITE_BUDGET}; +use store::{Volume, DIR, MAX_LOG_BYTES, MAX_LOG_FILES, MAX_LOG_PARTS}; /// Records asked of `SYS_LOG_READ` at once: large enough that an ordinary /// boot's burst is a handful of syscalls rather than one per line. @@ -157,16 +163,18 @@ fn main() { std::thread::Builder::new() .name("log-volume".into()) .spawn(move || { - let volume = Volume::open(stem, MAX_LOG_BYTES, |line| say!("{line}")); + let volume = Volume::open(DIR, stem, MAX_LOG_BYTES, |line| say!("{line}")); match &volume { // This program's half of the startup report, in one line: the // kernel says whether it has a console, this program whether it // has a volume and what the name it chose was decided by. - Some(v) => say!("{LOG_OPENED}{} ({dated})", v.path()), - None => say!( - "logkeeper: no {DIR} on this machine - this boot's kernel log is on the console only \ - ({dated})" + Some(v) if v.is_open() => say!("{LOG_OPENED}{} ({dated})", v.path()), + Some(v) => say!( + "logkeeper: {} is asked again - this boot's lines are kept until {DIR} takes them ({dated})", + v.path() ), + // `Volume::open` has said why. + None => say!("logkeeper: this boot's kernel log is on the console only ({dated})"), } opened.send(volume).expect("logkeeper: the loop holds the volume's receiver for good"); }) @@ -204,6 +212,7 @@ fn main() { waiting: Vec::new(), volume: None, opening: Some(opening), + retry: None, degraded: false, boot_secs, hub, @@ -234,6 +243,8 @@ struct Log { volume: Option, /// The volume's opener, until it answers. opening: Option>>, + /// When a part the volume refused is next opened again. + retry: Option, /// Whether the volume answers, slower than `LOG_WRITE_BUDGET` a round. degraded: bool, boot_secs: Option, @@ -251,6 +262,14 @@ struct Stopping { marked: bool, } +/// How long the stop's flush may spend giving the file what it is owed: half +/// the supervisor's bound on the whole flush ([`toyos_quiesce::FLUSH_MS`]), +/// leaving the rest to drain the rings, the final sync and the answer. What +/// it does not reach is said, in the file and on the console. +const STOP_CATCH_UP: Duration = Duration::from_millis(toyos_quiesce::FLUSH_MS / 2); + +const OPENER_ENDED: &str = "logkeeper: the volume's thread ended without answering"; + /// Text a stop may hold back from the file: a refused stop writes it, and /// what is past it is counted and said then. const STOPPING_BYTES: usize = 1 << 20; @@ -279,6 +298,7 @@ impl Log { let state = match (&self.volume, self.degraded) { (None, _) if self.opening.is_some() => inspect::State::Opening, (None, _) => inspect::State::ConsoleOnly, + (Some(v), _) if !v.is_open() => inspect::State::Degraded, (Some(_), true) => inspect::State::Degraded, (Some(_), false) => inspect::State::Writing, }; @@ -571,8 +591,9 @@ impl Log { } return; } - // The file first: once it is open, a reader is served only what /log - // already holds, whenever the machine stops between the two. + // The file first: while it has all of the replay, a reader is served + // only what /log already holds, whenever the machine stops between the + // two. A part refused or catching up takes this text from the replay. self.to_volume(file.as_bytes(), !flushing); self.hub.append(file.as_bytes(), Some(Arc::clone(&through))); if let Some(spoken) = &self.spoken { @@ -588,6 +609,9 @@ impl Log { /// flush's rounds, which `flushed` makes durable together: a stop's flush /// is bounded by the supervisor, and a sync per round of it is what outruns that. fn to_volume(&mut self, text: &[u8], sync: bool) { + if !self.caught_up() { + return; + } let Some(v) = self.volume.as_mut() else { return }; let began = Instant::now(); let mut refused = v.write(text).err().map(|e| (Step::Append, e.to_string())); @@ -601,34 +625,80 @@ impl Log { self.answered(refused); } - /// The volume, once its opener answers outside a stop: the file first - /// takes the boot so far from the replay. + /// The volume, once its opener answers outside a stop. Its file stands + /// at the replay's start and is given the boot so far, a part's worth + /// here and a part's worth each round after. fn adopt(&mut self) { if self.stopping.is_some() { return; } let Some(opening) = &self.opening else { return }; - let volume = match opening.try_recv() { - Ok(volume) => volume, - Err(TryRecvError::Empty) => return, - Err(TryRecvError::Disconnected) => panic!("logkeeper: the volume's thread ended without answering"), - }; + match opening.try_recv() { + Ok(volume) => { + self.took(volume); + self.caught_up(); + } + Err(TryRecvError::Empty) => {} + Err(TryRecvError::Disconnected) => panic!("{OPENER_ENDED}"), + } + } + + /// The opener's answer. + fn took(&mut self, volume: Option) { self.opening = None; self.volume = volume; - let mut at = 0; - while self.volume.is_some() { - let text = match self.hub.owed(&mut at) { - Owed::Text(text) => text, - Owed::Gap(lost) => self.hub.own_line( - Severity::Warn, - &format!("logkeeper: the first {lost} bytes of this boot were let go before {DIR} opened"), - ), - Owed::Nothing => break, - }; - self.to_volume(&text, false); + } + + /// Whether the file has every byte the replay has, so a round's text + /// goes after them. A part the volume refused is opened again once the + /// retry is due, and the file is given what it is owed, a part's worth a + /// round. + fn caught_up(&mut self) -> bool { + let now = Instant::now(); + let Some(v) = self.volume.as_mut() else { return false }; + let was_open = v.is_open(); + if was_open && v.through() == self.hub.end() { + return true; + } + if !was_open && !self.retry.get_or_insert(Retry::after(now)).due(now) { + return false; + } + let hub = &self.hub; + let caught = v.catch_up( + |at, max| hub.owed(at, max), + |lost| hub.own_line(Severity::Warn, &format!("logkeeper: {lost} bytes of this boot were let go before {DIR} took them")), + MAX_LOG_BYTES, + |line| say!("{line}"), + ); + match caught { + Ok(caught) => { + if !was_open { + self.retry = None; + say!("logkeeper: {DIR} answers again - {} holds this boot's lines", v.path()); + } + if !v.is_open() { + // A continuation's create was refused. + self.retry = Some(Retry::after(now)); + } + caught && v.is_open() + } + Err(e) if v.part() >= MAX_LOG_PARTS => { + toyos::error!("logkeeper: {DIR} would not take a continuation ({e}) - {}", v.path()); + self.volume = None; + false + } + Err(e) => { + v.lost(); + match self.retry.as_mut() { + Some(retry) if !was_open => retry.again(now), + _ => { + toyos::error!("logkeeper: {DIR} has not answered ({e}) - {} is asked again", v.path()); + self.retry = Some(Retry::after(now)); + } + } + false + } } - let refused = self.sync().err(); - self.answered(refused); } /// Make the volume durable now. @@ -650,14 +720,14 @@ impl Log { return; }; match step { - // Stop feeding the volume, say so once, and keep running. + // Say so once, keep the lines in the replay, and ask again. Step::Append | Step::Flush => { toyos::error!( - "logkeeper: {DIR} has not answered ({}: {why}) - this boot's log is on the console \ - only from {path}", + "logkeeper: {DIR} has not answered ({}: {why}) - {path} is asked again, and this \ + boot's lines are kept until it answers", step.as_str() ); - self.volume = None; + self.lost(); } // Every call answered, slowly: the round is durable. Step::TooSlow => { @@ -679,23 +749,68 @@ impl Log { if !v.full() { return; } - if let Err(e) = v.rotate(|line| say!("{line}")) { - toyos::error!("logkeeper: {DIR} would not take a continuation ({e}) - {}", v.path()); - self.volume = None; + match v.rotate(|line| say!("{line}")) { + Ok(()) if v.is_open() => {} + Ok(()) => self.lost(), + Err(e) if v.part() >= MAX_LOG_PARTS => { + toyos::error!("logkeeper: {DIR} would not take a continuation ({e}) - {}", v.path()); + self.volume = None; + } + Err(e) => { + toyos::error!("logkeeper: {DIR} has not answered ({e}) - {} is asked again", v.path()); + self.lost(); + } + } + } + + /// The volume refused the part: it is asked again once the retry is due. + fn lost(&mut self) { + if let Some(v) = self.volume.as_mut() { + v.lost(); } + self.retry = Some(Retry::after(Instant::now())); } /// The supervisor asked for the log whole before the machine stops: every line is /// written by now, so the volume is made durable, and the supervisor is told. fn flushed(&mut self) { - // An opener that answered during the flush's rounds: the file takes - // the boot before the sync below makes it durable. - self.adopt(); - if self.opening.is_some() { - toyos::error!("logkeeper: {DIR} had not opened by the stop - this boot's log is on the console only"); + let deadline = Instant::now() + STOP_CATCH_UP; + // An opener that has answered since the loop last asked, or answers + // in the stop's own time: the file takes the boot in `finish` below, + // inside the deadline. + if let Some(opening) = &self.opening { + match opening.recv_timeout(deadline.saturating_duration_since(Instant::now())) { + Ok(volume) => self.took(volume), + Err(RecvTimeoutError::Timeout) => { + toyos::error!("logkeeper: {DIR} had not opened by the stop - this boot's log is on the console only") + } + Err(RecvTimeoutError::Disconnected) => panic!("{OPENER_ENDED}"), + } + } + if let Some(v) = self.volume.as_mut() { + let hub = &self.hub; + let finished = v.finish( + |at, max| hub.owed(at, max), + |lost| hub.own_line(Severity::Warn, &format!("logkeeper: {lost} bytes of this boot were let go before {DIR} took them")), + hub.end(), + MAX_LOG_BYTES, + || Instant::now() < deadline, + |short| { + let text = format!("logkeeper: this file ends {short} bytes short of the boot: the stop came before {DIR} took them"); + hub.own_line(Severity::Error, &text) + }, + |line| say!("{line}"), + ); + let short = hub.end() - v.through(); + match finished { + Ok(_) if short == 0 => {} + Ok(_) => toyos::error!("logkeeper: {} ends {short} bytes short of the boot at the stop", v.path()), + Err(e) => toyos::error!( + "logkeeper: {DIR} refused the stop's flush ({e}) - {} ends {short} bytes short of the boot", + v.path() + ), + } } - let refused = self.sync().err(); - self.answered(refused); // The console's carriers take the log up to here, and the UART what // the machine says from here as it stops. if let Some(carriers) = &self.carriers { @@ -703,18 +818,25 @@ impl Log { } self.stopping = Some(Stopping { text: String::new(), unwritten: 0, marked: false }); if let Err(e) = self.from_supervisor.signal(FLUSHED) { - panic!("logkeeper: the supervisor could not be told the log is whole: {e:?}"); + panic!("logkeeper: the supervisor could not be told the flush is done: {e:?}"); } } /// The supervisor's stop has synced its files and is about to stop its - /// drivers: write the stop mark over a durable log, and say so. No volume, - /// no call: `/log`'s server may never have answered. + /// drivers: write the stop mark over a durable log, and say so. The mark + /// is withheld, and no call made, unless the part is open: `/log`'s + /// server may never have answered, and a part the volume refused is not + /// this boot's log, so the mark would call a stop clean whose log is not + /// on the volume. fn mark(&mut self) { let Some(stopping) = self.stopping.as_mut() else { panic!("logkeeper: the supervisor asked for the stop mark, and no stop was flushed for"); }; - stopping.marked = self.volume.is_some() + let open = self.volume.as_ref().is_some_and(Volume::is_open); + if !open { + toyos::warn!("logkeeper: {DIR} does not hold this boot's log, so the stop is not marked - the next boot will not count this one as clean"); + } + stopping.marked = open && store::mark_stopped() .inspect_err(|e| { toyos::error!("logkeeper: {DIR} would not take the stop mark ({e}) - the next boot will not count this one as clean") diff --git a/system/logkeeper/src/policy.rs b/system/logkeeper/src/policy.rs index b957fc2167f..4f495f08ac5 100644 --- a/system/logkeeper/src/policy.rs +++ b/system/logkeeper/src/policy.rs @@ -1,4 +1,4 @@ -use std::time::Duration; +use std::time::{Duration, Instant}; /// The round time past which a volume that answered every call is announced /// slow. Nothing in this program ends a volume for slowness: only the device's @@ -10,6 +10,37 @@ use std::time::Duration; /// learns about it while they are still watching. pub const LOG_WRITE_BUDGET: Duration = Duration::from_secs(5); +/// How soon a volume that refused is asked again, and the longest the wait +/// doubles to. Policy: how often a dead volume is asked, and its refusal +/// said, against how soon one that answers again is written; the file loses +/// nothing either way while the replay holds what it is owed. +pub const RETRY_FIRST: Duration = Duration::from_secs(1); +pub const RETRY_MAX: Duration = Duration::from_secs(64); + +/// When a volume that refused is next asked again. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Retry { + at: Instant, + wait: Duration, +} + +impl Retry { + /// A first refusal at `now`. + pub fn after(now: Instant) -> Self { + Self { at: now + RETRY_FIRST, wait: RETRY_FIRST } + } + + pub fn due(&self, now: Instant) -> bool { + now >= self.at + } + + /// Asked again at `now` and refused: twice the wait, to [`RETRY_MAX`]. + pub fn again(&mut self, now: Instant) { + self.wait = (self.wait * 2).min(RETRY_MAX); + self.at = now + self.wait; + } +} + /// Which call refused. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Step { @@ -37,6 +68,22 @@ impl Step { mod tests { use super::*; + #[test] + fn a_refused_volume_is_asked_again_at_a_doubling_wait_to_a_bound() { + let mut now = Instant::now(); + let mut retry = Retry::after(now); + let mut waits = Vec::new(); + for _ in 0..9 { + let next = retry.at; + assert!(!retry.due(next - Duration::from_millis(1))); + assert!(retry.due(next)); + waits.push((next - now).as_secs()); + now = next; + retry.again(now); + } + assert_eq!(waits, [1, 2, 4, 8, 16, 32, 64, 64, 64]); + } + /// The console words, which the table in `main` is written against. #[test] fn every_step_names_itself() { diff --git a/system/logkeeper/src/serve.rs b/system/logkeeper/src/serve.rs index 0257d35b6ce..b8d08cb831c 100644 --- a/system/logkeeper/src/serve.rs +++ b/system/logkeeper/src/serve.rs @@ -3,9 +3,11 @@ //! //! **Every reader gets the boot from its first line**, however late it //! connects: the main loop appends each round's lines to one -//! [`toyos_logstream::Replay`], after the file has them once it is open, and a -//! reader is a thread with an offset into it. So the same text reaches `/log` and every -//! reader, in the same order, and nothing a reader does can reach the file. +//! [`toyos_logstream::Replay`], after the file has them while the file has +//! all of the replay; a part not open yet, refused or catching up takes them +//! from the replay later. A reader is a thread with an offset into it. So the +//! same text reaches `/log` and every reader, in the same order, and nothing +//! a reader does can reach the file. //! //! A reader that takes none for [`STALLED`] while bytes are owed to it is let //! go, and that is a line in the log. @@ -32,6 +34,8 @@ use toyos_abi::syscall::SyscallError; use toyos::say; use toyos_logstream::{Next, ProgramLine, Replay, Severity, Tag, LOGKEEPER, SERVED}; +use crate::store::Owed; + /// The most bytes one reader is handed per wake: bounds how long the replay's /// lock is held for a copy, not how far a reader may fall behind. const CHUNK: usize = 64 * 1024; @@ -172,18 +176,13 @@ impl Hub { self.lock().replay.end() } - /// What a file at the replay's offset `at` is owed next, without waiting, - /// and `at` past it. - pub fn owed(&self, at: &mut u64) -> Owed { - match self.lock().replay.next(*at, CHUNK) { - Next::Bytes(bytes) => { - *at += bytes.len() as u64; - Owed::Text(bytes.to_vec()) - } - Next::Evicted { lost, at: resume } => { - *at = resume; - Owed::Gap(lost) - } + /// What a file at the replay's offset `at` is owed next: at most `max` + /// bytes of whole lines, or one line longer than that, or the gap the + /// replay let go of. Never waits. + pub fn owed(&self, at: u64, max: usize) -> Owed { + match self.lock().replay.next(at, max) { + Next::Bytes(bytes) => Owed::Text(bytes.to_vec()), + Next::Evicted { lost, at } => Owed::Gap { lost, at }, Next::CaughtUp => Owed::Nothing, } } @@ -225,16 +224,6 @@ impl Hub { } } -/// What a file is owed of the replay. -pub enum Owed { - /// Whole lines. - Text(Vec), - /// Bytes the replay let go of before the file took them. - Gap(u64), - /// Everything the replay holds. - Nothing, -} - /// What a carrier is fed. pub enum Fed { /// Bytes of the replay, which a channel taking them speaks for. @@ -457,20 +446,4 @@ mod tests { assert_eq!(hub.wire_start(15), (15, None)); assert_eq!(hub.wire_start(3).1.as_deref(), Some(&b"program\n"[..]), "one that took less owes them again"); } - - /// A volume that opens late is owed the boot from its first line, without - /// waiting, and told how much the replay let go of before it opened. - #[test] - fn a_late_file_is_owed_the_boot_whole_and_what_was_let_go() { - let hub = Hub::start(12, None); - hub.append(b"one\n", None); - let mut at = 0; - assert!(matches!(hub.owed(&mut at), Owed::Text(text) if text == b"one\n")); - assert!(matches!(hub.owed(&mut at), Owed::Nothing), "caught up is answered, not waited on"); - hub.append(b"two\nthree\n", None); - let mut late = 0; - assert!(matches!(hub.owed(&mut late), Owed::Gap(8)), "past its 12 bytes the replay let `one` and `two` go"); - assert!(matches!(hub.owed(&mut late), Owed::Text(text) if text == b"three\n")); - assert!(matches!(hub.owed(&mut late), Owed::Nothing)); - } } diff --git a/system/logkeeper/src/store.rs b/system/logkeeper/src/store.rs index 4dda48e7346..38bd7b0911d 100644 --- a/system/logkeeper/src/store.rs +++ b/system/logkeeper/src/store.rs @@ -17,7 +17,7 @@ //! standing between those and `delete_file` is that neither names them. use std::fs::{self, File, OpenOptions}; -use std::io::Write; +use std::io::{Seek, SeekFrom, Write}; use std::path::PathBuf; use toyos_logstream::LOG_CONTINUES; @@ -74,8 +74,8 @@ fn stopped_path() -> String { /// Where one file of this boot's sequence is: `toyos_wallclock::Part` is its /// name. -pub fn path(stem: &str, part: u32) -> String { - format!("{DIR}/{}", Part { stem, part }) +pub fn path(dir: &str, stem: &str, part: u32) -> String { + format!("{dir}/{}", Part { stem, part }) } /// A path of this program's as its boot's stem and its part number. @@ -84,9 +84,9 @@ fn stem_and_part(path: &str) -> Option<(&str, u32)> { Some((stem, part)) } -/// Every file on `/log` that this program wrote, oldest first. -pub fn ours() -> Vec { - let Ok(entries) = fs::read_dir(DIR) else { return Vec::new() }; +/// Every file in `dir` that this program wrote, oldest first. +pub fn ours(dir: &str) -> Vec { + let Ok(entries) = fs::read_dir(dir) else { return Vec::new() }; let mut ours: Vec<(Class, String)> = entries .flatten() .filter_map(|entry| { @@ -95,7 +95,7 @@ pub fn ours() -> Vec { }) .collect(); ours.sort(); - ours.into_iter().map(|(_, name)| format!("{DIR}/{name}")).collect() + ours.into_iter().map(|(_, name)| format!("{dir}/{name}")).collect() } /// Which of `existing` — this program's files, oldest first — to delete so the @@ -140,8 +140,8 @@ pub fn retire(existing: &[String], current: &str) -> Vec { /// Delete what [`retire`] names, and answer what is left. Every deletion is /// named on the log. -fn sweep(current: &str, mut say: impl FnMut(String)) -> Vec { - let existing = ours(); +fn sweep(dir: &str, current: &str, mut say: impl FnMut(String)) -> Vec { + let existing = ours(dir); let going = retire(&existing, current); let mut kept: Vec = existing.into_iter().filter(|p| !going.contains(p)).collect(); for path in going { @@ -149,7 +149,7 @@ fn sweep(current: &str, mut say: impl FnMut(String)) -> Vec { // saying why is indistinguishable from a bug in this program. match fs::remove_file(&path) { Ok(()) => say(format!( - "logkeeper: {DIR} holds more than {MAX_LOG_FILES} logs, so {path} was deleted" + "logkeeper: {dir} holds more than {MAX_LOG_FILES} logs, so {path} was deleted" )), Err(e) => { say(format!( @@ -163,44 +163,70 @@ fn sweep(current: &str, mut say: impl FnMut(String)) -> Vec { } /// The lowest index no undated log on the volume is using. -pub fn undated_stem(kept: &[String]) -> Option { +pub fn undated_stem(dir: &str, kept: &[String]) -> Option { (0..MAX_LOG_FILES) .map(|i| format!("{UNDATED_STEM}-{i:02}")) - .find(|stem| !kept.iter().any(|path| path.starts_with(&format!("{DIR}/{stem}")))) + .find(|stem| !kept.iter().any(|path| path.starts_with(&format!("{dir}/{stem}")))) } /// This boot's file, and everything about where it is in its own sequence. +/// +/// **The file is the replay's text** (`crate::serve::Hub`), part after part, +/// with a note where the replay had let go of what the file was owed. Where +/// each part stands is kept twice, in its file's bytes and in the replay's +/// offsets, because a note is in the one and not the other. +/// +/// **A part the volume refused is opened again** ([`Volume::catch_up`]): cut +/// to what its last sync made durable — or rebuilt whole, where its file is +/// shorter than that — and given the rest from the replay through the same +/// part-sized writes and continuations as any other. pub struct Volume { - file: File, + dir: String, + /// `None` while the part must be opened again. + file: Option, stem: String, part: u32, - /// Bytes in the current part so far. Kept here rather than read back from - /// the filesystem so a disagreement shows up as a wrong length rather than - /// being silently corrected. + /// The replay's offset of this part's first byte of the boot. + from: u64, + /// Bytes in the part's file, and the replay's offset its text reaches. size: u64, + through: u64, + /// The same, as the part's last sync left them. + durable: u64, + durable_through: u64, rotate_at: u64, } +/// What the replay gives a file next ([`Volume::catch_up`]). +pub enum Owed { + /// The boot's text from the asked offset, whole lines. + Text(Vec), + /// The replay let go of `lost` bytes from the asked offset, and holds the + /// boot again from `at`. + Gap { lost: u64, at: u64 }, + /// The file has every byte the replay has. + Nothing, +} + impl Volume { - /// Open this boot's first file, after making room for it. + /// Open this boot's first file in `dir`, after making room for it. A + /// create the volume refuses leaves the part to be opened again; `None` + /// is a boot with no name to write under. /// /// `stem` is a wall-clock stamp, or `None` for a boot that could not be /// placed in time — which takes the lowest free `unknown-NN`. - pub fn open(stem: Option, rotate_at: u64, mut say: impl FnMut(String)) -> Option { + pub fn open(dir: &str, stem: Option, rotate_at: u64, mut say: impl FnMut(String)) -> Option { let stem = match stem { Some(stem) => stem, - None => match undated_stem(&ours()) { + None => match undated_stem(dir, &ours(dir)) { Some(stem) => stem, None => { - say(format!( - "logkeeper: no free {UNDATED_STEM} name on {DIR}; this boot's log has nowhere \ - to go" - )); + say(format!("logkeeper: no free {UNDATED_STEM} name on {dir}; this boot's log has nowhere to go")); return None; } }, }; - let kept = sweep(&stem, &mut say); + let kept = sweep(dir, &stem, &mut say); // The first part number this boot's name does not already have on the // volume. Two boots inside one second is a machine nobody has, but a // test that stages the wall clock has it every run, and a colliding @@ -208,40 +234,79 @@ impl Volume { // // **Exhaustion says so before it gives up**, so the machine's log never // goes to the console only with nothing saying why. - let Some(part) = (1..=MAX_LOG_PARTS).find(|p| !kept.contains(&path(&stem, *p))) else { + let Some(part) = (1..=MAX_LOG_PARTS).find(|p| !kept.contains(&path(dir, &stem, *p))) else { say(format!( "logkeeper: every one of the {MAX_LOG_PARTS} part numbers under {stem} is taken on \ - {DIR}; this boot's log has nowhere to go" + {dir}; this boot's log has nowhere to go" )); return None; }; - let full = path(&stem, part); - let file = match create(&full) { - Ok(file) => file, - Err(e) => { - say(format!("logkeeper: cannot create {full}: {e}")); - return None; - } - }; - Some(Volume { file, stem, part, size: 0, rotate_at }) + let full = path(dir, &stem, part); + let file = create(&full) + .inspect_err(|e| say(format!("logkeeper: cannot create {full}: {e}; it is asked again"))) + .ok(); + Some(Volume { + dir: dir.to_string(), + file, + stem, + part, + from: 0, + size: 0, + through: 0, + durable: 0, + durable_through: 0, + rotate_at, + }) } /// Where this boot's log is being written. pub fn path(&self) -> String { - path(&self.stem, self.part) + path(&self.dir, &self.stem, self.part) + } + + /// Whether the part is open: false once the volume refused it, until + /// [`Self::catch_up`] opens it again. + pub fn is_open(&self) -> bool { + self.file.is_some() + } + + /// The volume refused the part: it is opened again before the next + /// write, and holds no more than its last sync made durable. + pub fn lost(&mut self) { + self.file = None; + (self.size, self.through) = (self.durable, self.durable_through); + } + + /// The replay's offset the file's text reaches: anything past it is owed. + pub fn through(&self) -> u64 { + self.through } - /// Write already-rendered lines at the part's end. The caller batches; - /// this does no buffering of its own beyond what the file cache is. + /// Write a round of the boot's text, the replay's next bytes, at the + /// part's end. The caller batches; this does no buffering of its own + /// beyond what the file cache is. pub fn write(&mut self, lines: &[u8]) -> std::io::Result<()> { - self.file.write_all(lines)?; - self.size += lines.len() as u64; + self.put(lines)?; + self.through += lines.len() as u64; + Ok(()) + } + + /// Bytes into the file that are none of the replay's: a note. + fn put(&mut self, bytes: &[u8]) -> std::io::Result<()> { + self.open_file()?.write_all(bytes)?; + self.size += bytes.len() as u64; Ok(()) } /// Get everything written so far onto the device, cache flush included. pub fn sync(&mut self) -> std::io::Result<()> { - self.file.sync_all() + self.open_file()?.sync_all()?; + (self.durable, self.durable_through) = (self.size, self.through); + Ok(()) + } + + fn open_file(&mut self) -> std::io::Result<&mut File> { + self.file.as_mut().ok_or_else(|| std::io::Error::other("the part is not open")) } /// Bytes in the current part so far. @@ -259,7 +324,101 @@ impl Volume { self.size >= self.rotate_at } - /// Finish this part and carry on in the next file of this boot's sequence. + /// Open the part again and write it what it is owed, `owed(offset, max)` + /// answering the replay from an offset, at most `max` bytes. At most + /// `budget` bytes of the boot this call, through part-sized writes and + /// their continuations, each part synced as it fills; `note(lost)` is the + /// line in place of a gap. Whether the file has every byte the replay + /// has; an `Err` is the volume refusing again. + pub fn catch_up( + &mut self, + mut owed: impl FnMut(u64, usize) -> Owed, + note: impl Fn(u64) -> Vec, + budget: u64, + mut say: impl FnMut(String), + ) -> std::io::Result { + if self.file.is_none() { + self.reopen(&mut say)?; + } + let mut written = 0u64; + loop { + if self.full() { + self.rotate(&mut say)?; + if !self.is_open() { + return Ok(false); + } + } + if written >= budget { + return Ok(false); + } + let room = (self.rotate_at - self.size).min(budget - written); + match owed(self.through, room as usize) { + Owed::Text(text) => { + written += text.len() as u64; + self.write(&text)?; + } + Owed::Gap { lost, at } => { + self.put(¬e(lost))?; + self.through = at; + } + Owed::Nothing => { + self.sync()?; + return Ok(true); + } + } + } + } + + /// The stop: catch up while `more()` says there is time, `budget` at a + /// time, then make durable what was written, with `shortfall(bytes)` in + /// the file where the replay holds bytes up to `end` it does not. The + /// bytes it ends short by; an `Err` is the volume refusing again, the part + /// [`Self::lost`] and the file as its last sync left it. + #[allow(clippy::too_many_arguments)] + pub fn finish( + &mut self, + mut owed: impl FnMut(u64, usize) -> Owed, + note: impl Fn(u64) -> Vec, + end: u64, + budget: u64, + mut more: impl FnMut() -> bool, + shortfall: impl Fn(u64) -> Vec, + mut say: impl FnMut(String), + ) -> std::io::Result { + let mut stop = || -> std::io::Result { + while !self.catch_up(&mut owed, ¬e, budget, &mut say)? && self.is_open() && more() {} + let short = end - self.through; + if self.is_open() { + if short > 0 { + self.put(&shortfall(short))?; + } + self.sync()?; + } + Ok(short) + }; + stop().inspect_err(|_| self.lost()) + } + + /// Open the part again at what its last sync made durable, or empty and + /// from its start where its file is shorter than that: missing, cut, or + /// never as long as the sync was told. + fn reopen(&mut self, say: &mut impl FnMut(String)) -> std::io::Result<()> { + let path = self.path(); + let mut file = OpenOptions::new().write(true).create(true).truncate(false).open(PathBuf::from(&path))?; + if file.metadata()?.len() < self.durable { + say(format!("logkeeper: {path} is shorter than its last sync left it; it is written again whole")); + (self.durable, self.durable_through) = (0, self.from); + } + file.set_len(self.durable)?; + file.seek(SeekFrom::End(0))?; + (self.size, self.through) = (self.durable, self.durable_through); + self.file = Some(file); + Ok(()) + } + + /// Finish this part and carry on in the next file of this boot's + /// sequence: `Err` where this part did not finish, and a refused create + /// leaves the next part to be opened again. pub fn rotate(&mut self, mut say: impl FnMut(String)) -> std::io::Result<()> { let full = self.path(); let bytes = self.size; @@ -268,13 +427,16 @@ impl Volume { } // Durable before it is let go: nothing syncs a part once the next is // the one being written. - self.file.sync_all()?; - sweep(&self.stem, &mut say); + self.sync()?; + sweep(&self.dir, &self.stem, &mut say); self.part += 1; - self.size = 0; + self.from = self.through; + (self.size, self.durable, self.durable_through) = (0, 0, self.through); let next = self.path(); - self.file = create(&next)?; say(format!("logkeeper: {full} reached {bytes}{LOG_CONTINUES}{next}")); + self.file = create(&next) + .inspect_err(|e| say(format!("logkeeper: cannot create {next}: {e}; it is asked again"))) + .ok(); Ok(()) } } @@ -290,9 +452,196 @@ fn create(path: &str) -> std::io::Result { #[cfg(test)] mod tests { use super::*; + use crate::serve::Hub; + + const STEM: &str = "2026-11-05-120000"; + + fn note(lost: u64) -> Vec { + format!("[{lost} bytes let go]\n").into_bytes() + } + + /// One of the loop's rounds: the file given what it is owed, then the + /// round's text written and synced where it is caught up, then the + /// replay's. + fn round(v: &mut Volume, hub: &Hub, text: &[u8]) { + let caught = (v.is_open() && v.through() == hub.end()) + || v.catch_up(|at, max| hub.owed(at, max), note, MAX_LOG_BYTES, |_| {}).unwrap_or(false); + if caught { + v.write(text).unwrap(); + v.sync().unwrap(); + if v.full() { + v.rotate(|_| {}).unwrap(); + } + } + hub.append(text, None); + } + + /// Every part of the boot, in order, as the volume holds them; none past + /// its bound by more than the line that filled it. + fn on_disk(dir: &str, v: &Volume) -> Vec { + let parts: Vec> = (1..=v.part()).map(|p| fs::read(path(dir, STEM, p)).unwrap_or_default()).collect(); + for (p, part) in parts.iter().enumerate() { + assert!(part.len() < 64 + 24, "part {} is {} bytes, past its 64", p + 1, part.len()); + } + parts.concat() + } + + fn line(n: usize) -> Vec { + format!("line {n:04}\n").into_bytes() + } + + /// The record's own case: the first part's create refused at open. The + /// part is made at the first catch-up, given the boot from its start. + #[test] + fn a_first_part_refused_at_open_is_made_and_given_the_boot() { + let tmp = toyos_tmpdir::TempDir::new("logkeeper-store"); + // A log directory not there yet refuses the create, whoever asks. + let log = tmp.path().join("log"); + let dir = log.to_str().unwrap(); + let hub = Hub::start(1 << 16, None); + let mut v = Volume::open(dir, Some(STEM.into()), 1 << 16, |_| {}).unwrap(); + assert!(!v.is_open(), "a create the volume refused left the part open"); + let mut boot = line(0); + round(&mut v, &hub, &boot); + fs::create_dir(&log).unwrap(); + round(&mut v, &hub, &line(1)); + boot.extend(line(1)); + assert_eq!(on_disk(dir, &v), boot); + } + + /// **The stop says what it could not write.** A file owed more than the + /// stop has time for is synced as far as it got and ends with the + /// shortfall; one the stop has time for is whole. + #[test] + fn a_stop_with_more_owed_than_its_time_syncs_and_says_the_shortfall() { + let shortfall = |short: u64| format!("[{short} bytes short]\n").into_bytes(); + for time in [false, true] { + let tmp = toyos_tmpdir::TempDir::new("logkeeper-store"); + let dir = tmp.path().to_str().unwrap(); + let hub = Hub::start(1 << 16, None); + let mut v = Volume::open(dir, Some(STEM.into()), 1 << 16, |_| {}).unwrap(); + let mut boot = Vec::new(); + for n in 0..2 { + round(&mut v, &hub, &line(n)); + boot.extend(line(n)); + } + v.lost(); + for n in 2..40 { + hub.append(&line(n), None); + boot.extend(line(n)); + } + let short = v + .finish(|at, max| hub.owed(at, max), note, hub.end(), 64, || time, shortfall, |_| {}) + .unwrap(); + let mut want = boot[..boot.len() - short as usize].to_vec(); + if short > 0 { + want.extend(shortfall(short)); + } + assert_eq!(short == 0, time, "the stop with time is whole, the one without is short"); + assert_eq!(fs::read(v.path()).unwrap(), want); + assert_eq!(v.durable, v.size, "what was written is synced"); + } + } + + /// A stop whose reopen is refused after an unsynced write: the file is + /// what its last sync left, and the bytes it lacks are counted from + /// there, the unsynced write's included. + #[test] + fn a_stop_refused_after_an_unsynced_write_counts_from_the_last_sync() { + let tmp = toyos_tmpdir::TempDir::new("logkeeper-store"); + let dir = tmp.path().to_str().unwrap(); + let hub = Hub::start(1 << 16, None); + let mut v = Volume::open(dir, Some(STEM.into()), 1 << 16, |_| {}).unwrap(); + round(&mut v, &hub, &line(0)); + let synced = hub.end(); + v.write(&line(1)).unwrap(); + hub.append(&line(1), None); + v.lost(); + hub.append(&line(2), None); + // A directory where the part was refuses its reopen, whoever asks. + fs::remove_file(v.path()).unwrap(); + fs::create_dir(v.path()).unwrap(); + let stopped = v.finish(|at, max| hub.owed(at, max), note, hub.end(), 64, || true, |_| Vec::new(), |_| {}); + assert!(stopped.is_err(), "a reopen onto a directory was answered"); + assert_eq!(hub.end() - v.through(), hub.end() - synced, "the shortfall counts from the last sync"); + } + + /// **The file is the replay's text, whatever it was refused.** Through + /// an unsynced write the volume holds past its sync, a gap the replay let + /// go of, a continuation whose create is refused, and a part cut short + /// behind the volume's back, the parts on disk are the boot with one + /// note where the gap was; the file's offsets and the replay's stay + /// apart across the note. + #[test] + fn a_refused_file_is_given_back_the_boot_the_replay_holds() { + let tmp = toyos_tmpdir::TempDir::new("logkeeper-store"); + let dir = tmp.path().to_str().unwrap(); + let hub = Hub::start(256, None); + let mut v = Volume::open(dir, Some(STEM.into()), 64, |_| {}).unwrap(); + let mut boot = Vec::new(); + let n = std::cell::Cell::new(0); + let rounds = |v: &mut Volume, count: usize, boot: &mut Vec| { + for _ in 0..count { + let text = line(n.replace(n.get() + 1)); + round(v, &hub, &text); + boot.extend(text); + } + }; + rounds(&mut v, 4, &mut boot); + assert_eq!(on_disk(dir, &v), boot); + + // A write that reached the file and no sync, more bytes after it, and + // the volume refusing: the reopen cuts both away. + let text = line(900); + v.write(&text).unwrap(); + hub.append(&text, None); + boot.extend(&text); + fs::OpenOptions::new().append(true).open(v.path()).unwrap().write_all(b"torn").unwrap(); + v.lost(); + rounds(&mut v, 3, &mut boot); + assert_eq!(on_disk(dir, &v), boot, "after an unsynced write"); + + // Refused long enough that the replay lets go of what the file is owed. + v.lost(); + let gap_from = v.through() as usize; + for _ in 0..40 { + let text = line(n.replace(n.get() + 1)); + hub.append(&text, None); + boot.extend(text); + } + let Owed::Gap { lost, at } = hub.owed(v.through(), 0) else { panic!("the replay let nothing go") }; + assert_eq!(at, gap_from as u64 + lost); + rounds(&mut v, 2, &mut boot); + let mut want = boot[..gap_from].to_vec(); + want.extend(note(lost)); + want.extend(&boot[at as usize..]); + assert_eq!(String::from_utf8(on_disk(dir, &v)).unwrap(), String::from_utf8(want.clone()).unwrap(), "across the gap"); + + // A continuation whose create is refused: its name is a directory. + let next = path(dir, STEM, v.part() + 1); + fs::create_dir(&next).unwrap(); + while v.is_open() { + let text = line(n.replace(n.get() + 1)); + round(&mut v, &hub, &text); + boot.extend(&text); + want.extend(text); + } + fs::remove_dir(&next).unwrap(); + rounds(&mut v, 2, &mut boot); + want.extend(&boot[boot.len() - line(0).len() * 2..]); + assert_eq!(on_disk(dir, &v), want, "after a refused continuation"); + + // The part cut to nothing behind the volume's back: rebuilt whole. + fs::write(v.path(), b"").unwrap(); + v.lost(); + rounds(&mut v, 2, &mut boot); + want.extend(&boot[boot.len() - line(0).len() * 2..]); + assert_eq!(on_disk(dir, &v), want, "after the part was cut"); + assert!(v.part() > 2, "the parts rotated: {}", v.part()); + } fn boot(stem: &str, parts: u32) -> Vec { - (1..=parts).map(|p| path(stem, p)).collect() + (1..=parts).map(|p| path(DIR, stem, p)).collect() } /// A volume under the bound loses nothing. @@ -312,9 +661,9 @@ mod tests { existing.extend(boot("2026-09-26-000000", 8)); assert_eq!(existing.len(), 16); let going = retire(&existing, "2026-09-26-000000"); - assert_eq!(going, [path("2026-09-26-000000", 2)]); + assert_eq!(going, [path(DIR, "2026-09-26-000000", 2)]); for earlier in 0..8 { - assert!(!going.contains(&path(&format!("2026-09-1{earlier}-000000"), 1))); + assert!(!going.contains(&path(DIR, &format!("2026-09-1{earlier}-000000"), 1))); } } @@ -331,21 +680,21 @@ mod tests { assert_eq!(existing.len(), 16); // Eleven earlier boots, eight kept: the three oldest may go whole, and // the oldest alone makes the room one new part needs. - assert_eq!(retire(&existing, "2026-09-26-000000"), [path("2026-08-10-000000", 1)]); + assert_eq!(retire(&existing, "2026-09-26-000000"), [path(DIR, "2026-08-10-000000", 1)]); let mut existing: Vec = (0..8).flat_map(|b| boot(&format!("2026-08-{:02}-000000", 10 + b), 1)).collect(); existing.extend(boot("2026-09-01-000000", 5)); existing.extend(boot("2026-09-26-000000", 3)); assert_eq!(existing.len(), 16); - assert_eq!(retire(&existing, "2026-09-26-000000"), [path("2026-08-10-000000", 1)]); + assert_eq!(retire(&existing, "2026-09-26-000000"), [path(DIR, "2026-08-10-000000", 1)]); let mut existing: Vec = (0..7).flat_map(|b| boot(&format!("2026-08-{:02}-000000", 10 + b), 1)).collect(); existing.extend(boot("2026-09-01-000000", 6)); existing.extend(boot("2026-09-26-000000", 3)); assert_eq!(existing.len(), 16); - assert_eq!(retire(&existing, "2026-09-26-000000"), [path("2026-09-01-000000", 2)]); + assert_eq!(retire(&existing, "2026-09-26-000000"), [path(DIR, "2026-09-01-000000", 2)]); } }