diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index b81b8fd..c85a597 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -205,6 +205,8 @@ This is the lesson of Saltzer, Reed & Clark, *End-to-End Arguments in System Des **Cursor authority covers rejected entries.** `batch_high` tracks the highest revision *received* since the last flush, including updates that `parse` rejected (corrupt bytes, irrelevant keys). A rejected entry is still "nothing to apply," so it is covered by the cursor — and because NATS delivers in revision order, advancing to the max revision after one atomic `apply` is sound: having seen the max means every revision below it has been seen too. Without this, a run of irrelevant keys would pin the cursor in place and force redundant replay on every restart. +The one exception: an update carrying the **unknown** version (an unparseable ACK subject on the hand-built multi-prefix consumer path) never touches `batch_high` — it can neither mint a cursor position nor clobber the real high from earlier in the batch. The update is still applied; skipping it under-advances at worst, and re-delivery on resume is idempotent. (The pre-guard code fabricated revision 0 for these, which the loop adopted as a real position — regressing the persisted cursor to 0 and forcing a full replay on the next restart.) + **Snapshot consistency.** Raw `KvUpdate`s stream to the snapshot log as they arrive, but the *checkpoint* cursor is the post-apply cursor. A crash after a raw record is written but before its `apply`/checkpoint leaves the log holding data *ahead* of its cursor — which is safe: the cursor never names a revision whose `apply` had not returned, so resume re-delivers and re-applies that tail rather than skipping it. Compaction runs off the hot path via `spawn_blocking`, as everywhere else in the snapshot subsystem. **Flush triggers.** A batch flushes when any of these fires: the `window` elapses, `batch.len()` reaches `config.max`, a shutdown is signalled, or the channel closes with a pending batch (the remainder is flushed before returning). @@ -347,7 +349,8 @@ Machine-checked as: _"bootstrap never silently diverges."_ Empirically pinned by | From | Trigger | Guard | What Actually Happens | |---|---|---|---| -| Watching | `rx.recv()` delivers a `KvUpdate` | — | `batch_high` updated; raw update pushed to `raw_batch`; `parse()` called; window timer armed on first update | +| Watching | `rx.recv()` delivers a `KvUpdate` | version not unknown | `batch_high` updated; raw update pushed to `raw_batch`; `parse()` called; window timer armed on first update | +| Watching | `rx.recv()` delivers an unknown-version `KvUpdate` | — | Same, except `batch_high` untouched — an unparseable revision neither mints nor clobbers a cursor position | | Batching | Window elapses | — | `apply(batch)`; cursor = `batch_high`; `store.apply(raw_batch, cursor)` on blocking task; `on_applied` fired | | Batching | `batch.len() == config.max` | — | Same as window flush | | Flush | `store.apply()` succeeds | streak < 16 | Streak reset; `on_applied` fired; continue watching | @@ -523,7 +526,7 @@ async-nats's bare `watch`/`watch_all`/`watch_many` ride `DeliverPolicy::New` — ### Why does the cursor-expired resync list keys instead of re-scanning values? -The fallback watch's re-list already carries every live key's value, so the resync only needs to learn which keys *no longer exist* — `reader.keys()` (headers only, no value bytes) is sufficient and cheap. The synthetic deletes carry an unknown version and never advance the cursor: the fold's persisted cursor stays at its (expired) position until real re-list revisions move it, so a crash mid-resync just re-runs the same idempotent diff on the next start. Ordering, not versioning, provides correctness: the resync acks before the fallback watch is established, so a synthetic delete can never land after the re-list put that resurrects the same key. +The fallback watch's re-list already carries every live key's value, so the resync only needs to learn which keys *no longer exist* — `reader.keys()` (headers only, no value bytes) is sufficient and cheap. The fold side of the diff is equally key-only: it streams via `for_each_in_range` rather than `range()`, so an All-scope resync against an on-disk backend never materializes the whole fold (values included) on the repair path. The synthetic deletes carry an unknown version and never advance the cursor: the fold's persisted cursor stays at its (expired) position until real re-list revisions move it, so a crash mid-resync just re-runs the same idempotent diff on the next start. Ordering, not versioning, provides correctness: the resync acks before the fallback watch is established, so a synthetic delete can never land after the re-list put that resurrects the same key. ### Why KvError: Clone instead of Box? @@ -545,7 +548,7 @@ Synadia Cloud returns `$JS.API.STREAM.CREATE` response shapes that `async-nats`' - `400` + "maximum number of streams" → Synadia at-limit but bucket may exist (non-fatal) - Anything else → hard failure -The standard async-nats path is tried first; raw is a fallback. +The standard async-nats path is tried first; raw is a fallback — taken on a parse-shaped error **or a timeout**. A timeout must not skip the fallback: Synadia Cloud, the deployment the raw path exists for, is also the most likely to be slow or distant, and the raw path's own per-op timeouts bound a genuinely dead connection anyway. ### Why subscribe-before-create in scan/keys? @@ -637,7 +640,8 @@ Production code and an exhaustive model checker running the same logic closes th | ------------------------------------ | --------------------------------------------------------------------------- | ------------------------------------------------- | | Transient `store.apply()` failure | Raw batch re-queued at front; next flush commits cumulatively; streak counter incremented | Automatic up to 16 consecutive failures | | 16 consecutive `store.apply()` failures | Watch fail-stops: `KvError::WatchError` returned | Restart re-folds from last good cursor; investigate store (disk, permissions) | -| Live watch retention overrun | Floor guard detects in-band (gapped delivery) or via 30 s backstop; watch fails with `KvError::WatchError` | Caller restarts; resume hits `CursorExpired`; resync repairs stale keys | +| Live watch retention overrun (All scope) | Floor guard detects in-band (gapped delivery) or via 30 s backstop; watch fails with `KvError::WatchError` | Caller restarts; resume hits `CursorExpired`; resync repairs stale keys | +| Live watch retention overrun (Prefix scope) | **Undetected until the next restart** — prefix watches deliver sparse revisions, so eviction gaps are indistinguishable from non-matching subjects client-side; the floor guard is sound for All scope only (a periodic probe here would false-positive into spurious resyncs, a design the model checker rejected) | Next restart's resume-window check hits `CursorExpired` → resync repairs; operate prefix-scoped watches with retention window comfortably above the restart/redeploy interval | | Resync `reader.keys()` failure | Watch fail-stops (fail-stop, not degrade — degrade semantics proven to leave stale keys permanently) | Restart re-runs full resume → expiry → resync | | `CursorExpired` | `watch_applied` lists live keys, diffs fold, applies synthetic deletes, then falls back to state-sync re-list | Automatic; raw callers use `stale_keys()` manually | | `WatchError` | Watch stream dropped (NATS restart, reconnect) | Re-subscribe | @@ -725,6 +729,7 @@ Credentials priority: `creds` > `creds_file` > URL-embedded > no auth. | -------- | ------- | -------------------------------------------------------------------------- | | `window` | 10 ms | Max time a batch stays open before flush | | `max` | 100 | Max updates per batch before early flush | +| `channel_capacity` | 256 | Depth of the watch-task → loop channel. A full channel backpressures the watch task (by design); during state-sync hydration of a large bucket it can become the effective batch boundary, so tune it together with `max` | | Constant | Value | Effect | |---|---|---| diff --git a/Cargo.lock b/Cargo.lock index ac0eaae..1ef38f5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -143,7 +143,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" [[package]] name = "beyond-slipstream" -version = "0.6.1" +version = "0.7.0" dependencies = [ "aho-corasick", "async-nats", diff --git a/Cargo.toml b/Cargo.toml index 0c5b052..2356021 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,7 +8,7 @@ rust-version = "1.92" [package] name = "beyond-slipstream" -version = "0.6.1" +version = "0.7.0" edition.workspace = true license.workspace = true rust-version.workspace = true diff --git a/benches/applied.rs b/benches/applied.rs index 06c1bb6..96e93d6 100644 --- a/benches/applied.rs +++ b/benches/applied.rs @@ -95,6 +95,7 @@ fn bench_batch_throughput(c: &mut Criterion) { BatchConfig { window: Duration::from_millis(10), max: 100, + ..BatchConfig::default() }, // parse: keep every put. |u: &KvUpdate| match u { diff --git a/src/applied.rs b/src/applied.rs index 65f7143..1429020 100644 --- a/src/applied.rs +++ b/src/applied.rs @@ -140,15 +140,25 @@ pub struct BatchConfig { pub window: Duration, /// Maximum number of parsed updates in a batch before forcing a flush. pub max: usize, + /// Capacity of the internal watch-task → main-loop channel. When the main + /// loop falls behind (slow `apply`, blocking store flush), a full channel + /// backpressures the watch task — that is the design — but during initial + /// state-sync hydration of a large bucket the channel can fill faster than + /// the window flushes, making *this* the effective batch boundary rather + /// than [`max`](Self::max). Tune it together with `max` for high-fanout + /// hydration; clamped to a minimum of 1. + pub channel_capacity: usize, } impl Default for BatchConfig { /// 10 ms / 100 updates — the de-facto default every hand-rolled caller - /// already used, lifted into one place. + /// already used, lifted into one place — and the 256-deep channel the + /// loop always allocated, now tunable. fn default() -> Self { Self { window: Duration::from_millis(10), max: 100, + channel_capacity: 256, } } } @@ -274,7 +284,7 @@ where // Spawn the watch task. It owns the cursor-expired fallback so the main loop // only ever sees a clean ordered stream of updates on `rx`. - let (tx, mut rx) = mpsc::channel::(256); + let (tx, mut rx) = mpsc::channel::(config.channel_capacity.max(1)); let handle = { let watcher = Arc::clone(&watcher); tokio::spawn( @@ -471,29 +481,33 @@ where let mut stale: Vec = Vec::new(); if let Some(st) = &store { for prefix in &scope_prefixes { - match st.range(prefix) { - Ok(entries) => stale.extend( - entries - .into_iter() - .filter(|e| !live.contains(e.key.as_str())) - .map(|e| e.key), - ), - Err(e) => { - // FATAL, not a degrade: an incomplete - // diff silently leaves deleted keys in - // the fold forever (tests/model.rs - // proves the divergence reachable - // under degrade semantics). Fail the - // watch; the restart re-runs the - // resume → expiry → resync from - // scratch. - warn!(error = %e, prefix = %prefix, - "resync fold range failed; aborting watch rather than diverging"); - handle.abort(); - return Err(KvError::WatchError(format!( - "cursor-expired resync failed listing fold prefix {prefix:?}: {e}" - ))); + // Stream the fold's keys rather than `range()`, + // which buffers every in-scope entry — values + // included — into one Vec. On an on-disk backend + // holding a fold larger than RAM (the case those + // backends exist for), an All-scope resync would + // materialize the entire fold on the repair + // path. Only the keys matter for the diff. + if let Err(e) = st.for_each_in_range(prefix, |entry| { + if !live.contains(entry.key.as_str()) { + stale.push(entry.key); } + Ok(()) + }) { + // FATAL, not a degrade: an incomplete + // diff silently leaves deleted keys in + // the fold forever (tests/model.rs + // proves the divergence reachable + // under degrade semantics). Fail the + // watch; the restart re-runs the + // resume → expiry → resync from + // scratch. + warn!(error = %e, prefix = %prefix, + "resync fold scan failed; aborting watch rather than diverging"); + handle.abort(); + return Err(KvError::WatchError(format!( + "cursor-expired resync failed listing fold prefix {prefix:?}: {e}" + ))); } } } @@ -595,8 +609,15 @@ where Some(u) => { // Cursor authority: every received update bumps the // pending high-water, regardless of whether `parse` - // keeps it. - batch_high = WatchCursor::from_version(u.version().clone()); + // keeps it — but only when it carries a real position. + // An unknown version (e.g. an unparseable ACK subject + // on the hand-built multi-prefix consumer path) must + // neither mint a fake cursor nor clobber the real high + // from earlier in the batch; skipping it under-advances + // at worst, and re-delivery on resume is idempotent. + if !u.version().is_unknown() { + batch_high = WatchCursor::from_version(u.version().clone()); + } // Buffer the raw update for the durable store fold (which // commits the whole batch + cursor atomically on flush). @@ -912,6 +933,10 @@ mod tests { unreachable!("resync only lists keys") } + async fn entry(&self, _key: &str) -> Result, KvError> { + unreachable!("resync only lists keys") + } + async fn keys(&self, prefix: &str) -> Result, KvError> { Ok(self .live @@ -1063,6 +1088,7 @@ mod tests { BatchConfig { window: Duration::from_secs(3600), // effectively never max, + ..BatchConfig::default() }, parse_put, move |batch: Vec>| f.lock().unwrap().push(batch.len()), @@ -1104,6 +1130,7 @@ mod tests { BatchConfig { window: Duration::from_secs(3600), // window won't fire max: 100, + ..BatchConfig::default() }, parse_put, move |_batch: Vec>| {}, @@ -1156,6 +1183,7 @@ mod tests { BatchConfig { window: Duration::from_secs(3600), max, + ..BatchConfig::default() }, parse_put, move |_batch: Vec>| { @@ -1223,6 +1251,55 @@ mod tests { assert_eq!(on_applied_max.load(Ordering::SeqCst), 7); } + /// An update carrying the UNKNOWN version (an unparseable ACK subject on + /// the hand-built multi-prefix consumer path) must neither mint a cursor + /// position nor clobber the real high-water from earlier in the batch. + /// Pre-guard behavior: `kv_message_to_update` fabricated revision 0 for + /// such updates and the unconditional `batch_high = ...` adopted it, + /// regressing the persisted cursor to 0. The update itself is still + /// applied — only the cursor ignores it. + #[tokio::test] + async fn unknown_version_update_does_not_move_or_clobber_cursor() { + let unknown_put = KvUpdate::Put(KvEntry { + key: "u".to_string(), + value: b"x".to_vec(), + version: VersionToken::unknown(), + }); + let updates = vec![put("a", b"1", 5), unknown_put]; + let watcher = Arc::new(MockWatcher::new(updates, false)); // close after + + let applied_batches = Arc::new(Mutex::new(Vec::>::new())); + let ab = Arc::clone(&applied_batches); + let (_sd_tx, sd_rx) = watch::channel(false); + + let cursor = watch_applied( + watcher, + WatchScope::All, + None, + None, // reader (no resync in this test) + None::, + None, + BatchConfig::default(), + parse_put, + move |batch: Vec>| ab.lock().unwrap().extend(batch), + move |_| {}, + sd_rx, + ) + .await + .unwrap(); + + assert_eq!( + cursor.as_u64(), + Some(5), + "the unknown-version update must not clobber the real batch high" + ); + assert_eq!( + *applied_batches.lock().unwrap(), + vec![b"1".to_vec(), b"x".to_vec()], + "the unknown-version update is still applied" + ); + } + /// A resume whose cursor has expired falls back to the full watch and still /// applies the delivered updates. #[tokio::test] @@ -1831,6 +1908,7 @@ mod tests { BatchConfig { window: Duration::from_secs(3600), // window never fires max: 100, + ..BatchConfig::default() }, parse_put, move |_batch: Vec>| {}, @@ -2100,6 +2178,7 @@ mod tests { BatchConfig { window: Duration::from_secs(3600), max: 1, // one update per flush → a compaction per update + ..BatchConfig::default() }, parse_put, move |_batch: Vec>| {}, @@ -2205,6 +2284,7 @@ mod tests { BatchConfig { window: Duration::from_millis(1), max: 1, + ..BatchConfig::default() }, parse_put, |_batch: Vec>| {}, @@ -2242,6 +2322,9 @@ mod tests { async fn get(&self, _key: &str) -> Result, KvError> { unreachable!("resync only lists keys") } + async fn entry(&self, _key: &str) -> Result, KvError> { + unreachable!("resync only lists keys") + } async fn keys(&self, _prefix: &str) -> Result, KvError> { Err(KvError::OperationFailed("injected listing failure".into())) } diff --git a/src/export_lease.rs b/src/export_lease.rs index 9933843..d9db933 100644 --- a/src/export_lease.rs +++ b/src/export_lease.rs @@ -219,6 +219,10 @@ impl LeaseGuard { /// Best-effort: a CAS conflict (someone already took over) or write error /// is logged, not surfaced — worst case the round waits out its ttl, which /// is the no-abandon behavior anyway. + /// + /// An unawaited `abandon()` does nothing (and would trip the [`Drop`] + /// warning); the future's inherent `#[must_use]` plus this crate's + /// `deny(unused_must_use)` make that a compile error, not a silent leak. pub async fn abandon(mut self) { self.resolved = true; match self @@ -247,6 +251,10 @@ impl LeaseGuard { /// Best-effort observability: a CAS conflict means the lease was already /// taken over (this round overran its ttl) and is logged, not surfaced — /// the artifact is already safe wherever the caller put it. + /// + /// An unawaited `complete()` records nothing (and would trip the [`Drop`] + /// warning); the future's inherent `#[must_use]` plus this crate's + /// `deny(unused_must_use)` make that a compile error, not a silent leak. pub async fn complete(mut self, cursor: &WatchCursor) -> Result<(), KvError> { self.resolved = true; self.record.completed_cursor_hex = Some(hex_encode(cursor.version().as_bytes())); diff --git a/src/kv.rs b/src/kv.rs index b943d5d..00b93ef 100644 --- a/src/kv.rs +++ b/src/kv.rs @@ -244,11 +244,17 @@ pub trait KvReader: Send + Sync { /// entries written by `delete_with_version`). Most callers should use /// `get()` instead, which filters tombstones for consistency with `scan()`. /// - /// Override in backends where tombstone version access is needed for - /// CAS conflict detection. - async fn entry(&self, key: &str) -> Result, KvError> { - self.get(key).await - } + /// REQUIRED (no default) — deliberately. A default delegating to `get()` + /// silently hid tombstones on any backend that forgot to override it, + /// which breaks CAS callers that need the tombstone's version: e.g. + /// [`ExportLease::try_acquire`](crate::ExportLease::try_acquire) reads an + /// abandoned (CAS-deleted) lease's version through `entry()` for its + /// takeover write — with a `get()` default it would see `None` and report + /// the round as live instead of stealing it. Backends without empty-value + /// tombstone semantics (where delete genuinely removes the key) should + /// implement this as a delegation to `get()` — explicitly, so the choice + /// is a reviewed decision rather than an inherited footgun. + async fn entry(&self, key: &str) -> Result, KvError>; } /// Watch capability - optional, not all stores support real-time updates. diff --git a/src/nats.rs b/src/nats.rs index d45292c..050bd92 100644 --- a/src/nats.rs +++ b/src/nats.rs @@ -439,39 +439,50 @@ impl NatsConnection { let max_bytes = config.max_bytes.unwrap_or(10 * 1024 * 1024); // Default 10MB for Synadia Cloud kv_config.max_bytes = max_bytes; - // Try normal create first, fall back to raw API if it fails (Synadia Cloud compatibility) - match timed(js.create_key_value(kv_config)).await? { - Ok(kv) => Ok(kv), - Err(e) => { + // Try normal create first, fall back to raw API if it fails (Synadia Cloud compatibility). + // + // A TIMEOUT also takes the fallback, not an early return: the raw path + // exists for Synadia Cloud, which is the deployment most likely to be + // slow or distant — `?`-propagating the timeout here would skip the + // exact workaround built for it. If the connection is genuinely dead, + // the raw path's own `timed()` bounds surface that promptly anyway. + match timed(js.create_key_value(kv_config)).await { + Ok(Ok(kv)) => return Ok(kv), + Ok(Err(e)) => { warn!( bucket = config.name, error = ?e, "create_key_value failed, trying raw JetStream API" ); - - // Try raw JetStream API as fallback - create_kv_bucket_raw( - client, - &config.name, - max_bytes, - history, - max_age_nanos, - config.num_replicas.unwrap_or(1), - ) - .await?; - - // Re-verify the bucket exists. This upholds the INVARIANT in - // `classify_raw_create_response`: the raw path reports `Created` - // on an unparseable response, so this round-trip is what actually - // confirms the bucket — do not remove it. - timed(js.get_key_value(&config.name)) - .await? - .map_err(|e| { - error!(bucket = config.name, error = ?e, "failed to get bucket after raw create"); - KvError::ConnectionFailed(format!("get bucket after raw create: {:?}", e)) - }) + } + Err(_) => { + warn!( + bucket = config.name, + timeout = ?KV_OP_TIMEOUT, + "create_key_value timed out, trying raw JetStream API" + ); } } + + // Try raw JetStream API as fallback + create_kv_bucket_raw( + client, + &config.name, + max_bytes, + history, + max_age_nanos, + config.num_replicas.unwrap_or(1), + ) + .await?; + + // Re-verify the bucket exists. This upholds the INVARIANT in + // `classify_raw_create_response`: the raw path reports `Created` + // on an unparseable response, so this round-trip is what actually + // confirms the bucket — do not remove it. + timed(js.get_key_value(&config.name)).await?.map_err(|e| { + error!(bucket = config.name, error = ?e, "failed to get bucket after raw create"); + KvError::ConnectionFailed(format!("get bucket after raw create: {:?}", e)) + }) } } @@ -1189,12 +1200,17 @@ struct NatsKvWatcher { /// surfaced, matching `kv::Watch`'s behavior. fn kv_message_to_update(msg: &async_nats::Message, kv_prefix: &str) -> Option { let key = msg.subject.strip_prefix(kv_prefix)?.to_string(); - let revision = msg + // An unparseable (or absent) ACK subject yields the UNKNOWN version, not a + // fabricated revision 0: `from_u64(0)` is a *parseable* position that + // `watch_applied` would adopt as its batch high-water — regressing the + // persisted cursor to 0 and forcing a full replay on the next restart. + // `unknown()` is the honest value; the cursor-authority loop skips it. + let version = msg .reply .as_deref() .and_then(stream_sequence_from_ack) - .unwrap_or(0); - let version = VersionToken::from_u64(revision); + .map(VersionToken::from_u64) + .unwrap_or_else(VersionToken::unknown); let operation = msg .headers .as_ref() @@ -1782,12 +1798,18 @@ mod tests { } #[test] - fn kv_message_without_reply_gets_revision_zero() { - // No ACK reply subject → revision unparseable → 0, the same "unknown - // version" convention scan() uses. + fn kv_message_without_reply_gets_unknown_version() { + // No ACK reply subject → revision unparseable → the UNKNOWN token, + // never a fabricated revision 0. A parseable `Some(0)` would be + // adopted by watch_applied as a real batch high-water and regress + // the persisted cursor to 0 (full replay on next restart); unknown + // is skipped by the cursor-authority loop instead. let msg = raw_kv_msg("$KV.certs.node.a", None, b"v", None); match kv_message_to_update(&msg, "$KV.certs.").expect("in keyspace") { - KvUpdate::Put(e) => assert_eq!(e.version.as_u64(), Some(0)), + KvUpdate::Put(e) => { + assert!(e.version.is_unknown()); + assert_eq!(e.version.as_u64(), None); + } other => panic!("expected Put, got {other:?}"), } }