From 320db0e37b1daae7e7df241e58f8e1aa4b1afef6 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Sat, 3 Oct 2026 08:53:07 -0700 Subject: [PATCH 1/2] gateway: allowance stays fail-closed on a cursorless snapshot; atomic snapshot rebuild (D257) A rebuild removed the snapshot, appended Puts, and wrote the cursor last, in place. A crash midway left some Puts and no cursor; the next boot with NATS down installed them as a READY allowance-set holding only part of the exhausted tenants, so the rest were served and /readyz said 200. - WatchedSet::SEED_FROM_CURSORLESS_SNAPSHOT: deny (fail-open) still seeds from any snapshot; allowance (fail-closed) seeds only from one with a cursor record, decided before anything is installed or the ready gauge flips. - rebuild_snapshot writes {path}.tmp, fsyncs it, renames over path, fsyncs the directory, and reopens path for appends. A stale .tmp is removed first; any failure aborts to snapshot-less. - beyond-slipstream 0.1.0 -> 0.8.0 (same v2 on-disk format, API compatible). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JimHGjsfk2Ktm5GxyZJKKk --- Cargo.lock | 26 +- crates/gateway/ARCHITECTURE.md | 17 +- crates/gateway/Cargo.toml | 2 +- crates/gateway/src/store_watch.rs | 261 ++++++++++++++++--- crates/gateway/tests/reliability_snapshot.rs | 194 ++++++++++++++ 5 files changed, 459 insertions(+), 41 deletions(-) create mode 100644 crates/gateway/tests/reliability_snapshot.rs diff --git a/Cargo.lock b/Cargo.lock index 696808e5..ab7f9a68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -542,15 +542,18 @@ dependencies = [ [[package]] name = "beyond-slipstream" -version = "0.1.0" +version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3b07e54aae9b02cf7d2e9d935bd99cbc4a045f19d00738f069f44ba238a01600" +checksum = "8fc9323367b81de1797666a11e983d7d8d7bda76f4e18860ab931e601207725b" dependencies = [ + "aho-corasick", "async-nats", "async-trait", "base64 0.22.1", + "blake3", "crc32fast", "futures", + "serde", "serde_json", "tempfile", "thiserror 2.0.18", @@ -589,6 +592,19 @@ dependencies = [ "digest", ] +[[package]] +name = "blake3" +version = "1.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae" +dependencies = [ + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures 0.3.0", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -905,6 +921,12 @@ dependencies = [ "tiny-keccak", ] +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + [[package]] name = "core-foundation" version = "0.9.4" diff --git a/crates/gateway/ARCHITECTURE.md b/crates/gateway/ARCHITECTURE.md index 76c181e9..aae5a5ac 100644 --- a/crates/gateway/ARCHITECTURE.md +++ b/crates/gateway/ARCHITECTURE.md @@ -243,6 +243,21 @@ snapshot exists so _enforcement_ survives a cold start before NATS reconnects. C enforcement — "captured nothing for the first few seconds after a restart" is a non-event — so `Capture::snapshot_path` returns `None`. +**A snapshot seeds a set by its fail direction** (`WatchedSet::SEED_FROM_CURSORLESS_SNAPSHOT`). +Every writer puts the cursor after the data it covers, so a snapshot with a cursor record (even the +revision 0 of an empty scan) is a complete read, and one without is not. Deny (fail-open) installs +whatever a snapshot holds, cursor or not, because any surviving deny entry is strictly better than +the empty set. Allowance (fail-closed) installs a snapshot only when it has a cursor; otherwise it +stays unseeded (503, `/readyz` 503) until a complete snapshot or a NATS scan, so a partial file can +never flip it ready while missing exhausted tenants (D257). Skipping the scan on connect is a +separate test for every set: only an `is_resumable` cursor (revision > 0) does that. + +**The rebuild after a scan is atomic.** `rebuild_snapshot` writes every entry and then the cursor to +`{path}.tmp`, fsyncs it, renames it over `path`, fsyncs the directory, and reopens `path` for +appends. A crash leaves either the old complete file or the new complete one, never `Put`s without +their cursor. A stale `{path}.tmp` from an earlier crash is removed first, and any failure (removal, +rename) aborts the rebuild so the gateway runs snapshot-less rather than append to a stale log. + Deny is **fail-open** on an unread store (empty map = allow). Allowance is **fail-closed** until `from_entries` has run (even on an empty scan): managed traffic gets a retryable `503` with `ai_rejections_total{reason="allowance_unavailable"}`. After seed, a NATS blip keeps the last-known @@ -2833,7 +2848,7 @@ already gone gets nothing, and a response that has started cannot be replaced. | Failure | What Actually Happens | Recovery | | ------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| NATS unreachable at boot | Deny-set starts empty (fail-open). Allowance is unread → managed traffic gets a retryable 503 (`Retry-After: 5`), fail-closed. Auth still works — keys from config. | Watcher reconnects; seeds from NATS or disk snapshot on connect. Allowance flips remaining-ok once `from_entries` runs (empty scan included). Until then `/readyz` is 503 so the pod receives no traffic. | +| NATS unreachable at boot | Deny-set starts empty (fail-open), or from its snapshot when `snapshot_path` is set (cursor or not). Allowance is unread unless `{snapshot_path}.allowance` carries a cursor → managed traffic gets a retryable 503 (`Retry-After: 5`), fail-closed. Auth still works — keys from config. | Watcher reconnects; seeds from NATS or disk snapshot on connect. Allowance flips remaining-ok once `from_entries` runs (empty scan included). Until then `/readyz` is 503 so the pod receives no traffic. | | NATS disconnects mid-run | Last-known deny-set and allowance-set stay active. New entries not applied until reconnect. Unready allowance (never seeded) keeps refusing managed traffic with a retryable 503. | Watcher reconnects (1s→30s exponential backoff, reset only after a watch that ran ≥30s — _connecting_ is not success, or a reachable NATS with a broken watch loops at 1 Hz forever) and resumes from the saved revision. Rescans instead when the seed found no entries, since revision 0 is not resumable — see above. | | NATS history compacted past snapshot cursor | `CursorExpired` → full re-scan from current NATS state. | After re-scan, new cursor set; delta watch resumes normally. | | Virtual key tampered or forged | Prefix matches (`bai_v1`) and Ed25519 verify fails → **401**, never BYO. No billing event. The error does not name which part of the token failed. | Client retries with a valid key. A public listener that forwarded a failed verify as BYO would open junk-auth egress while the rate guard had already exempted the token from the BYO aggregate. | diff --git a/crates/gateway/Cargo.toml b/crates/gateway/Cargo.toml index 0c6f7e03..1ee6b47e 100644 --- a/crates/gateway/Cargo.toml +++ b/crates/gateway/Cargo.toml @@ -22,7 +22,7 @@ workspace = true [dependencies] # slipstream is published — consume it from crates.io, aliased to `store` so the code's # `use store::...` is unchanged. No path deps into the `beyond` repo: this crate builds standalone. -store = { package = "beyond-slipstream", version = "0.1.0" } +store = { package = "beyond-slipstream", version = "0.8.0" } # The provider table (`route`'s `KNOWN_PROVIDERS`/`AuthScheme`/`Dialect`), shared with `crates/agent` # so the gateway's proxy routing and the agent's direct routing can never disagree about a provider's diff --git a/crates/gateway/src/store_watch.rs b/crates/gateway/src/store_watch.rs index 12cf5876..8e0d19e4 100644 --- a/crates/gateway/src/store_watch.rs +++ b/crates/gateway/src/store_watch.rs @@ -54,10 +54,10 @@ use async_trait::async_trait; use pingora_core::server::ShutdownWatch; use pingora_core::services::background::BackgroundService; use std::marker::PhantomData; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; -use store::snapshot::SnapshotWriter; +use store::snapshot::{SnapshotError, SnapshotWriter}; use store::{ Connection, KvEntry, KvError, KvStore, KvUpdate, NatsConnection, NatsConnectionConfig, StoreConfig, WatchCursor, @@ -137,6 +137,24 @@ pub trait WatchedSet: Send + Sync + 'static { /// scan lands — so it has nothing to gain from durable seeding and returns `None`, which makes /// every snapshot path below inert without needing to special-case it. fn snapshot_path(state: &GatewayState) -> Option; + + /// May a snapshot with **no cursor record** still seed this set at boot? + /// + /// Every writer puts the cursor *after* the data it covers (a rebuild after it has written every + /// entry, a batch after its deltas), so a snapshot that carries a cursor — even the revision 0 + /// of an empty scan — reflects a complete read. One without a cursor does not: the old in-place + /// rebuild crashed midway left some `Put`s and no cursor. Which way that cuts depends on the + /// set's fail direction: + /// + /// - **Fail-open** (deny): any surviving entry is strictly better than the empty set it would + /// otherwise boot with, so `true`. + /// - **Fail-closed** (allowance): installing it would flip the set *ready* holding only part of + /// the exhausted tenants, serving the rest and turning `/readyz` 200 (D257). So `false`: the + /// set stays unseeded (503) until a snapshot with a cursor or a NATS scan. + /// + /// This is about *seeding*, not resuming: only an [`is_resumable`] cursor (revision > 0) skips + /// the scan on connect, for every set. + const SEED_FROM_CURSORLESS_SNAPSHOT: bool; } /// The deny-set: spend/fraud holds under `blackhole.{tenant}` and `blackhole.key.{id}`. @@ -147,6 +165,7 @@ impl WatchedSet for Deny { const PREFIX: &'static str = "blackhole."; const NOUN: &'static str = "deny-set"; + const SEED_FROM_CURSORLESS_SNAPSHOT: bool = true; fn from_entries<'a>( _state: &GatewayState, @@ -188,6 +207,8 @@ impl WatchedSet for Capture { const PREFIX: &'static str = "aicapture."; const NOUN: &'static str = "capture-set"; + // Inert (no snapshot), and fail-open like deny if it ever gets one. + const SEED_FROM_CURSORLESS_SNAPSHOT: bool = true; fn from_entries<'a>( state: &GatewayState, @@ -265,6 +286,7 @@ impl WatchedSet for Allowance { const PREFIX: &'static str = "allowance."; const NOUN: &'static str = "allowance-set"; + const SEED_FROM_CURSORLESS_SNAPSHOT: bool = false; fn from_entries<'a>( _state: &GatewayState, @@ -389,18 +411,37 @@ impl BackgroundService for WatcherService { let load_path = path.clone(); match tokio::task::spawn_blocking(move || store::snapshot::load(&load_path)).await { Ok(Ok(Some(snap))) => { - let set = W::from_entries(&self.state, snap.entries.values()); - let count = W::len(&set); - info!(set = W::NOUN, count, "seeded from on-disk snapshot"); - W::record_size(&self.state, count); - W::slot(&self.state).store(Arc::new(set)); // A snapshot without a *resumable* cursor can't safely resume (a bare watch // would race), so only treat it as seeded when it carries a real resume point; // otherwise fall through to a NATS scan on connect. Note this must test // `is_resumable`, not `!is_none()`: a snapshot checkpointed after an empty scan // carries revision 0, which is a *present* cursor that slipstream still can't // resume from. - if is_resumable(&snap.cursor) { + // + // Separately, a fail-closed set must not go live from a snapshot that isn't + // proof of a complete read, i.e. one with no cursor record at all (D257, see + // `SEED_FROM_CURSORLESS_SNAPSHOT`). Decide before installing anything: + // `record_size` is the allowance ready flip, so it is skipped too. + let resumable = is_resumable(&snap.cursor); + let complete = !snap.cursor.is_none(); + if complete || W::SEED_FROM_CURSORLESS_SNAPSHOT { + let set = W::from_entries(&self.state, snap.entries.values()); + let count = W::len(&set); + info!( + set = W::NOUN, + count, complete, resumable, "seeded from on-disk snapshot" + ); + W::record_size(&self.state, count); + W::slot(&self.state).store(Arc::new(set)); + } else { + warn!( + set = W::NOUN, + entries = snap.entries.len(), + "on-disk snapshot has no cursor (not a complete read); a \ + fail-closed set stays unseeded until a NATS scan" + ); + } + if resumable { cursor = snap.cursor; seeded = true; } @@ -584,40 +625,21 @@ fn allowanceset_from_entries<'a>(entries: impl Iterator) -> set } -/// Rewrite the on-disk snapshot from a fresh scan: truncate, write one `Put` per live entry, and -/// checkpoint the cursor. Returns the reopened writer, or `None` if the rewrite failed (the gateway -/// then runs snapshot-less — the in-memory deny-set is unaffected). Synchronous file I/O, so it runs -/// on a blocking thread off the serving runtime. +/// Rewrite the on-disk snapshot from a fresh scan, atomically: write every live entry plus the +/// cursor to `{path}.tmp`, fsync it, rename it over `path`, fsync the directory, and reopen `path` +/// for appends. A crash at any point leaves either the old complete file or the new complete one. +/// Returns the reopened writer, or `None` if the rewrite failed (the gateway then runs +/// snapshot-less — the in-memory set is unaffected). Synchronous file I/O, so it runs on a blocking +/// thread off the serving runtime. async fn rebuild_snapshot( path: PathBuf, entries: Vec, cursor: WatchCursor, ) -> Option { - let res = tokio::task::spawn_blocking( - move || -> Result { - // Remove the old log so we don't replay a deleted-but-uncompacted key on a later load. - // A failed removal is *not* ignorable: if `SnapshotWriter::open` then appends to the - // surviving file, a compacted-away `Delete` can't undo its stale `Put`, and a later - // `load()` resurrects a tenant we no longer deny — the exact corruption this rebuild - // exists to prevent. `NotFound` is the expected, benign case (first boot, or scratch - // storage); any other error aborts the rebuild so we run snapshot-less rather than on - // poisoned state. - match std::fs::remove_file(&path) { - Ok(()) => {} - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => return Err(e.into()), - } - let mut w = SnapshotWriter::open(&path, SNAPSHOT_COMPACT_THRESHOLD)?; - // Consume `entries` — it was moved into this closure and `write_update` only borrows, - // so wrapping each entry in a `Put` by value costs nothing, where cloning it copied a - // `String` key and a `Vec` value (2 heap allocations) per scanned entry. - for e in entries { - w.write_update(&KvUpdate::Put(e))?; - } - w.checkpoint(&cursor)?; - Ok(w) - }, - ) + let res = tokio::task::spawn_blocking(move || { + let tmp = write_snapshot_tmp(&path, entries, &cursor)?; + commit_snapshot(&tmp, &path) + }) .await; match res { Ok(Ok(w)) => Some(w), @@ -632,6 +654,63 @@ async fn rebuild_snapshot( } } +/// `{path}.tmp`: where a rebuild is staged before it replaces `path`. +fn snapshot_tmp_path(path: &Path) -> PathBuf { + let mut tmp = path.as_os_str().to_owned(); + tmp.push(".tmp"); + PathBuf::from(tmp) +} + +/// First half of a rebuild: write the complete new snapshot (every entry, then the cursor) to +/// `{path}.tmp` and fsync it. `path` itself is untouched, so a crash anywhere in here leaves the +/// previous complete snapshot in place. (D257: the old in-place rebuild removed `path` first, so a +/// crash left some `Put`s and no cursor.) +fn write_snapshot_tmp( + path: &Path, + entries: Vec, + cursor: &WatchCursor, +) -> Result { + let tmp = snapshot_tmp_path(path); + // `SnapshotWriter::open` appends, so a temp file left by an earlier crash must go first, or its + // records would be replayed under ours. `NotFound` is the normal case; any other failure aborts. + match std::fs::remove_file(&tmp) { + Ok(()) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(e.into()), + } + let mut w = SnapshotWriter::open(&tmp, SNAPSHOT_COMPACT_THRESHOLD)?; + // Consume `entries` — it was moved in and `write_update` only borrows, so wrapping each entry + // in a `Put` by value costs nothing, where cloning it copied a key and a value per entry. + for e in entries { + w.write_update(&KvUpdate::Put(e))?; + } + // `checkpoint` flushes to the OS (its compaction hint is moot for a fresh file); the fsync + // below makes the file durable before the rename can expose it. + let _needs_compact = w.checkpoint(cursor)?; + drop(w); + std::fs::File::open(&tmp)?.sync_all()?; + Ok(tmp) +} + +/// Second half of a rebuild: rename the staged `tmp` over `path`, fsync the directory so the rename +/// itself is durable, and reopen `path` for appends. +/// +/// A failed rename is *not* ignorable: carrying on would append to the surviving old file, where a +/// `Put` whose `Delete` was compacted away can no longer be undone, and a later `load()` would +/// resurrect a tenant we no longer deny — the exact corruption the rebuild exists to prevent. So +/// any error here aborts and the gateway runs snapshot-less rather than on poisoned state. The +/// writer is opened on `path` *after* the rename, so it appends to the new inode, never to the one +/// renamed away. +fn commit_snapshot(tmp: &Path, path: &Path) -> Result { + std::fs::rename(tmp, path)?; + let dir = match path.parent() { + Some(d) if !d.as_os_str().is_empty() => d, + _ => Path::new("."), + }; + std::fs::File::open(dir)?.sync_all()?; + SnapshotWriter::open(path, SNAPSHOT_COMPACT_THRESHOLD) +} + async fn connect(state: &GatewayState) -> crate::error::Result> { let cfg = &state.config; // `expose().to_string()` lifts the creds out of our `Secret` into the plain `String` the store's @@ -1111,4 +1190,112 @@ mod tests { b.credit(RECONNECT_BACKOFF_MAX); assert_eq!(b.delay(), RECONNECT_BACKOFF_BASE); } + + /// A scratch directory for one snapshot test, removed on drop. + struct ScratchDir(PathBuf); + + impl ScratchDir { + fn new(tag: &str) -> Self { + let d = std::env::temp_dir().join(format!( + "beyond-ai-snapunit-{tag}-{}-{:?}", + std::process::id(), + std::thread::current().id() + )); + let _ = std::fs::remove_dir_all(&d); + std::fs::create_dir_all(&d).unwrap(); + Self(d) + } + } + + impl Drop for ScratchDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + fn put(key: &str, rev: u64) -> KvEntry { + KvEntry { + key: key.to_string(), + value: b"1".to_vec(), + version: VersionToken::from_u64(rev), + } + } + + fn keys(snap: &store::snapshot::Snapshot) -> Vec<&str> { + let mut k: Vec<&str> = snap.entries.keys().map(String::as_str).collect(); + k.sort_unstable(); + k + } + + fn full_rebuild(path: &Path, entries: Vec, rev: u64) -> SnapshotWriter { + let tmp = write_snapshot_tmp(path, entries, &WatchCursor::from_u64(rev)).unwrap(); + commit_snapshot(&tmp, path).unwrap() + } + + /// A crash after the new snapshot is staged but before the rename must leave the previous + /// complete snapshot, cursor included, not a partial one (the old in-place rebuild removed the + /// file first, so a crash left some `Put`s and no cursor). + /// claim: REL-13 + /// defect: D257 + #[test] + fn a_crash_before_the_rename_leaves_the_previous_complete_snapshot() { + let dir = ScratchDir::new("crash"); + let path = dir.0.join("snap.log"); + drop(full_rebuild( + &path, + vec![put("allowance.1", 1), put("allowance.2", 2)], + 2, + )); + + // "Crash" between the steps: stage the next rebuild and never commit it. + let tmp = write_snapshot_tmp( + &path, + vec![put("allowance.3", 9)], + &WatchCursor::from_u64(9), + ) + .unwrap(); + assert!(tmp.exists()); + + let snap = store::snapshot::load(&path).unwrap().unwrap(); + assert_eq!(keys(&snap), ["allowance.1", "allowance.2"]); + assert_eq!(snap.cursor.as_u64(), Some(2)); + assert!(is_resumable(&snap.cursor)); + } + + /// A staged file left by a crash is discarded, not appended to, by the next rebuild; and the + /// writer the rebuild returns appends to the renamed file, so later deltas survive a reload. + /// claim: REL-13 + /// defect: D257 + #[test] + fn a_rebuild_discards_a_stale_staged_file_and_reopens_the_new_one() { + let dir = ScratchDir::new("reopen"); + let path = dir.0.join("snap.log"); + // A crashed earlier rebuild left a staged file holding a key the next scan no longer has. + drop(write_snapshot_tmp( + &path, + vec![put("allowance.9", 1)], + &WatchCursor::from_u64(1), + )); + + let mut w = full_rebuild(&path, vec![put("allowance.1", 3)], 3); + assert!(!snapshot_tmp_path(&path).exists()); + w.write_update(&KvUpdate::Put(put("allowance.2", 4))) + .unwrap(); + let _ = w.checkpoint(&WatchCursor::from_u64(4)).unwrap(); + drop(w); + + let snap = store::snapshot::load(&path).unwrap().unwrap(); + assert_eq!(keys(&snap), ["allowance.1", "allowance.2"]); + assert_eq!(snap.cursor.as_u64(), Some(4)); + } + + /// The trait's policy, pinned: only the fail-closed set refuses a cursorless snapshot. + /// defect: D257 + #[test] + fn only_allowance_refuses_an_unresumable_snapshot() { + const { + assert!(Deny::SEED_FROM_CURSORLESS_SNAPSHOT); + assert!(!Allowance::SEED_FROM_CURSORLESS_SNAPSHOT); + } + } } diff --git a/crates/gateway/tests/reliability_snapshot.rs b/crates/gateway/tests/reliability_snapshot.rs new file mode 100644 index 00000000..963273fd --- /dev/null +++ b/crates/gateway/tests/reliability_snapshot.rs @@ -0,0 +1,194 @@ +//! Booting from the on-disk snapshots with NATS unreachable: which sets a snapshot may seed, and +//! that a fail-closed set stays fail-closed when the snapshot is not proof of a complete read. +//! +//! Hermetic: NATS is a closed port, and the snapshot files are written here before boot. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +mod common; + +use std::path::{Path, PathBuf}; + +use beyond_ai::key::{VirtualKey, mint}; +use common::*; +use store::snapshot::SnapshotWriter; +use store::{KvEntry, KvUpdate, VersionToken, WatchCursor}; + +const CHAT_PATH: &str = "/openai/v1/chat/completions"; +const CHAT: &str = r#"{"model":"gpt-4o","messages":[{"role":"user","content":"hi"}]}"#; + +fn vkey(sk: &ed25519_dalek::SigningKey, tenant_id: u64) -> String { + mint( + &VirtualKey { + tenant_id, + vpc_id: 7, + key_id: None, + }, + 1, + sk, + ) +} + +/// A per-test deny snapshot path; the allowance snapshot is `{path}.allowance`. +struct SnapPaths(PathBuf); + +impl SnapPaths { + fn new(tag: &str) -> Self { + let p = std::env::temp_dir().join(format!( + "beyond-ai-snap-{tag}-{}-{}.log", + std::process::id(), + free_port() + )); + let s = Self(p); + s.cleanup(); + s + } + fn deny(&self) -> &Path { + &self.0 + } + fn allowance(&self) -> PathBuf { + PathBuf::from(format!("{}.allowance", self.0.display())) + } + fn as_str(&self) -> String { + self.0.to_str().unwrap().to_string() + } + fn cleanup(&self) { + let _ = std::fs::remove_file(self.deny()); + let _ = std::fs::remove_file(self.allowance()); + } +} + +impl Drop for SnapPaths { + fn drop(&mut self) { + self.cleanup(); + } +} + +/// Write a snapshot holding one `Put` per key and, when `cursor` is `Some`, a resume cursor. +/// `None` is what a rebuild that crashed before its cursor record leaves behind. +fn write_snapshot(path: &Path, keys: &[(&str, &[u8])], cursor: Option) { + let mut w = SnapshotWriter::open(path, u64::MAX).unwrap(); + for (i, (key, value)) in keys.iter().enumerate() { + w.write_update(&KvUpdate::Put(KvEntry { + key: (*key).to_string(), + value: value.to_vec(), + version: VersionToken::from_u64(i as u64 + 1), + })) + .unwrap(); + } + match cursor { + Some(rev) => { + let _ = w.checkpoint(&WatchCursor::from_u64(rev)).unwrap(); + } + None => w.flush().unwrap(), + } +} + +async fn status(gw: &Gateway, key: &str) -> (u16, String) { + let resp = test_client() + .post(format!("{}{CHAT_PATH}", gw.url())) + .header("authorization", format!("Bearer {key}")) + .header("content-type", "application/json") + .body(CHAT) + .send() + .await + .unwrap(); + let s = resp.status().as_u16(); + (s, resp.text().await.unwrap_or_default()) +} + +/// An allowance snapshot without a cursor record is not proof of a complete read (a rebuild that +/// crashed midway leaves some exhausted tenants and no cursor). Booted with NATS down, the +/// allowance-set must stay unseeded: managed requests get 503 and `/readyz` is 503, rather than +/// serving every exhausted tenant the partial file is missing. +/// claim: REL-5, REL-13 +/// defect: D257 +#[tokio::test] +async fn an_allowance_snapshot_without_a_cursor_stays_fail_closed() { + let snap = SnapPaths::new("allow-nocursor"); + write_snapshot(&snap.allowance(), &[("allowance.2571", b"1")], None); + let (pubkey, sk) = test_keypair(57); + let mock = MockUpstream::start(Mode::Json).await; + let gw = Gateway::builder(closed_port(), &mock.authority(), &b64(&pubkey)) + .snapshot_path(snap.as_str()) + .skip_allowance_ready() + .start() + .await; + // Give the watcher time to load the snapshot (it would have flipped ready by now). + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + for tenant in [2571, 2572] { + let (s, body) = status(&gw, &vkey(&sk, tenant)).await; + assert_eq!( + s, 503, + "tenant {tenant}: a partial allowance snapshot seeded the set ({s} {body})" + ); + assert!(body.contains("allowance unavailable"), "{body}"); + } + let (ready, body) = gw.admin_get("/readyz").await; + assert_eq!( + ready, 503, + "readyz {ready} {body} from a cursorless snapshot" + ); + assert_eq!( + mock.hits(), + 0, + "an unseeded allowance-set must not reach upstream" + ); +} + +/// The same allowance snapshot with a resumable cursor is a complete read: it seeds the set and the +/// pod is ready before NATS is reachable. The exhausted tenant gets 402, the others are served. +/// claim: REL-5, REL-13 +/// defect: D257 +#[tokio::test] +async fn an_allowance_snapshot_with_a_cursor_seeds_ready() { + let snap = SnapPaths::new("allow-cursor"); + write_snapshot(&snap.allowance(), &[("allowance.2573", b"1")], Some(5)); + let (pubkey, sk) = test_keypair(58); + let mock = MockUpstream::start(Mode::Json).await; + let gw = Gateway::builder(closed_port(), &mock.authority(), &b64(&pubkey)) + .snapshot_path(snap.as_str()) + .start() + .await; + let (s, body) = status(&gw, &vkey(&sk, 2573)).await; + assert_eq!(s, 402, "the exhausted tenant from the snapshot: {body}"); + let (s, body) = status(&gw, &vkey(&sk, 2574)).await; + assert_eq!(s, 200, "a tenant absent from the snapshot: {body}"); + let (ready, body) = gw.admin_get("/readyz").await; + assert_eq!(ready, 200, "readyz {ready} {body}"); +} + +/// The deny-set is fail-open, so any entry surviving in a snapshot is strictly better than none: a +/// deny snapshot without a cursor still blocks its denied tenant while NATS is down. +/// claim: REL-13 +/// defect: D257 +#[tokio::test] +async fn a_deny_snapshot_without_a_cursor_still_blocks_with_nats_down() { + let snap = SnapPaths::new("deny-nocursor"); + write_snapshot(snap.deny(), &[("blackhole.2575", b"fraud")], None); + // A complete allowance snapshot, so the pod is ready and the deny check is what decides. It is + // the empty scan's revision 0: complete (a cursor record) but not resumable, which still seeds. + write_snapshot(&snap.allowance(), &[], Some(0)); + let (pubkey, sk) = test_keypair(59); + let mock = MockUpstream::start(Mode::Json).await; + let gw = Gateway::builder(closed_port(), &mock.authority(), &b64(&pubkey)) + .snapshot_path(snap.as_str()) + .start() + .await; + // The deny watcher loads its snapshot independently of the allowance one the builder waited + // for, so poll briefly. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + let (s, body) = loop { + let got = status(&gw, &vkey(&sk, 2575)).await; + if got.0 == 403 || std::time::Instant::now() > deadline { + break got; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + }; + assert_eq!( + s, 403, + "the denied tenant from a cursorless snapshot: {body}" + ); + let (s, body) = status(&gw, &vkey(&sk, 2576)).await; + assert_eq!(s, 200, "an undenied tenant: {body}"); +} From 5300210929ee83dbb6102b5c86383068228a7507 Mon Sep 17 00:00:00 2001 From: Jared Lunde Date: Sat, 3 Oct 2026 08:53:54 -0700 Subject: [PATCH 2/2] verify: D257 fixed (320db0e), tagged on REL-5 and REL-13 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01JimHGjsfk2Ktm5GxyZJKKk --- crates/gateway/src/store_watch.rs | 2 +- verify/claims.toml | 4 ++-- verify/defects.toml | 14 ++++++++++++++ 3 files changed, 17 insertions(+), 3 deletions(-) diff --git a/crates/gateway/src/store_watch.rs b/crates/gateway/src/store_watch.rs index 8e0d19e4..06b0da6d 100644 --- a/crates/gateway/src/store_watch.rs +++ b/crates/gateway/src/store_watch.rs @@ -1292,7 +1292,7 @@ mod tests { /// The trait's policy, pinned: only the fail-closed set refuses a cursorless snapshot. /// defect: D257 #[test] - fn only_allowance_refuses_an_unresumable_snapshot() { + fn only_allowance_refuses_a_cursorless_snapshot() { const { assert!(Deny::SEED_FROM_CURSORLESS_SNAPSHOT); assert!(!Allowance::SEED_FROM_CURSORLESS_SNAPSHOT); diff --git a/verify/claims.toml b/verify/claims.toml index 410247f0..c74c9189 100644 --- a/verify/claims.toml +++ b/verify/claims.toml @@ -1567,7 +1567,7 @@ touches = [ "crates/gateway/src/main.rs", "crates/gateway/src/config.rs", ] -defects = ["D11"] +defects = ["D11", "D257"] [[claim]] id = "REL-6" @@ -1712,7 +1712,7 @@ touches = [ "crates/gateway/src/main.rs", "crates/gateway/src/config.rs", ] -defects = [] +defects = ["D257"] [[claim]] id = "REL-14" diff --git a/verify/defects.toml b/verify/defects.toml index accb8dc3..c82c23cd 100644 --- a/verify/defects.toml +++ b/verify/defects.toml @@ -2541,3 +2541,17 @@ state = "fixed" test = "a_reclaimed_probe_releasing_late_does_not_mint_a_second_probe" fix = "4c9fa18" note = "Fixed lock-free in the same packed AtomicU64: while OPEN or HALF_OPEN the 14-bit failure field holds a probe generation, bumped on OPEN -> HALF_OPEN and on every reclaim (an opening breaker seeds it from the clock). allow() returns a Permit carrying it (a CLOSED permit carries none); release(permit), record_success_for and record_failure_for are no-ops in HALF_OPEN for any other generation, and release still saturates at half_open_permits. proxy.rs's ledger RequestCtx::breaker_pending is now Option. Unpermitted record_success/record_failure (the error-in-200 settle) are unchanged. Also tested: a_permit_from_another_generation_neither_releases_nor_decides, concurrent_stale_releases_never_mint_permits, test_a_stalled_probe_permit_is_reclaimed_after_the_reset_timeout (stale outcomes no longer decide)." + +[[defect]] +id = "D257" +severity = "med" +area = "Reliability" +title = "A snapshot rebuild that crashed midway booted the allowance-set READY with only some exhausted tenants" +impact = "Reliability and security: rebuild_snapshot removed the file, appended one Put per scanned entry and wrote the cursor last, in place. A crash midway left some Puts and no cursor. The next boot with NATS unreachable installed that partial file as a ready allowance-set: the exhausted tenants it was missing were served, and /readyz returned 200, breaking allowance's fail-closed guarantee." +confidence = "high" +evidence = "store_watch.rs: the boot path stored W::from_entries(snapshot) in the live slot and called record_size (the allowance ready flip) before checking is_resumable(&snap.cursor); allowanceset_from_entries starts from AllowanceSet::from_ready(); rebuild_snapshot did remove_file, SnapshotWriter::open, write_update per entry, checkpoint, with no temp file or rename." +claims = ["REL-5", "REL-13"] +state = "fixed" +test = "an_allowance_snapshot_without_a_cursor_stays_fail_closed" +fix = "320db0e" +note = "Two fixes. (1) WatchedSet::SEED_FROM_CURSORLESS_SNAPSHOT: deny (fail-open) still seeds from any snapshot; allowance (fail-closed) seeds only from a snapshot with a cursor record, decided before anything is installed. Every writer puts the cursor after the data it covers, so a cursor (even an empty scan's revision 0) means a complete read; only a revision > 0 cursor skips the scan, unchanged. (2) rebuild_snapshot now writes {path}.tmp, fsyncs it, renames over path, fsyncs the directory and reopens path, so a crash leaves the old or the new complete file. Gating on is_resumable instead (as first proposed) would leave an edge gateway with an empty allowance bucket unready with NATS down; e2e on_disk_snapshot_enforces_across_restart_without_nats caught that. Also tested: an_allowance_snapshot_with_a_cursor_seeds_ready, a_deny_snapshot_without_a_cursor_still_blocks_with_nats_down, a_crash_before_the_rename_leaves_the_previous_complete_snapshot, a_rebuild_discards_a_stale_staged_file_and_reopens_the_new_one, only_allowance_refuses_a_cursorless_snapshot."