From c1f791db5f1acf279047e8ebf6b3d4d7f69b01e6 Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Fri, 30 Oct 2026 23:03:08 -0500 Subject: [PATCH 1/3] id(waiting): add a total cap to the waiting pool (#1567) --- pkgs/id/src/event_store/ingest.rs | 29 ++++++ pkgs/id/src/event_store/mod.rs | 2 +- pkgs/id/src/event_store/waiting_rows.rs | 2 +- pkgs/id/src/waiting.rs | 121 ++++++++++++++++++++++-- 4 files changed, 144 insertions(+), 10 deletions(-) diff --git a/pkgs/id/src/event_store/ingest.rs b/pkgs/id/src/event_store/ingest.rs index 80916f843..bd086e209 100644 --- a/pkgs/id/src/event_store/ingest.rs +++ b/pkgs/id/src/event_store/ingest.rs @@ -2332,6 +2332,35 @@ mod tests { store.check_invariants().unwrap(); } + #[test] + fn test_a_smaller_total_on_reopen_drops_the_excess_and_still_opens() { + // Three authors each hold one waiting event under the default total. + // Reopening under a total of two must not brick the store: the excess + // row is dropped, the store opens, and the drop is durable. + let dir = tempfile::tempdir().unwrap(); + let mut ids = Vec::new(); + { + let mut store = reopen(&dir, WaitingLimits::default(), 0); + for n in 41..=43 { + let signer = key(n); + let events = chain_of(&signer, 2); + assert_eq!( + ingest(&mut store, &events[1]), + Ingest::Waiting { id: events[1].id } + ); + ids.push(events[1].id); + } + assert_eq!(store.waiting_ids().unwrap().len(), 3); + } + let limits = WaitingLimits::default().with_total(2); + let store = reopen(&dir, limits, 0); + assert_eq!(store.waiting_ids().unwrap().len(), 2); + assert_eq!(ids.iter().filter(|id| store.is_waiting(id)).count(), 2); + store.check_invariants().unwrap(); + let store = reopen(&dir, limits, 0); + assert_eq!(store.waiting_ids().unwrap().len(), 2); + } + #[test] fn test_a_parked_event_is_held_across_a_restart_and_still_sees_a_conflict() { let dir = tempfile::tempdir().unwrap(); diff --git a/pkgs/id/src/event_store/mod.rs b/pkgs/id/src/event_store/mod.rs index 65c4b60c4..e14a7b79d 100644 --- a/pkgs/id/src/event_store/mod.rs +++ b/pkgs/id/src/event_store/mod.rs @@ -1014,7 +1014,7 @@ impl EventStore { let event = row_event(&row)?; match pool.restore(row.author, row.id, row.deps, event, row.arrived_at) { Ok(()) => {} - Err(WaitRefusal::AuthorFull) => { + Err(WaitRefusal::AuthorFull | WaitRefusal::PoolFull) => { waiting_rows::delete(&txn, &row.id)?; dropped = true; } diff --git a/pkgs/id/src/event_store/waiting_rows.rs b/pkgs/id/src/event_store/waiting_rows.rs index d1bc0572e..c8eec76d3 100644 --- a/pkgs/id/src/event_store/waiting_rows.rs +++ b/pkgs/id/src/event_store/waiting_rows.rs @@ -259,7 +259,7 @@ impl Mirror<'_> { delete(self.txn, &id)?; Ok(Hold::Skipped) } - Err(WaitRefusal::AuthorFull) => { + Err(WaitRefusal::AuthorFull | WaitRefusal::PoolFull) => { delete(self.txn, &id)?; Ok(Hold::Full) } diff --git a/pkgs/id/src/waiting.rs b/pkgs/id/src/waiting.rs index b114c0f28..574645426 100644 --- a/pkgs/id/src/waiting.rs +++ b/pkgs/id/src/waiting.rs @@ -2,9 +2,12 @@ //! //! An event waits when a dependency has not arrived: a backlink, an `after` //! hash, or an auth dependency. The pool holds the event until every dependency -//! has arrived, then releases it. Two caps bound what a peer can make a node -//! hold: at most [`WaitingLimits::per_author`] waiting events per author, and a -//! drop once an event has waited [`WaitingLimits::max_age_ms`]. A dropped event +//! has arrived, then releases it. Three caps bound what a peer can make a node +//! hold: at most [`WaitingLimits::per_author`] waiting events per author, at +//! most [`WaitingLimits::total`] waiting events across all authors, and a drop +//! once an event has waited [`WaitingLimits::max_age_ms`]. The total cap is +//! the one that binds an attacker: author keys are self-certifying, so minting +//! authors is free and a per-author cap alone bounds nothing. A dropped event //! is not acknowledged, and sync sends it again once its dependency exists. //! //! The pool does no I/O and takes the clock as an explicit `now_ms`, so tests @@ -34,6 +37,12 @@ pub type AuthorKey = [u8; 32]; /// Waiting events allowed per author, unless a world sets its own. pub const DEFAULT_PER_AUTHOR: usize = 256; +/// Waiting events allowed across all authors, unless a world sets its own: +/// 32 authors' worth of the per-author cap. The per-author cap shapes one +/// author's share; this cap is the node's actual memory and disk bound, +/// because author keys are free to mint. +pub const DEFAULT_TOTAL: usize = 32 * DEFAULT_PER_AUTHOR; + const MS_PER_DAY: u64 = 24 * 60 * 60 * 1_000; /// Age after which a waiting event is dropped, unless a world sets its own: 7 days. @@ -42,31 +51,48 @@ pub const DEFAULT_MAX_AGE_MS: u64 = 7 * MS_PER_DAY; /// The caps a world sets on its waiting events. /// /// [`WaitingLimits::default`] gives the accepted values: 256 waiting events per -/// author, and a 7-day age cap. A world sets its own with [`WaitingLimits::new`]. +/// author, 8,192 across all authors, and a 7-day age cap. A world sets its own +/// with [`WaitingLimits::new`] and [`WaitingLimits::with_total`]. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct WaitingLimits { per_author: usize, + total: usize, max_age_ms: u64, } impl WaitingLimits { /// Caps for one world. `per_author` is the most waiting events one author may /// hold, and `max_age_ms` is how long an event may wait before it expires. + /// The total across all authors is [`DEFAULT_TOTAL`]; set it with + /// [`with_total`](Self::with_total). /// /// Zero is accepted. A `per_author` of zero refuses every event that would /// wait, and a `max_age_ms` of zero expires events at the next call. pub const fn new(per_author: usize, max_age_ms: u64) -> Self { Self { per_author, + total: DEFAULT_TOTAL, max_age_ms, } } + /// The same caps with a different total: the most waiting events the pool + /// may hold across all authors. A total of zero refuses every event that + /// would wait. + pub const fn with_total(self, total: usize) -> Self { + Self { total, ..self } + } + /// The most waiting events one author may hold. pub const fn per_author(self) -> usize { self.per_author } + /// The most waiting events the pool may hold across all authors. + pub const fn total(self) -> usize { + self.total + } + /// How long an event may wait before it expires, in milliseconds. pub const fn max_age_ms(self) -> u64 { self.max_age_ms @@ -97,6 +123,13 @@ pub enum WaitRefusal { /// The author already holds the per-author cap of waiting events. #[error("the author already holds the maximum number of waiting events")] AuthorFull, + /// The pool already holds the total cap of waiting events across all + /// authors. Distinct from [`AuthorFull`](Self::AuthorFull) so a caller can + /// tell a Sybil-shaped flood (many authors, pool full) from one author + /// over its own share; the event store's ingest surfaces both as its one + /// waiting-full refusal, because the remedy for the sender is the same. + #[error("the pool already holds the maximum number of waiting events")] + PoolFull, } /// One held event and the dependencies it still waits for. @@ -189,7 +222,8 @@ where /// # Errors /// /// [`WaitRefusal::AuthorFull`] when the event would wait and its author - /// already holds the per-author cap. + /// already holds the per-author cap, or [`WaitRefusal::PoolFull`] when the + /// pool already holds the total cap across all authors. pub fn insert( &mut self, author: AuthorKey, @@ -222,7 +256,9 @@ where /// /// # Errors /// - /// [`WaitRefusal::AuthorFull`] when the author already holds the per-author cap. + /// [`WaitRefusal::AuthorFull`] when the author already holds the per-author + /// cap, or [`WaitRefusal::PoolFull`] when the pool already holds the total + /// cap across all authors. pub fn park( &mut self, author: AuthorKey, @@ -253,7 +289,12 @@ where /// /// # Errors /// - /// [`WaitRefusal::AuthorFull`] when the author already holds the per-author cap. + /// [`WaitRefusal::AuthorFull`] when the author already holds the per-author + /// cap, or [`WaitRefusal::PoolFull`] when the pool already holds the total + /// cap across all authors. On restore-on-open the store drops a refused + /// row's event, exactly as it drops an expired one, so a store written + /// under a larger cap still opens under a smaller one: the earliest + /// arrivals are kept and the excess is dropped, never wedged. pub fn restore( &mut self, author: AuthorKey, @@ -280,7 +321,10 @@ where .map(|entry| &entry.event) } - /// Stores `event` under `key`, waiting for `missing`, under the author's cap. + /// Stores `event` under `key`, waiting for `missing`, under the caps. + /// The total cap is checked first: it is the node's resource bound, and + /// when the pool is full every further hold is refused whatever author it + /// names. fn hold( &mut self, author: AuthorKey, @@ -289,6 +333,9 @@ where event: E, now_ms: u64, ) -> Result<(), WaitRefusal> { + if self.len() >= self.limits.total() { + return Err(WaitRefusal::PoolFull); + } if self.waiting_count(author) >= self.limits.per_author() { return Err(WaitRefusal::AuthorFull); } @@ -424,6 +471,7 @@ mod tests { fn test_defaults_are_256_per_author_and_seven_days() { let limits = WaitingLimits::default(); assert_eq!(limits.per_author(), 256); + assert_eq!(limits.total(), DEFAULT_TOTAL); assert_eq!(limits.max_age_ms(), 604_800_000); } @@ -431,6 +479,8 @@ mod tests { fn test_limits_are_set_per_world() { let limits = WaitingLimits::new(2, 1_000); assert_eq!((limits.per_author(), limits.max_age_ms()), (2, 1_000)); + assert_eq!(limits.total(), DEFAULT_TOTAL); + assert_eq!(limits.with_total(7).total(), 7); let mut pool = WaitingPool::new(limits); pool.insert(author(1), 1, [10], "a", 0).unwrap(); pool.insert(author(1), 2, [10], "b", 0).unwrap(); @@ -666,4 +716,59 @@ mod tests { } } } + + #[test] + fn test_total_cap_refuses_across_authors_below_the_per_author_cap() { + // The (total+1)-th held event is refused even though every author is + // far below the per-author cap: minting authors does not mint room. + let limits = WaitingLimits::new(256, WEEK_MS).with_total(3); + let mut pool = WaitingPool::new(limits); + for n in 1..=3u8 { + pool.insert(author(n), u32::from(n), [100], "held", 0) + .unwrap(); + assert_eq!(pool.waiting_count(author(n)), 1); + } + assert_eq!(pool.len(), 3); + assert_eq!( + pool.insert(author(4), 4, [100], "refused", 0), + Err(WaitRefusal::PoolFull) + ); + assert_eq!(pool.len(), 3); + assert!(!pool.contains(&4), "a refused event is not held"); + } + + #[test] + fn test_total_cap_bounds_park_and_restore_too() { + // Every path into the pool funnels through the one hold, so the total + // cap binds parked and restored events as it binds inserts. + let limits = WaitingLimits::new(256, WEEK_MS).with_total(1); + let mut pool = WaitingPool::new(limits); + pool.park(author(1), 1, "parked", 0).unwrap(); + assert_eq!( + pool.insert(author(2), 2, [100], "refused", 0), + Err(WaitRefusal::PoolFull) + ); + assert_eq!( + pool.restore(author(2), 2, [100], "refused", 0), + Err(WaitRefusal::PoolFull) + ); + assert_eq!(pool.len(), 1); + } + + #[test] + fn test_restore_over_a_lowered_total_keeps_the_earliest_arrivals() { + // Rows written under a larger total, restored under a smaller one: + // the earliest arrivals are held and the excess is refused, so the + // store that owns the rows can drop them and open. + let limits = WaitingLimits::new(256, WEEK_MS).with_total(2); + let mut pool = WaitingPool::new(limits); + pool.restore(author(1), 1, [100], "first", 0).unwrap(); + pool.restore(author(2), 2, [100], "second", 1).unwrap(); + assert_eq!( + pool.restore(author(3), 3, [100], "excess", 2), + Err(WaitRefusal::PoolFull) + ); + assert_eq!(pool.len(), 2); + assert!(pool.contains(&1) && pool.contains(&2) && !pool.contains(&3)); + } } From 97386867d2a9e8047e15adb9b7fd30c687bf9f1a Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Fri, 30 Oct 2026 23:19:28 -0500 Subject: [PATCH 2/3] id(waiting): scope the middle store in the lowered-total reopen test (#1567) --- pkgs/id/src/event_store/ingest.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/pkgs/id/src/event_store/ingest.rs b/pkgs/id/src/event_store/ingest.rs index bd086e209..79dd44d94 100644 --- a/pkgs/id/src/event_store/ingest.rs +++ b/pkgs/id/src/event_store/ingest.rs @@ -2353,10 +2353,12 @@ mod tests { assert_eq!(store.waiting_ids().unwrap().len(), 3); } let limits = WaitingLimits::default().with_total(2); - let store = reopen(&dir, limits, 0); - assert_eq!(store.waiting_ids().unwrap().len(), 2); - assert_eq!(ids.iter().filter(|id| store.is_waiting(id)).count(), 2); - store.check_invariants().unwrap(); + { + let store = reopen(&dir, limits, 0); + assert_eq!(store.waiting_ids().unwrap().len(), 2); + assert_eq!(ids.iter().filter(|id| store.is_waiting(id)).count(), 2); + store.check_invariants().unwrap(); + } let store = reopen(&dir, limits, 0); assert_eq!(store.waiting_ids().unwrap().len(), 2); } From cd92cf3653b29b7d36a525e52369a0e084de2b21 Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Fri, 30 Oct 2026 23:54:55 -0500 Subject: [PATCH 3/3] docs(id): record dated design for waiting pool total cap (#1567) --- ...04-00-00Z_design_waiting_pool_total_cap.md | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 doc/2026-10-31T04-00-00Z_design_waiting_pool_total_cap/2026-10-31T04-00-00Z_design_waiting_pool_total_cap.md diff --git a/doc/2026-10-31T04-00-00Z_design_waiting_pool_total_cap/2026-10-31T04-00-00Z_design_waiting_pool_total_cap.md b/doc/2026-10-31T04-00-00Z_design_waiting_pool_total_cap/2026-10-31T04-00-00Z_design_waiting_pool_total_cap.md new file mode 100644 index 000000000..7c0f92eb8 --- /dev/null +++ b/doc/2026-10-31T04-00-00Z_design_waiting_pool_total_cap/2026-10-31T04-00-00Z_design_waiting_pool_total_cap.md @@ -0,0 +1,72 @@ +# Waiting Pool Total Cap — Bounding Memory and Disk Waiting Rows Against Sybil Floods + +- Date: 2026-10-31 +- Status: **implemented** +- Issue: [#1567](https://github.com/developing-today/code/issues/1567) +- Pull Request: [#2940](https://github.com/developing-today/code/pull/2940) + +## 1. Intent and Background + +In `pkgs/id`, an event waits in `WaitingPool` when a dependency has not yet arrived: +a backlink, an `after` hash, or an auth dependency. The pool holds the event in +memory and mirrors it to durable on-disk storage (`redb` `waiting_rows` table) +until its dependencies arrive or it expires. + +Prior to this change, `WaitingLimits` enforced only two caps: +1. `per_author`: maximum waiting events for a single author (default: 256) +2. `max_age_ms`: maximum age before an unreleased event is dropped (default: 7 days) + +While the module documentation claimed that these two caps bounded what a peer could +make a node hold, author keys are self-certifying 32-byte Ed25519 public keys. A peer +can generate arbitrary numbers of author keys at zero cost. Consequently, an attacker +could flood a node with waiting events under unique author identities, causing +unbounded memory growth and unbounded on-disk storage in `redb` across the 7-day +window. + +## 2. Total Cap Specification + +To establish a true resource bound, `WaitingLimits` gains a pool-wide total cap: + +- **Constant**: `pub const DEFAULT_TOTAL: usize = 32 * DEFAULT_PER_AUTHOR;` (8,192 events). +- **Rationale**: 8,192 events accommodates 32 concurrent authors fully saturating + their 256-event per-author quotas, providing ample headroom for legitimate multi-author + activity while strictly bounding memory footprint and on-disk waiting rows. +- **Configurability**: A world or embedder can adjust the cap via builder method: + - `WaitingLimits::with_total(self, total: usize) -> Self` + - `WaitingLimits::total(self) -> usize` + - `WaitingLimits::new(per_author, max_age_ms)` initializes with `total: DEFAULT_TOTAL`. + +## 3. Admission and Refusal Policy + +Admission is gated in `WaitingPool::hold()`: + +```rust +if self.len() >= self.limits.total() { + return Err(WaitRefusal::PoolFull); +} +if self.waiting_count(author) >= self.limits.per_author() { + return Err(WaitRefusal::AuthorFull); +} +``` + +- **Priority**: The total cap is checked **first**, prior to the per-author cap. + When the pool is saturated, all incoming waiting events are refused regardless of author. +- **Refusal Distinction**: `WaitRefusal::PoolFull` is introduced alongside `WaitRefusal::AuthorFull`. + This allows internal callers and telemetry to differentiate between single-author quota + exhaustion and pool-wide saturation (e.g. Sybil attacks). +- **Ingest Handling**: Ingest and sync paths treat both refusals identically: the event + is not acknowledged to the peer, leaving sync to retry once missing dependencies have + arrived. + +## 4. Restore-on-Open and Excess Row Pruning + +When opening an existing `EventStore`, waiting events stored in `redb` are restored into +`WaitingPool` in traversal order: + +- If a database was written under a higher cap (or an earlier un-capped release) and is + subsequently opened under a lower total cap, `pool.restore()` returns + `Err(WaitRefusal::PoolFull)` once the cap is reached. +- `EventStore::open` handles both `WaitRefusal::AuthorFull` and `WaitRefusal::PoolFull` + by durably deleting the excess row from `redb` (`waiting_rows::delete(&txn, &row.id)?`). +- Earlier arrivals are preserved; excess rows are pruned cleanly without wedging the store, + ensuring `check_invariants()` continues to hold.