From ed2dfc4a4bd6c99e1275984e35b2ac5a2720bd26 Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 14:45:57 -0700 Subject: [PATCH 1/3] feat(mirror): drive the mirror pass from disk events, keeping the timer as the backstop (#465) --- Cargo.lock | 156 ++++++- Cargo.toml | 2 +- SPEC.md | 25 +- crates/dig-node-service/Cargo.toml | 6 + crates/dig-node-service/src/mirror/events.rs | 416 +++++++++++++++++++ crates/dig-node-service/src/mirror/mod.rs | 1 + crates/dig-node-service/src/server.rs | 84 +++- 7 files changed, 685 insertions(+), 5 deletions(-) create mode 100644 crates/dig-node-service/src/mirror/events.rs diff --git a/Cargo.lock b/Cargo.lock index 2eba9957..442626d3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.195.0" +version = "0.202.0" dependencies = [ "async-trait", "axum", @@ -3065,6 +3065,7 @@ dependencies = [ "getrandom 0.2.17", "hex", "libc", + "notify", "num-bigint", "reqwest", "rpassword", @@ -4009,6 +4010,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "funty" version = "2.0.0" @@ -4696,6 +4706,26 @@ dependencies = [ "rustversion", ] +[[package]] +name = "inotify" +version = "0.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "153be1941a183ec9ccd095ddbe17a8b8d435ef6c76e9e02451b933c3999af2c8" +dependencies = [ + "bitflags", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "inout" version = "0.1.4" @@ -4796,6 +4826,26 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "kqueue" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags", + "libc", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -5010,6 +5060,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -5047,6 +5098,33 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -8459,6 +8537,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" +dependencies = [ + "windows-targets 0.53.5", +] + [[package]] name = "windows-sys" version = "0.61.2" @@ -8492,13 +8579,30 @@ dependencies = [ "windows_aarch64_gnullvm 0.52.6", "windows_aarch64_msvc 0.52.6", "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm", + "windows_i686_gnullvm 0.52.6", "windows_i686_msvc 0.52.6", "windows_x86_64_gnu 0.52.6", "windows_x86_64_gnullvm 0.52.6", "windows_x86_64_msvc 0.52.6", ] +[[package]] +name = "windows-targets" +version = "0.53.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" +dependencies = [ + "windows-link", + "windows_aarch64_gnullvm 0.53.1", + "windows_aarch64_msvc 0.53.1", + "windows_i686_gnu 0.53.1", + "windows_i686_gnullvm 0.53.1", + "windows_i686_msvc 0.53.1", + "windows_x86_64_gnu 0.53.1", + "windows_x86_64_gnullvm 0.53.1", + "windows_x86_64_msvc 0.53.1", +] + [[package]] name = "windows-threading" version = "0.2.1" @@ -8520,6 +8624,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" + [[package]] name = "windows_aarch64_msvc" version = "0.48.5" @@ -8532,6 +8642,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" +[[package]] +name = "windows_aarch64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" + [[package]] name = "windows_i686_gnu" version = "0.48.5" @@ -8544,12 +8660,24 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" +[[package]] +name = "windows_i686_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" + [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" +[[package]] +name = "windows_i686_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" + [[package]] name = "windows_i686_msvc" version = "0.48.5" @@ -8562,6 +8690,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" +[[package]] +name = "windows_i686_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" + [[package]] name = "windows_x86_64_gnu" version = "0.48.5" @@ -8574,6 +8708,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" +[[package]] +name = "windows_x86_64_gnu" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" + [[package]] name = "windows_x86_64_gnullvm" version = "0.48.5" @@ -8586,6 +8726,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" + [[package]] name = "windows_x86_64_msvc" version = "0.48.5" @@ -8598,6 +8744,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "windows_x86_64_msvc" +version = "0.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + [[package]] name = "winnow" version = "0.5.40" diff --git a/Cargo.toml b/Cargo.toml index 37dc3503..ff2d2f8e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.195.0" +version = "0.202.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/SPEC.md b/SPEC.md index 581d1c87..181e9d73 100644 --- a/SPEC.md +++ b/SPEC.md @@ -8448,7 +8448,9 @@ coin. The `intended_coin_id` is recorded at submission so §23.5's reconcile acc > not.** The stability-across-a-window rule, in both directions, is satisfied by the pure tracker in > `mirror/presence.rs` and its `SETTLING_WINDOW_MS`, and `mirror::runner::PassRunner` debounces the > advertisable half of every observation through it. **Nothing feeds THAT**: there is no periodic -> scan, no start-up scan and no watcher, because `MirrorEffects::observe_disk` has no implementation +> scan, no start-up scan, because `MirrorEffects::observe_disk` has no implementation. The WATCHER half is +> no longer pending: `mirror/events.rs` watches the capsule cache and accelerates the pass, bounded +> by the four rules below > — so the scanning cadence described below, the un-debounced start-up exemption, and the claim that > the periodic pass is the correctness mechanism are all still pending, tracked as > . @@ -8457,6 +8459,27 @@ Presence changes are detected by SCANNING, with an optional watcher as an accele reverse. A watcher event is exactly what a crash, an unmounted volume, or an uncovered path loses; the periodic pass (§25.4) is the correctness mechanism. +The watcher is implemented (`mirror/events.rs`). It watches the capsule cache directory and MUST +obey all four of the following; a node whose watcher cannot be established, or whose events are all +dropped, MUST still converge on the round timer alone. + +1. **An event MAY only lower the instant of the next pass, never raise it.** The round deadline is + computed on entry to the wait and every return happens at or before it, so the timer is a + backstop rather than a fallback. The round length MUST NOT be lengthened to compensate for having + events. +2. **Events are COALESCED, never queued.** The pending state is one instant, so N events in a window + — including events arriving while a pass is running or wedged — owe exactly ONE wake. +3. **An observing wake fires after `QUIET_PERIOD_MS` (5_000) of quiet**, and schedules exactly one + **settling** wake `SETTLING_WINDOW_MS` later. That second wake MUST NOT re-arm, so a burst of any + size causes at most two passes. +4. **No two event-driven passes are closer than `SETTLING_WINDOW_MS`.** A pass cannot act on a change + the tracker has not yet seen hold for a window, so anything closer is amplification on a path that + spends money. + +**Chain events do NOT trigger a pass.** A create is decided from disk presence, a new peak arrives +roughly every 18.75 seconds, and the epoch rollover a reclaim waits on is wall-clock rather than a +chain event. Chain is observed inside the pass, on the round timer. + The debounce is **presence-stable-for-a-window**, not a timer after an event: a bond must be observed in the SAME state across `SETTLING_WINDOW_MS` (default 30_000) before that state is acted on, in BOTH directions. An event-reset timer never settles under repeated rewrites and cannot see diff --git a/crates/dig-node-service/Cargo.toml b/crates/dig-node-service/Cargo.toml index 726c28fe..63b727a8 100644 --- a/crates/dig-node-service/Cargo.toml +++ b/crates/dig-node-service/Cargo.toml @@ -250,6 +250,12 @@ dig-urn-resolver = "0.5.3" # line (dig_ecosystem#3161) — so the two node crates can never disagree about a canonical value. dig-constants = "0.13.0" +# Filesystem notification, for the §25 mirror pass's disk-event accelerant (dig-node#465). The +# watcher is strictly an ACCELERANT: the round timer and the start-up reconcile remain the +# correctness path, and `watch_capsule_cache` returns `None` rather than erroring when no watcher +# can be established, so a platform this crate cannot watch simply keeps the timer. +notify = "8" + # Windows Service Control Protocol. service-manager only REGISTERS the service in the # SCM; the binary the SCM launches must itself speak the service protocol # (StartServiceCtrlDispatcher → report RUNNING/STOPPED) or the SCM kills it with error diff --git a/crates/dig-node-service/src/mirror/events.rs b/crates/dig-node-service/src/mirror/events.rs new file mode 100644 index 00000000..019a80c3 --- /dev/null +++ b/crates/dig-node-service/src/mirror/events.rs @@ -0,0 +1,416 @@ +//! Event-driven wakes for the §25 mirror pass — an accelerant over the round timer (dig-node#465). +//! +//! # Events buy latency. They never buy correctness. +//! +//! [`super`]'s module doc rejects a file watcher **as the correctness mechanism**, and that stands: +//! *"a watcher's event is exactly what a crash loses; a scan at start-up re-derives the whole answer +//! from two observations that survive anything."* Nothing here weakens it. The round timer and the +//! start-up reconcile are untouched — not deleted, and deliberately not lengthened to compensate for +//! having events, which would trade the correctness floor for responsiveness. +//! +//! So the invariant this module is built to keep is: **no behaviour depends on an event arriving.** +//! A wake decides only *when* the next pass runs, never *whether* one runs and never what it +//! concludes. Silence every source forever and the node converges exactly as it does today, one +//! round later. That is asserted rather than intended — see the silencing test below. +//! +//! # Why a disk watcher, and why NOT a chain subscription +//! +//! **Disk** is where the latency is. A create is decided from capsule presence, and a capsule +//! landing is otherwise invisible until the next scan. +//! +//! **Chain events do not trigger a pass, on purpose.** Three reasons, in order of weight: +//! +//! 1. A new peak arrives roughly every 18.75 seconds. Each pass reads chain and may *spend money*, +//! so peaks are the single largest amplification source available on this path — the trigger most +//! worth suppressing rather than adding. +//! 2. Chain activity cannot make a create newly correct. Creates are decided from disk presence; +//! chain is read to learn what is already bonded. +//! 3. The reclaim deadline the chain seems to imply — an epoch rolling over — is *wall-clock*, not a +//! chain event: `crate::collateral::current_epoch_now` derives it locally. A peak subscription +//! would be new integration bought to obtain a signal the node already has. +//! +//! Chain therefore stays observed inside the pass, on the round timer, where one round (10 minutes) +//! already bounds reclaim latency far inside an epoch. +//! +//! # Two wakes per burst, and why it is two rather than one +//! +//! [`super::presence`] settles a bond by observing it in the SAME state across +//! [`super::presence::SETTLING_WINDOW_MS`], and `since_ms` is stamped at the first *observation* — +//! not at the moment the file appeared. A single wake would therefore only ever RECORD the +//! appearance; something has to look again, later, for it to be acted on. +//! +//! So one burst of writes produces at most two passes, whatever its size: +//! +//! * an **observing** wake once writes have been quiet for [`QUIET_PERIOD_MS`], which stamps the +//! bond, and +//! * one **settling** wake [`super::presence::SETTLING_WINDOW_MS`] later, which is the earliest +//! instant the tracker can act on it. +//! +//! The settling wake does not re-arm, so the sequence terminates: N events in a quiet window cause +//! exactly one observing pass and exactly one settling pass, never N of either. +//! +//! Latency for a capsule copied in by hand goes from up to two round timers (~20 minutes, because +//! the first round only stamps it) to roughly `QUIET_PERIOD_MS + SETTLING_WINDOW_MS` — about 35 +//! seconds — without any figure in the settling contract changing. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use tokio::sync::Notify; + +/// How long disk writes must be quiet before the observing wake fires. +/// +/// A debounce, not a delay: it exists so that copying a large `.dig` in — which lands as a long +/// stream of write events — causes one pass when the copy finishes rather than one per event. +/// +/// Short on purpose, and NOT the settling window. Settling is a stability requirement on the +/// CAPSULE, enforced by [`super::presence`] regardless of what this module does. This is only how +/// long the waker waits to be reasonably sure the writing has stopped; lengthening it would delay +/// the observing pass without making any decision safer. +pub const QUIET_PERIOD_MS: u64 = 5_000; + +/// The floor on how often events may cause a pass, whatever arrives. +/// +/// The bound that matters for money: a pass reads chain and may spend, so a pathological tool — or +/// anything with write access to the cache directory — must not be able to drive the pass rate. +/// +/// It is one **settling window**, and that figure is derived rather than picked. A pass can only act +/// on a change once [`super::presence`] has seen that change hold for a settling window, so two +/// event-driven passes closer together than one window cannot reach a decision the later of them +/// would not have reached alone — the extra passes are pure amplification. +/// +/// [`QUIET_PERIOD_MS`] alone does NOT bound this, which is why the floor is not redundant: writes +/// arriving just slower than the quiet period quiesce every time, and would otherwise wake a pass +/// every few seconds indefinitely. +pub const MIN_EVENT_PASS_INTERVAL_MS: u64 = super::presence::SETTLING_WINDOW_MS; + +/// Decides WHEN disk events are allowed to become a pass. Pure, so the bound is testable. +/// +/// Holds no clock and no channel: every method takes `now_ms`, exactly as +/// [`super::presence::PresenceTracker`] does, so a test measures the window written in the test +/// rather than however long the test happened to take. +#[derive(Debug, Clone, Default)] +pub struct WakeCoalescer { + /// When the most recent event arrived, while a wake is still owed for it. + last_event_ms: Option, + /// A settling wake scheduled by a previous observing wake, and its instant. + follow_up_ms: Option, + /// When the last wake was taken, for the [`MIN_EVENT_PASS_INTERVAL_MS`] floor. + last_wake_ms: Option, +} + +impl WakeCoalescer { + /// A coalescer that has seen nothing. + pub fn new() -> Self { + Self::default() + } + + /// Record that something changed under the capsule cache. + /// + /// Collapsing rather than queueing is the whole point: the state is one instant, not a list, so + /// a burst of any size — including one that arrives while a pass is running, or while one is + /// wedged — occupies the same fixed space and produces the same single owed wake. + pub fn record_event(&mut self, now_ms: u64) { + self.last_event_ms = Some(now_ms); + } + + /// The instant at which a wake is next owed, if one is. + /// + /// `None` means nothing is pending and the caller waits on the round timer alone. + pub fn due_at_ms(&self) -> Option { + let quiet = self.last_event_ms.map(|at| at + QUIET_PERIOD_MS); + let due = match (quiet, self.follow_up_ms) { + (Some(a), Some(b)) => Some(a.min(b)), + (a, b) => a.or(b), + }?; + Some(match self.last_wake_ms { + Some(last) => due.max(last + MIN_EVENT_PASS_INTERVAL_MS), + None => due, + }) + } + + /// Take the owed wake if it is due at `now_ms`. + /// + /// Taking an OBSERVING wake schedules the one settling wake the module doc describes. Taking + /// that settling wake schedules nothing, so the sequence terminates and a burst can never + /// sustain itself. + pub fn take_due(&mut self, now_ms: u64) -> bool { + match self.due_at_ms() { + Some(due) if now_ms >= due => { + let was_observing = self.last_event_ms.is_some(); + self.last_event_ms = None; + self.follow_up_ms = if was_observing { + Some(now_ms + super::presence::SETTLING_WINDOW_MS) + } else { + None + }; + self.last_wake_ms = Some(now_ms); + true + } + _ => false, + } + } +} + +/// A live source of disk events, and the signal a waiting pass loop selects on. +/// +/// Held as an `Option` by the caller, so `None` is a node with no event source at all — the +/// configuration the convergence assertion runs in. +pub struct DiskEvents { + signal: Arc, + /// Kept alive for as long as events are wanted; dropping it stops the watcher. + _watcher: Box, +} + +impl DiskEvents { + /// Wait for the next disk event. Cancel-safe, so it composes with `tokio::select!`. + pub async fn changed(&self) { + self.signal.notified().await; + } +} + +/// Watch `cache_dir` for capsules appearing and disappearing. +/// +/// Watching the directory rather than hooking the code that writes into it is deliberate: it sees a +/// `.dig` copied in by hand or written by an unrelated tool, which is precisely the case +/// [`super::presence`] says the settling window exists for, and it needs no cooperation from any +/// write path. +/// +/// Returns `None` when no watcher can be established — an unsupported filesystem, a missing +/// directory, a platform limit. That is a latency loss and nothing more, which is why it is an +/// `Option` rather than an error worth failing bring-up over. +pub fn watch_capsule_cache(cache_dir: &Path) -> Option { + use notify::Watcher as _; + + let signal = Arc::new(Notify::new()); + let sink = Arc::clone(&signal); + let mut watcher = notify::recommended_watcher(move |res: notify::Result| { + // Any event under the cache is a hint that the capsule set MAY have changed. Deliberately + // unfiltered by path or kind: the pass re-derives the whole answer from a scan anyway, so a + // false hint costs one early pass and a missed one costs only latency. + if res.is_ok() { + sink.notify_one(); + } + }) + .ok()?; + watcher + .watch(cache_dir, notify::RecursiveMode::Recursive) + .ok()?; + + Some(DiskEvents { + signal, + _watcher: Box::new(watcher), + }) +} + +/// Where capsules land — `/modules`, the directory the inventory scan reads. +pub fn capsule_cache_dir(cache_root: &Path) -> PathBuf { + cache_root.join("modules") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::mirror::presence::SETTLING_WINDOW_MS; + + /// An explicit instant, for the same reason `presence.rs` pins one: the window under test must + /// be the one written here, not however long the test took to run. + const T0: u64 = 1_700_000_000_000; + + /// The round timer, for comparison. A wake is only worth anything if it is far inside this. + const ROUND_MS: u64 = dig_constants::MIRROR_ROUND_LENGTH_MS as u64; + + #[test] + fn a_capsule_appearing_wakes_a_pass_far_inside_the_round_timer() { + let mut c = WakeCoalescer::new(); + c.record_event(T0); + + let due = c.due_at_ms().expect("an event owes a wake"); + let delay = due - T0; + + // The DELAY is the assertion, not that a wake happened: a wake arriving at the round + // boundary would satisfy the latter and be worth nothing. + assert_eq!(delay, QUIET_PERIOD_MS); + assert!( + delay * 20 < ROUND_MS, + "an event-driven wake must be an order of magnitude inside the round timer, \ + but {delay}ms is not far inside {ROUND_MS}ms" + ); + assert!(c.take_due(T0 + delay)); + } + + #[test] + fn a_burst_of_events_in_one_window_produces_exactly_one_observing_pass() { + let mut c = WakeCoalescer::new(); + + // 200 events spread across a settling window — a large `.dig` being copied in. Recorded + // and drained in the SAME loop, as they arrive in production: a fixture that recorded them + // all up front would only ever see the last one and could not tell coalescing from luck. + let step = SETTLING_WINDOW_MS / 200; + let last_event = T0 + 199 * step; + + let mut passes = 0; + let mut now = T0; + // Up to the last event's quiet period plus one step, which is still well before the settling + // wake the observing pass schedules, so only observing passes are counted here. + while now <= last_event + QUIET_PERIOD_MS + step { + if now <= last_event && (now - T0) % step == 0 { + c.record_event(now); + } + if c.take_due(now) { + passes += 1; + } + now += step; + } + + assert_eq!( + passes, 1, + "200 events in one window must cause ONE observing pass, not one per event" + ); + } + + #[test] + fn a_burst_produces_a_settling_pass_and_then_stops() { + let mut c = WakeCoalescer::new(); + c.record_event(T0); + + let observing = c.due_at_ms().unwrap(); + assert!(c.take_due(observing)); + + // The settling wake is owed exactly one settling window later — the earliest instant + // `PresenceTracker` can act on a bond first stamped by the observing pass. + let settling = c + .due_at_ms() + .expect("an observing pass owes one settling pass"); + assert_eq!(settling - observing, SETTLING_WINDOW_MS); + assert!(c.take_due(settling)); + + // And then nothing. If this re-armed, one write would drive the pass forever. + assert_eq!( + c.due_at_ms(), + None, + "the settling wake must not re-arm, or a single event sustains passes indefinitely" + ); + assert!(!c.take_due(settling + ROUND_MS)); + } + + #[test] + fn events_arriving_during_a_pass_collapse_into_one_pending_wake() { + let mut c = WakeCoalescer::new(); + c.record_event(T0); + let observing = c.due_at_ms().unwrap(); + assert!(c.take_due(observing)); + + // The pass is now running. Thousands of events land while it does. + for i in 0..5_000u64 { + c.record_event(observing + i); + } + + // Exactly one further wake is owed over the next settling window. Not 5,000, and not one + // per event drained as a burst once the pass finishes. + let mut passes = 0; + let mut now = observing; + while now <= observing + SETTLING_WINDOW_MS + 1_000 { + if c.take_due(now) { + passes += 1; + } + now += 250; + } + assert_eq!( + passes, 1, + "events arriving during a pass must collapse into ONE owed wake, not queue" + ); + } + + #[test] + fn no_two_event_driven_passes_are_closer_than_the_floor() { + let mut c = WakeCoalescer::new(); + let mut taken: Vec = Vec::new(); + + // The fixture that actually reaches the floor: writes spaced just LONGER than the quiet + // period, so every one of them quiesces and owes a wake. A continuous write storm would not + // test this — it never goes quiet, so it never wakes a pass at all, and the round timer is + // the only thing that runs. This is the case the floor exists for. + let spacing = QUIET_PERIOD_MS + 1_000; + let mut now = T0; + while now < T0 + ROUND_MS { + if (now - T0) % spacing == 0 { + c.record_event(now); + } + if c.take_due(now) { + taken.push(now); + } + now += 250; + } + + assert!( + taken.len() >= 2, + "the fixture must actually produce passes, or the floor is untested" + ); + // Without the floor these would land one `spacing` apart. State it, so a later edit that + // drops the floor fails here rather than silently multiplying the pass rate. + assert!( + MIN_EVENT_PASS_INTERVAL_MS > spacing, + "the fixture must write faster than the floor permits" + ); + for pair in taken.windows(2) { + assert!( + pair[1] - pair[0] >= MIN_EVENT_PASS_INTERVAL_MS, + "a pass at {} followed one at {}, closer than the {}ms floor", + pair[1], + pair[0], + MIN_EVENT_PASS_INTERVAL_MS + ); + } + } + + #[test] + fn silencing_every_event_source_leaves_the_timer_untouched() { + // A coalescer that is never fed owes nothing, ever. This is the assertion that keeps events + // an accelerant: with no source the loop has only the round timer to wait on, which is + // exactly the behaviour that shipped before this module existed. + let mut c = WakeCoalescer::new(); + let mut now = T0; + while now < T0 + ROUND_MS * 3 { + assert_eq!(c.due_at_ms(), None); + assert!(!c.take_due(now)); + now += 1_000; + } + + // And the constant the backstop is built from is unchanged — an events change must never + // lengthen the timer to compensate. + assert_eq!(ROUND_MS, 10 * 60 * 1_000); + } + + /// The watcher itself, against a real directory: a file appearing must signal, promptly. + /// + /// Measures the DELAY rather than asserting a signal eventually arrived, and bounds it well + /// inside the round timer. No `tokio::time::pause()` here on purpose — a paused clock would + /// auto-advance past the very wall-clock interval being measured. + #[tokio::test] + async fn a_file_appearing_signals_well_before_the_round_timer() { + let dir = tempfile::tempdir().expect("a temp dir"); + let Some(events) = watch_capsule_cache(dir.path()) else { + // No watcher on this platform is a latency loss, not a failure; the timer still holds. + return; + }; + + let started = std::time::Instant::now(); + let path = dir.path().join("0f.dig"); + tokio::task::spawn_blocking(move || { + std::thread::sleep(std::time::Duration::from_millis(50)); + std::fs::write(&path, b"capsule").expect("write the capsule"); + }); + + let signalled = tokio::time::timeout(std::time::Duration::from_secs(20), events.changed()) + .await + .is_ok(); + let elapsed = started.elapsed(); + + assert!(signalled, "a file appearing under the cache must signal"); + assert!( + (elapsed.as_millis() as u64) * 10 < ROUND_MS, + "the watcher signalled after {elapsed:?}, which is not far inside the {ROUND_MS}ms round" + ); + } +} diff --git a/crates/dig-node-service/src/mirror/mod.rs b/crates/dig-node-service/src/mirror/mod.rs index 8fe6cfca..e01db347 100644 --- a/crates/dig-node-service/src/mirror/mod.rs +++ b/crates/dig-node-service/src/mirror/mod.rs @@ -78,6 +78,7 @@ //! `*_mojos` and come from separate coins so a fee can never shave collateral. pub mod advertise; +pub mod events; pub mod funding; pub mod lifecycle; pub mod observe; diff --git a/crates/dig-node-service/src/server.rs b/crates/dig-node-service/src/server.rs index e2aec190..74f52f36 100644 --- a/crates/dig-node-service/src/server.rs +++ b/crates/dig-node-service/src/server.rs @@ -2749,6 +2749,32 @@ fn spawn_mirror_passes( let journal = lifecycle::journal(); let mut presence = crate::mirror::presence::PresenceTracker::new(); + // The disk-event accelerant (dig-node#465). Strictly a hint about WHEN to run the next + // pass: `wait_for_next_pass` never lets an event push the round deadline out, and a `None` + // here is a node that waits on the timer alone and converges identically, one round later. + let capsule_cache = crate::mirror::events::capsule_cache_dir(node.cache_dir_path()); + // The watcher can only attach to a directory that exists, and a node that has never cached + // anything has not created it yet. Creating it is harmless: the inventory scan already + // treats an empty cache and a missing one the same way. + let _ = std::fs::create_dir_all(&capsule_cache); + let disk_events = crate::mirror::events::watch_capsule_cache(&capsule_cache); + if disk_events.is_some() { + tracing::info!( + target: "mirror", + dir = %capsule_cache.display(), + "watching the capsule cache, so a capsule landing is reconciled in seconds rather \ + than at the next round; the round timer remains the backstop" + ); + } else { + tracing::info!( + target: "mirror", + dir = %capsule_cache.display(), + "the capsule cache could not be watched, so the round timer is the only trigger; \ + this costs latency and changes nothing the node concludes" + ); + } + let mut wake = crate::mirror::events::WakeCoalescer::new(); + loop { let epoch = match current_epoch_now() { CurrentEpoch::Final(epoch) => epoch as i64, @@ -2852,11 +2878,67 @@ fn spawn_mirror_passes( ), } - tokio::time::sleep(MIRROR_PASS_INTERVAL).await; + wait_for_next_pass(disk_events.as_ref(), &mut wake).await; } }); } +/// Wait until the next mirror pass should run: the round deadline, or sooner on a disk event. +/// +/// # The deadline is a CEILING that events can only lower +/// +/// `MIRROR_PASS_INTERVAL` from the moment this is entered is computed once, up front, and every +/// return happens at or before it. An event cannot postpone a pass, and neither can a flood of them: +/// the loop below only ever chooses an EARLIER instant to wake at. That is what makes the timer a +/// backstop rather than a fallback — it fires on its own schedule whether or not any event ever +/// arrives, and this function is why silencing the watcher changes latency and nothing else. +/// +/// # Passes still cannot overlap +/// +/// The caller is one sequential task: it runs a pass, then awaits this. Waking early moves the next +/// pass forward; it does not start a second one. A pass that wedges therefore blocks its own +/// successor and nothing else, and the events that arrive meanwhile collapse into the single owed +/// wake [`crate::mirror::events::WakeCoalescer`] holds — never a queue that drains as a burst of +/// passes once the wedge clears. +async fn wait_for_next_pass( + events: Option<&crate::mirror::events::DiskEvents>, + wake: &mut crate::mirror::events::WakeCoalescer, +) { + let deadline = tokio::time::Instant::now() + MIRROR_PASS_INTERVAL; + + // No watcher: the behaviour that shipped before events existed, unchanged. + let Some(events) = events else { + tokio::time::sleep_until(deadline).await; + return; + }; + + loop { + let now_ms = crate::mirror::lifecycle::now_unix_ms(); + if wake.take_due(now_ms) { + return; + } + + // Wake at whichever comes first: the owed event wake, or the round deadline. + let next = match wake.due_at_ms() { + Some(due) => (tokio::time::Instant::now() + + std::time::Duration::from_millis(due.saturating_sub(now_ms))) + .min(deadline), + None => deadline, + }; + + tokio::select! { + _ = tokio::time::sleep_until(next) => { + if tokio::time::Instant::now() >= deadline { + return; + } + // Otherwise an event wake has come due; the next turn of the loop takes it. + } + // Cancel-safe, so an event arriving mid-sleep is never lost to the select. + _ = events.changed() => wake.record_event(crate::mirror::lifecycle::now_unix_ms()), + } + } +} + /// Report what one mirror pass did, to whoever is reading the node's log. /// /// A free function so the lines an operator actually reads can be asserted against, rather than From 0b6163c94e4d561f7efe1add74fe9c1449db075a Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Mon, 31 Aug 2026 22:37:30 -0700 Subject: [PATCH 2/3] =?UTF-8?q?chore(release):=200.211.0=20=E2=80=94=20eve?= =?UTF-8?q?nt-driven=20mirror=20pass,=20rustfmt=20and=20clippy=20clean?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Cargo.lock | 2 +- Cargo.toml | 2 +- crates/dig-node-service/src/mirror/events.rs | 4 ++-- crates/dig-node-service/src/mirror/mod.rs | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 21bfb7e2..82fed3f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.206.0" +version = "0.211.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 5dce89d2..a9130c19 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.206.0" +version = "0.211.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/crates/dig-node-service/src/mirror/events.rs b/crates/dig-node-service/src/mirror/events.rs index 019a80c3..b898d387 100644 --- a/crates/dig-node-service/src/mirror/events.rs +++ b/crates/dig-node-service/src/mirror/events.rs @@ -254,7 +254,7 @@ mod tests { // Up to the last event's quiet period plus one step, which is still well before the settling // wake the observing pass schedules, so only observing passes are counted here. while now <= last_event + QUIET_PERIOD_MS + step { - if now <= last_event && (now - T0) % step == 0 { + if now <= last_event && (now - T0).is_multiple_of(step) { c.record_event(now); } if c.take_due(now) { @@ -334,7 +334,7 @@ mod tests { let spacing = QUIET_PERIOD_MS + 1_000; let mut now = T0; while now < T0 + ROUND_MS { - if (now - T0) % spacing == 0 { + if (now - T0).is_multiple_of(spacing) { c.record_event(now); } if c.take_due(now) { diff --git a/crates/dig-node-service/src/mirror/mod.rs b/crates/dig-node-service/src/mirror/mod.rs index b9eb8684..8e9171e6 100644 --- a/crates/dig-node-service/src/mirror/mod.rs +++ b/crates/dig-node-service/src/mirror/mod.rs @@ -78,9 +78,9 @@ //! `*_mojos` and come from separate coins so a fee can never shave collateral. pub mod advertise; -pub mod events; #[cfg(test)] mod converge_tests; +pub mod events; pub mod funding; pub mod lifecycle; pub mod observe; From 65065a0ed5cc8b39a85132e0163b0dba42e6b75c Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Tue, 1 Sep 2026 08:44:55 -0700 Subject: [PATCH 3/3] test(mirror): cover the wait_for_next_pass seam, and name why observe_chain is timer-only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gate found the seam this change creates untested: every property the PR is about lives in `wait_for_next_pass`, and both halves it composes were already covered while their composition was not. Three cases on a paused clock: the round timer with no events (both the no-watcher path and the silent-watcher path), a single event waking a quiet period later, and a storm that OUTLASTS the round still returning by its deadline. The third is the one that matters -- it is the branch where the `events.changed()` arm keeps winning the `select!` and the deadline check in the sibling arm is re-entered rather than reached. Testing it at all required one clock rather than two: the deadline and every sleep are on tokio's monotonic clock while the coalescer's windows are unix milliseconds, and reading the wall clock per turn compared two quantities free to disagree -- under a wall-clock step (NTP, suspend/resume) as much as under a paused test clock. The unix figure is now anchored to elapsed monotonic time at entry. SPEC.md §25.5 also stated the chain half as a preference. It is a limitation: `ChainSource` is request/response with no subscription surface, §14.2 chain-watch is itself a poll loop, and the one real push path publishes a FIELDLESS `SyncEvent::CoinState` that names no coin. Recorded with the mechanism that would have to exist, as DIG-Network/dig-node#482. Refs #465 Co-Authored-By: Claude --- SPEC.md | 26 +++- crates/dig-node-service/src/mirror/events.rs | 15 ++ crates/dig-node-service/src/server.rs | 141 ++++++++++++++++++- 3 files changed, 176 insertions(+), 6 deletions(-) diff --git a/SPEC.md b/SPEC.md index b52196eb..66c02f53 100644 --- a/SPEC.md +++ b/SPEC.md @@ -8579,9 +8579,29 @@ dropped, MUST still converge on the round timer alone. the tracker has not yet seen hold for a window, so anything closer is amplification on a path that spends money. -**Chain events do NOT trigger a pass.** A create is decided from disk presence, a new peak arrives -roughly every 18.75 seconds, and the epoch rollover a reclaim waits on is wall-clock rather than a -chain event. Chain is observed inside the pass, on the round timer. +**Chain events do NOT trigger a pass, and no mechanism exists by which they could.** Chain is +observed inside the pass, on the round timer. This is a limitation of the interfaces available, not +only a design preference, and the two are worth separating: + +- **There is no chain event to subscribe to.** `MirrorEffects::observe_chain` reads through + `ChainSource` (`dig-chainsource-interface`), whose entire surface is request/response — + `coin_record`, `coin_records_by_puzzle_hash`, `coin_records_by_parent`, `coin_spend`. It exposes + no subscription, no stream and no callback, so there is nothing for a waiting pass to select on. + The node's own §14.2 chain-watch is likewise a POLL loop, not a push source. +- **The one push path in the node cannot say a mirror coin changed.** The wallet's direct-peer sync + (§18.6) does hold a real `request_puzzle_state(subscribe = true)` subscription and publishes to + the §18.14 `EventBus` — but `SyncEvent::CoinState` is fieldless. It names no coin, no puzzle hash + and no height, and it reports the WALLET DB rather than the mirror's chain view. Waking a pass on + it would wake a money-spending pass on any wallet coin activity whatsoever, with no evidence the + event was relevant, at up to the `SETTLING_WINDOW_MS` floor rather than the round. + +Independently of both, the two things a pass acts on are not chain-shaped: a CREATE is decided from +disk presence, which IS event-driven; and the epoch rollover a RECLAIM waits on is wall-clock. + +What a chain event WOULD buy is freshness of the §25.8 observation — a mirror coin spent out from +under this node is reported up to one round late. That is a staleness bound on a read-only surface, +never a money-safety gap, and closing it needs a coin-state push carrying the coin it is about. +Tracked as . The debounce is **presence-stable-for-a-window**, not a timer after an event: a bond must be observed in the SAME state across `SETTLING_WINDOW_MS` (default 30_000) before that state is acted diff --git a/crates/dig-node-service/src/mirror/events.rs b/crates/dig-node-service/src/mirror/events.rs index b898d387..ab5f7e5b 100644 --- a/crates/dig-node-service/src/mirror/events.rs +++ b/crates/dig-node-service/src/mirror/events.rs @@ -167,6 +167,21 @@ impl DiskEvents { pub async fn changed(&self) { self.signal.notified().await; } + + /// Build a `DiskEvents` driven by `signal` instead of a filesystem watcher. + /// + /// The waiting seam in `server.rs` is the only place the four SPEC §25.5 rules are actually + /// enforced, and a real watcher cannot express the case that matters — a storm of events + /// arriving faster than the loop can retire them — reproducibly. Driving the same `Notify` the + /// watcher callback drives keeps the test on the production path: everything downstream of + /// this constructor is the shipped code. + #[cfg(test)] + pub(crate) fn from_signal(signal: Arc) -> Self { + Self { + signal, + _watcher: Box::new(()), + } + } } /// Watch `cache_dir` for capsules appearing and disappearing. diff --git a/crates/dig-node-service/src/server.rs b/crates/dig-node-service/src/server.rs index 0350311c..6bb2bc0d 100644 --- a/crates/dig-node-service/src/server.rs +++ b/crates/dig-node-service/src/server.rs @@ -2951,7 +2951,8 @@ async fn wait_for_next_pass( events: Option<&crate::mirror::events::DiskEvents>, wake: &mut crate::mirror::events::WakeCoalescer, ) { - let deadline = tokio::time::Instant::now() + MIRROR_PASS_INTERVAL; + let entered = tokio::time::Instant::now(); + let deadline = entered + MIRROR_PASS_INTERVAL; // No watcher: the behaviour that shipped before events existed, unchanged. let Some(events) = events else { @@ -2959,8 +2960,20 @@ async fn wait_for_next_pass( return; }; + // ONE clock for the whole wait. The deadline and every sleep are on tokio's monotonic clock, + // while the coalescer's windows are unix milliseconds — so reading the wall clock per turn + // would compare two quantities that are free to disagree. They do disagree in practice: a + // wall-clock step (NTP, a suspend/resume) would move the coalescer's idea of "now" without + // moving a single sleep, which either fires the owed wake early or strands it. Anchoring unix + // ms to elapsed monotonic time at entry makes the arithmetic coherent, and is what lets the + // seam be tested on a paused clock at all. + let clock = { + let base = crate::mirror::lifecycle::now_unix_ms(); + move || base + entered.elapsed().as_millis() as u64 + }; + loop { - let now_ms = crate::mirror::lifecycle::now_unix_ms(); + let now_ms = clock(); if wake.take_due(now_ms) { return; } @@ -2981,7 +2994,7 @@ async fn wait_for_next_pass( // Otherwise an event wake has come due; the next turn of the loop takes it. } // Cancel-safe, so an event arriving mid-sleep is never lost to the select. - _ = events.changed() => wake.record_event(crate::mirror::lifecycle::now_unix_ms()), + _ = events.changed() => wake.record_event(clock()), } } } @@ -3719,3 +3732,125 @@ mod tests { ); } } + +/// The waiting seam itself — SPEC.md §25.5's four rules live here, not in the coalescer. +/// +/// `WakeCoalescer` is a pure type and `DiskEvents` is a thin wrapper over `Notify`; both are +/// already covered. Neither is the seam. Every property this module's change is ABOUT — the round +/// deadline as a ceiling an event may only lower, the `None`-events path being the pre-event +/// behaviour unchanged, and the `select!` turning a watcher signal into an early return — is a +/// property of [`super::wait_for_next_pass`] composing the two, and is invisible to a test of +/// either half. +/// +/// Every case runs on a PAUSED clock, so the windows asserted are the ones written here rather +/// than however long the test happened to take, and a ten-minute round costs no wall-clock time. +#[cfg(test)] +mod wait_for_next_pass_tests { + use super::MIRROR_PASS_INTERVAL; + use crate::mirror::events::{DiskEvents, WakeCoalescer, QUIET_PERIOD_MS}; + use std::sync::Arc; + use tokio::sync::Notify; + use tokio::time::{Duration, Instant}; + + /// A `DiskEvents` whose signal the test drives, and the handle to drive it with. + fn driven_events() -> (DiskEvents, Arc) { + let signal = Arc::new(Notify::new()); + (DiskEvents::from_signal(Arc::clone(&signal)), signal) + } + + /// With nothing to accelerate it, the wait IS the round timer — on both the no-watcher path + /// and the watcher-attached-but-silent path. + /// + /// The first is the behaviour that shipped before events existed and must be bit-for-bit the + /// old `sleep(MIRROR_PASS_INTERVAL)`; the second is the node whose watcher is live and whose + /// cache simply is not changing, which is the common case and must cost exactly the same. + #[tokio::test(start_paused = true)] + async fn with_no_events_the_wait_is_the_round_timer() { + let mut wake = WakeCoalescer::new(); + + let started = Instant::now(); + super::wait_for_next_pass(None, &mut wake).await; + assert_eq!( + started.elapsed(), + MIRROR_PASS_INTERVAL, + "the no-watcher path must be the round sleep, unchanged" + ); + + let (events, _signal) = driven_events(); + let started = Instant::now(); + super::wait_for_next_pass(Some(&events), &mut wake).await; + assert_eq!( + started.elapsed(), + MIRROR_PASS_INTERVAL, + "a silent watcher must cost exactly the round, not a shorter or longer wait" + ); + } + + /// One capsule landing wakes the pass in seconds, not at the next round. + /// + /// This is the entire user-visible point of the change, and it is the one assertion that + /// fails outright if the `events.changed()` arm is never wired to the coalescer. + #[tokio::test(start_paused = true)] + async fn a_single_event_wakes_the_pass_a_quiet_period_later() { + let (events, signal) = driven_events(); + let mut wake = WakeCoalescer::new(); + + let at = Duration::from_millis(100); + tokio::spawn(async move { + tokio::time::sleep(at).await; + signal.notify_one(); + }); + + let started = Instant::now(); + super::wait_for_next_pass(Some(&events), &mut wake).await; + let elapsed = started.elapsed(); + + let expected = at + Duration::from_millis(QUIET_PERIOD_MS); + assert!( + elapsed >= expected && elapsed < expected + Duration::from_secs(1), + "one event owes a wake one quiet period after it, not at the round: {elapsed:?}" + ); + assert!( + elapsed < MIRROR_PASS_INTERVAL, + "the wake must be far inside the round or it accelerates nothing" + ); + } + + /// SPEC.md §25.5 rule 1, on the branch that can actually break it: a storm that OUTLASTS the + /// round. + /// + /// Every event resets the quiet period, so the owed wake recedes for as long as writes keep + /// arriving — and each event also wins the `select!`, so the arm that checks the deadline is + /// re-entered rather than reached. Without the ceiling the wait would end when the WRITER + /// stopped, which is an instant an attacker with write access to the cache directory chooses. + /// With it, the round deadline still ends the wait on its own schedule, which is what makes + /// the timer a backstop rather than a fallback. + /// + /// The storm deliberately ends AFTER the deadline: a storm that stopped first would return at + /// its own quiet period and pass whether or not any ceiling existed. + #[tokio::test(start_paused = true)] + async fn a_storm_outlasting_the_round_still_returns_by_the_deadline() { + let (events, signal) = driven_events(); + let mut wake = WakeCoalescer::new(); + + let storm = MIRROR_PASS_INTERVAL + Duration::from_secs(100); + let storm_task = tokio::spawn(async move { + let until = Instant::now() + storm; + while Instant::now() < until { + signal.notify_one(); + tokio::time::sleep(Duration::from_secs(1)).await; + } + }); + + let started = Instant::now(); + super::wait_for_next_pass(Some(&events), &mut wake).await; + let elapsed = started.elapsed(); + storm_task.abort(); + + assert!( + elapsed <= MIRROR_PASS_INTERVAL + Duration::from_secs(2), + "an unending stream of events may not push the pass past its round deadline: \ + returned after {elapsed:?}, the round is {MIRROR_PASS_INTERVAL:?}" + ); + } +}