From 36485ade2bd303e66618fc0ce4b6e89c0f44951f Mon Sep 17 00:00:00 2001 From: npub1jmc9dt2lyvzu3h0kxlwxt5zg4fxp9476awyxw6gwxn72g6cw7exqs64whm <96f056ad5f2305c8ddf637dc65d048aa4c12d7daeb8867690e34fca46b0ef64c@sprout-oss.stage.blox.sqprod.co> Date: Sat, 27 Jun 2026 00:15:09 -0400 Subject: [PATCH] feat(relay): emit read-row trace steps with (B) projection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Land the read-seam emitter for the runtime conformance gate. Two emit sites, one buzz-db helper, one negative fixture. ## buzz-db: `communities_of_channels` helper `Buzz::communities_of_channels(&[Uuid]) -> HashMap` — batched per-channel community lookup. Used by the relay emitters to project each row's true community label independently of the fetch query's WHERE clause. That independence is what makes the `Inv_NonInterference` / `Inv_ReadConfinement` bite non-vacuous: a mutation dropping `community_id = $X` from `query_events` would still let this helper return the row's true label and the checker would catch the mismatch. Channels missing from the result map are intentionally NOT mapped to a default — callers MUST treat "channel-id not in map" as a coverage breach, never as "use the resolved community." ## buzz-relay: projection + record helpers `crate::conformance` gains four new items: - `project_row_community` — single-row helper encoding the (B) strategy: channel-less → resolved (honest, not tautological); channel-scoped → lookup or `None` (caller fails closed). - `RowCommunityProjection` enum — Ok(Vec) OR MissingLookup discriminated outcome. - `record_read_message_rows` — non-search lane: emits `ReadMessageRows` on Ok projection, `ImplBug { kind: "row_community_lookup_missing" }` on MissingLookup. - `record_read_by_id_rows` — search lane companion, same shape but emits `ReadByIdRows`. `filter_channel` is `None` for the search lane (search at the abstract level isn't bound to a single channel; per-row `channel_id` carries channel identity honestly). ## req.rs wire-up: two emit sites - Site 1 (`req.rs` non-search loop, after `query_events`): collect distinct channel ids from the result set → `communities_of_channels` → `record_read_message_rows`. Production cost: one extra DB query per request (NoopTracer short-circuits in non-conformance builds). - Site 2 (`req.rs` search loop, after `get_events_by_ids`): same pattern → `record_read_by_id_rows`. `handle_search_req` gains a threaded-through `trace_state: Option<&AbstractState>` parameter. DB-helper errors on either site fall back to an empty lookup map. This intentionally triggers `MissingLookup` → `ImplBug` for any channel-scoped row in the result set, surfacing the helper failure as a coverage breach (fail-closed) rather than a silent resolved- label substitution. ## Negative fixture: foreign-row leak New `bad_foreign_row_leak.jsonl` + matching test. The fixture is a `ReadMessageRows` whose row_communities contains community B while the state is bound to community A. This is the proof artifact Eva requested for the (B)-projection guard-rail: if the row had been mis-projected as channel-less (defaulting to resolved A), the subset check would have passed vacuously. By recording the row's TRUE community independently, `Inv_NonInterference` surfaces it immediately as `NonInterference`. ## Unit tests (conformance::tests) Five new tests pinning every behavior: - `project_row_communities_channelless_uses_resolved` (positive) - `project_row_communities_channel_scoped_uses_lookup_label` (the non-tautological correctness — lookup label, NOT resolved) - `project_row_communities_channel_scoped_missing_is_breach` (the guard-rail bite) - `record_read_message_rows_missing_lookup_emits_impl_bug` - `record_read_by_id_rows_ok_emits_read_by_id_rows` ## Mutate → red → restore (three independent bites) 1. Make `project_row_community` fall back to resolved on missing- lookup (the tempting wrong-fix): `project_row_communities_channel_ scoped_missing_is_breach` + `record_read_message_rows_missing_ lookup_emits_impl_bug` go red with explicit messages ("missing lookup must be a breach, got Ok([...])", "expected ImplBug coverage breach, got ReadMessageRows {...}"). Restored. 2. Make every channel-scoped row project to resolved (the tautological projection): 4 of 9 unit tests go red — the lookup- label, missing-breach, and record-helper tests all bite. Restored. 3. Edit the negative fixture to use community_a instead of community_b: `foreign_row_leak_is_non_interference` reds ("foreign row community label must be rejected by Inv_NonInterference"). The fixture is load-bearing, not decorative. Restored. ## Test surfaces - `cargo test -p buzz-relay --lib` → **394/0** (was 387 baseline). - `cargo test -p buzz-conformance --lib` → 9/0. - `cargo test -p buzz-conformance --test replay_fixtures` → **6/0** (was 5; +foreign_row_leak). - `cargo test -p buzz-db --lib` → 75/0 (helper compiles; DB-driven integration coverage lives in `--include-ignored` lane on PG). - `cargo clippy -p buzz-relay -p buzz-db -p buzz-conformance --all-targets -- -D warnings` clean. - `cargo fmt --all -- --check` clean. Co-authored-by: Tyler Longwell Signed-off-by: Tyler Longwell --- .../tests/fixtures/bad_foreign_row_leak.jsonl | 1 + .../buzz-conformance/tests/replay_fixtures.rs | 40 +++ crates/buzz-db/src/lib.rs | 46 +++ crates/buzz-relay/src/conformance/mod.rs | 299 ++++++++++++++++++ crates/buzz-relay/src/handlers/req.rs | 76 +++++ 5 files changed, 462 insertions(+) create mode 100644 crates/buzz-conformance/tests/fixtures/bad_foreign_row_leak.jsonl diff --git a/crates/buzz-conformance/tests/fixtures/bad_foreign_row_leak.jsonl b/crates/buzz-conformance/tests/fixtures/bad_foreign_row_leak.jsonl new file mode 100644 index 000000000..8313562d4 --- /dev/null +++ b/crates/buzz-conformance/tests/fixtures/bad_foreign_row_leak.jsonl @@ -0,0 +1 @@ +{"schema_version":1,"action":{"type":"read_message_rows","channel":"cafe0000-0000-0000-0000-000000000010","row_communities":["bbbb0000-0000-0000-0000-000000000002"]},"state_after":{"resolved_community":"aaaa0000-0000-0000-0000-000000000001","bound_host":"a.example.test","actor":"0123456789abcdef"}} diff --git a/crates/buzz-conformance/tests/replay_fixtures.rs b/crates/buzz-conformance/tests/replay_fixtures.rs index 0d12b2437..b823a86a0 100644 --- a/crates/buzz-conformance/tests/replay_fixtures.rs +++ b/crates/buzz-conformance/tests/replay_fixtures.rs @@ -140,6 +140,32 @@ fn bad_coverage_breach_trace() -> Vec { )] } +/// A foreign-row trace: bound to community A but a `ReadMessageRows` +/// returns a row whose community label is community B. This is the +/// (B)-projection negative case Eva requested as the guard-rail for +/// "channel-scoped row masquerading as channel-less": IF the row had +/// been mis-projected as channel-less (and thus defaulted to the +/// resolved community A), the subset check would have passed +/// vacuously. By recording the row's TRUE community (B) — independent +/// of the fetch query's WHERE clause — the `Inv_NonInterference` / +/// `Inv_ReadConfinement` bite surfaces immediately as +/// `NonInterference`. This fixture is the proof artifact that the +/// projection helper's missing-lookup guard-rail is non-vacuous. +fn bad_foreign_row_leak_trace() -> Vec { + vec![TraceStep::new( + TraceAction::ReadMessageRows { + // The query was scoped to a channel in A (the host-resolved + // tenant). The relay's filter said "this row should belong + // to A." But the row's TRUE community is B — surfaced by + // the (B)-strategy projection reading the row's own + // `channel_id` against the channels table. + channel: Some(channel_in_a()), + row_communities: vec![community_b()], + }, + state_a(), + )] +} + // ---- Fixture round-trip ------------------------------------------------ fn fixture_path(name: &str) -> PathBuf { @@ -251,6 +277,20 @@ fn coverage_breach_is_caught() { ); } +#[test] +fn foreign_row_leak_is_non_interference() { + let trace = bad_foreign_row_leak_trace(); + assert_fixture_matches("bad_foreign_row_leak.jsonl", &trace); + + let scenario = Scenario::unstructured(trace); + let err = check_trace(&scenario) + .expect_err("foreign row community label must be rejected by Inv_NonInterference"); + assert!( + matches!(err, TransitionError::NonInterference { .. }), + "foreign row label must surface as NonInterference, got {err:?}" + ); +} + #[test] fn empty_trace_is_coverage_breach() { // Independent of the JSONL fixtures: the checker must fail closed on diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 697b568d2..499450807 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -312,6 +312,52 @@ impl Db { .transpose() } + /// Batched version of [`Self::community_of_channel`]: given a list of + /// channel UUIDs, returns a map from channel id → owning community + /// for every channel that exists (soft-deletes excluded). + /// + /// Used by the runtime conformance read-seam emitters in `buzz-relay`: + /// after a `query_events`/`get_events_by_ids` returns N rows, the + /// emitter collects distinct `channel_id`s, calls this once, then + /// projects each row's true community label independently of the + /// fetch query's WHERE clause. That independence is what makes the + /// `Inv_NonInterference` / `Inv_ReadConfinement` gate non-vacuous — + /// a mutation that dropped `community_id = $X` from the fetch query + /// would still let this helper return the row's true label, and the + /// checker would see the mismatch. + /// + /// Channels missing from the result map (deleted or never existed) + /// are intentionally not present rather than mapped to a default — + /// callers MUST treat "channel-id not in map" as a coverage breach, + /// never as "use the resolved community". + pub async fn communities_of_channels( + &self, + channel_ids: &[Uuid], + ) -> Result> { + if channel_ids.is_empty() { + return Ok(std::collections::HashMap::new()); + } + let rows = sqlx::query( + r#" + SELECT id, community_id + FROM channels + WHERE id = ANY($1) + AND deleted_at IS NULL + "#, + ) + .bind(channel_ids) + .fetch_all(&self.pool) + .await?; + + let mut out = std::collections::HashMap::with_capacity(rows.len()); + for row in rows { + let ch: Uuid = row.try_get("id")?; + let cm: Uuid = row.try_get("community_id")?; + out.insert(ch, CommunityId::from_uuid(cm)); + } + Ok(out) + } + /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate. pub async fn insert_event( &self, diff --git a/crates/buzz-relay/src/conformance/mod.rs b/crates/buzz-relay/src/conformance/mod.rs index f53f70de6..323d0aca0 100644 --- a/crates/buzz-relay/src/conformance/mod.rs +++ b/crates/buzz-relay/src/conformance/mod.rs @@ -165,6 +165,170 @@ pub fn record_req_authcheck( )); } +/// Project a row's true community label, independent of the fetch +/// query's WHERE clause. Encapsulates the (B) projection-strategy +/// guard-rail Eva specified: +/// +/// - If the row is **channel-scoped** (`row.channel_id == Some(ch)`): +/// look up `ch` in `channel_communities` (precomputed via +/// [`buzz_db::Buzz::communities_of_channels`]). On a hit, return the +/// looked-up label. On a miss, return `None` — the caller MUST treat +/// this as a coverage breach and fail closed. +/// - If the row is **channel-less** (`row.channel_id == None`): +/// project as the resolved community. Channel-less rows have no +/// independent per-channel community to look up — the projection is +/// honest for those rows, not a tautology (community-global rows are +/// genuinely tenant-scoped). +/// +/// The distinction "is this row channel-less?" comes from the row's +/// own `channel_id`, not from the query filter, so a channel-scoped +/// row CANNOT masquerade as channel-less to dodge the lookup. +fn project_row_community( + row_channel_id: Option, + resolved: &CommunityLabel, + channel_communities: &std::collections::HashMap, +) -> Option { + match row_channel_id { + None => Some(*resolved), + Some(ch) => channel_communities + .get(&ch) + .map(|cid| CommunityLabel::from_uuid(*cid.as_uuid())), + } +} + +/// Outcome of projecting a row set's community labels for the read +/// seam. Either a clean `Vec` (one per row, same order) +/// OR an [`TraceAction::ImplBug`] coverage breach if any channel-scoped +/// row's channel id was missing from the lookup map. +/// +/// Returning a discriminated outcome (rather than silently substituting +/// the resolved community on missing-lookup) is what keeps the gate +/// non-vacuous: a mutation that, say, soft-deletes a channel mid-query +/// would land here and surface as an `ImplBug`, not slip past as a +/// resolved-label projection. +#[derive(Debug)] +pub enum RowCommunityProjection { + /// One label per input row, in the same order. + Ok(Vec), + /// One or more channel-scoped rows had no entry in the lookup map. + /// Carries the seam-name + the first offending channel id for + /// debuggability; the checker treats `ImplBug` as a coverage breach. + MissingLookup { + /// Short stable tag identifying which seam projected without + /// a lookup (used as the `ImplBug.kind` value). + kind: &'static str, + /// First channel id whose lookup was missing — kept for log + /// debuggability, not consumed by the trace. + first_missing_channel: Uuid, + }, +} + +/// Project `row_communities` for a row set, applying the (B) strategy +/// guard-rail. +/// +/// `rows` is the list of `(channel_id_option)` per row in the result +/// set, in the order the relay will deliver them. +/// `channel_communities` is the result of +/// [`buzz_db::Buzz::communities_of_channels`] over the distinct +/// channel ids in `rows`. +pub fn project_row_communities( + rows: &[Option], + resolved: &CommunityLabel, + channel_communities: &std::collections::HashMap, +) -> RowCommunityProjection { + let mut out = Vec::with_capacity(rows.len()); + for row in rows { + match project_row_community(*row, resolved, channel_communities) { + Some(label) => out.push(label), + None => { + // `row` is Some(ch) (channel-less rows always project Some); + // the unwrap is the offending channel id. + let ch = row.expect("project_row_community returns None only for Some(ch)"); + return RowCommunityProjection::MissingLookup { + kind: "row_community_lookup_missing", + first_missing_channel: ch, + }; + } + } + } + RowCommunityProjection::Ok(out) +} + +/// Record a [`TraceAction::ReadMessageRows`] (non-search lane) or fail +/// closed with an `ImplBug` step if the row community projection hit a +/// missing-lookup. +/// +/// `filter_channel` is the channel filter the query was scoped to (or +/// `None` for a global query). Rows are presented as the row's own +/// `channel_id` value — independent of the filter — so a channel-scoped +/// row cannot evade the per-channel lookup. +pub fn record_read_message_rows( + tracer: &Arc, + state: &AbstractState, + filter_channel: Option, + rows: &[Option], + channel_communities: &std::collections::HashMap, +) { + let projection = project_row_communities(rows, &state.resolved_community, channel_communities); + match projection { + RowCommunityProjection::Ok(row_communities) => { + tracer.record(TraceStep::new( + TraceAction::ReadMessageRows { + channel: filter_channel.map(channel_label), + row_communities, + }, + state.clone(), + )); + } + RowCommunityProjection::MissingLookup { + kind, + first_missing_channel: _, + } => { + tracer.record(TraceStep::new( + TraceAction::ImplBug { + kind: kind.to_string(), + }, + state.clone(), + )); + } + } +} + +/// Search-lane companion to [`record_read_message_rows`]. Emits +/// [`TraceAction::ReadByIdRows`] for the per-hit refetch; same +/// projection + missing-lookup guard-rail. +pub fn record_read_by_id_rows( + tracer: &Arc, + state: &AbstractState, + filter_channel: Option, + rows: &[Option], + channel_communities: &std::collections::HashMap, +) { + let projection = project_row_communities(rows, &state.resolved_community, channel_communities); + match projection { + RowCommunityProjection::Ok(row_communities) => { + tracer.record(TraceStep::new( + TraceAction::ReadByIdRows { + channel: filter_channel.map(channel_label), + row_communities, + }, + state.clone(), + )); + } + RowCommunityProjection::MissingLookup { + kind, + first_missing_channel: _, + } => { + tracer.record(TraceStep::new( + TraceAction::ImplBug { + kind: kind.to_string(), + }, + state.clone(), + )); + } + } +} + /// RAII coverage-breach guard. Constructed at the top of any critical /// seam (currently: `ingest_event`); the guard observes a [`Tracer`] /// wrapper that counts emits. If the seam exits without any emit @@ -425,4 +589,139 @@ mod tests { other => panic!("expected AuthCheck action, got {other:?}"), } } + + /// Channel-less row projects as the resolved community — honest, not + /// tautological (community-global rows have no per-channel lookup). + #[test] + fn project_row_communities_channelless_uses_resolved() { + let resolved = CommunityLabel::from_uuid(Uuid::from_u128(0x42)); + let lookup = std::collections::HashMap::new(); + let rows = vec![None, None]; + + let projection = project_row_communities(&rows, &resolved, &lookup); + match projection { + RowCommunityProjection::Ok(labels) => { + assert_eq!(labels.len(), 2); + assert!( + labels.iter().all(|l| l == &resolved), + "all channel-less rows must project to resolved" + ); + } + RowCommunityProjection::MissingLookup { .. } => { + panic!("channel-less rows must not surface missing-lookup") + } + } + } + + /// Channel-scoped row with a foreign community in the lookup map + /// projects to the foreign label — independent of the fetch query + /// (this is the (B) strategy's non-tautological correctness). The + /// resolved community is NOT substituted. + #[test] + fn project_row_communities_channel_scoped_uses_lookup_label() { + use buzz_core::CommunityId; + let resolved = CommunityLabel::from_uuid(Uuid::from_u128(0xA)); + let foreign = Uuid::from_u128(0xF0); + let ch_id = Uuid::from_u128(0xC0); + let mut lookup = std::collections::HashMap::new(); + lookup.insert(ch_id, CommunityId::from_uuid(foreign)); + + let projection = project_row_communities(&[Some(ch_id)], &resolved, &lookup); + match projection { + RowCommunityProjection::Ok(labels) => { + assert_eq!(labels.len(), 1); + assert_eq!( + labels[0], + CommunityLabel::from_uuid(foreign), + "channel-scoped row must project to its OWN community, not resolved" + ); + assert_ne!(labels[0], resolved, "must not substitute resolved"); + } + other => panic!("expected Ok projection, got {other:?}"), + } + } + + /// The guard-rail bite: a channel-scoped row whose channel id is + /// absent from the lookup map MUST surface as `MissingLookup`, + /// never as a silent substitution to resolved. This is what makes + /// the negative fixture (channel-scoped foreign-community row + /// masquerading as channel-less) fail closed. + #[test] + fn project_row_communities_channel_scoped_missing_is_breach() { + let resolved = CommunityLabel::from_uuid(Uuid::from_u128(0xA)); + let ch_id = Uuid::from_u128(0xDEAD); + let lookup = std::collections::HashMap::new(); + + let projection = project_row_communities(&[Some(ch_id)], &resolved, &lookup); + match projection { + RowCommunityProjection::MissingLookup { + kind, + first_missing_channel, + } => { + assert_eq!(kind, "row_community_lookup_missing"); + assert_eq!(first_missing_channel, ch_id); + } + RowCommunityProjection::Ok(labels) => { + panic!("missing lookup must be a breach, got Ok({labels:?})") + } + } + } + + /// `record_read_message_rows` on a missing-lookup row must record + /// exactly one `ImplBug` step (not a `ReadMessageRows` with the + /// resolved label substituted). The checker treats `ImplBug` as a + /// coverage breach. + #[test] + fn record_read_message_rows_missing_lookup_emits_impl_bug() { + let typed = Arc::new(VecTracer::default()); + let inner: Arc = typed.clone(); + let state = dummy_state(); + let ch_id = Uuid::from_u128(0xDEAD); + let lookup = std::collections::HashMap::new(); + + record_read_message_rows(&inner, &state, Some(ch_id), &[Some(ch_id)], &lookup); + + let steps = typed.steps.lock().expect("vec tracer mutex"); + assert_eq!(steps.len(), 1); + match &steps[0].action { + TraceAction::ImplBug { kind } => { + assert_eq!(kind, "row_community_lookup_missing"); + } + other => panic!("expected ImplBug coverage breach, got {other:?}"), + } + } + + /// `record_read_by_id_rows` Happy path: channel-scoped row whose + /// lookup hits emits one `ReadByIdRows` with the per-row community + /// label, not the resolved one. + #[test] + fn record_read_by_id_rows_ok_emits_read_by_id_rows() { + use buzz_core::CommunityId; + let typed = Arc::new(VecTracer::default()); + let inner: Arc = typed.clone(); + let state = dummy_state(); + let foreign = Uuid::from_u128(0xF0); + let ch_id = Uuid::from_u128(0xC0); + let mut lookup = std::collections::HashMap::new(); + lookup.insert(ch_id, CommunityId::from_uuid(foreign)); + + record_read_by_id_rows(&inner, &state, None, &[Some(ch_id)], &lookup); + + let steps = typed.steps.lock().expect("vec tracer mutex"); + assert_eq!(steps.len(), 1); + match &steps[0].action { + TraceAction::ReadByIdRows { + channel, + row_communities, + } => { + assert!( + channel.is_none(), + "search lane uses None for filter_channel" + ); + assert_eq!(row_communities.len(), 1); + assert_eq!(row_communities[0], CommunityLabel::from_uuid(foreign)); + } + other => panic!("expected ReadByIdRows, got {other:?}"), + } + } } diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index dcf833d49..ebb60f7f9 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -213,6 +213,7 @@ pub async fn handle_req( &pubkey_bytes, &conn, &state, + trace_state.as_ref(), ) .await; return; @@ -283,6 +284,42 @@ pub async fn handle_req( } }; + // Conformance read-seam emit (non-search lane). Project each row's + // true community label via a per-channel lookup independent of the + // query's WHERE clause — see `record_read_message_rows` for the + // (B) projection strategy and the missing-lookup ImplBug + // guard-rail. Skipped silently if `trace_state` is `None` (only + // happens on malformed pubkey, a separate failure path). + if let Some(state_snap) = trace_state.as_ref() { + let row_channels: Vec> = + events.iter().map(|e| e.channel_id).collect(); + let distinct: Vec = { + let mut s: std::collections::BTreeSet = + std::collections::BTreeSet::new(); + for c in row_channels.iter().flatten() { + s.insert(*c); + } + s.into_iter().collect() + }; + let channel_communities = match state.db.communities_of_channels(&distinct).await { + Ok(m) => m, + Err(e) => { + warn!( + conn_id = %conn_id, sub_id = %sub_id, + "conformance row-community lookup failed: {e}" + ); + std::collections::HashMap::new() + } + }; + crate::conformance::record_read_message_rows( + &state.tracer, + state_snap, + per_filter_channel, + &row_channels, + &channel_communities, + ); + } + for stored in &events { // Per-filter NIP-01 matching — use the current filter only, not the // full filter set. OR semantics across filters are handled by the outer @@ -433,6 +470,7 @@ async fn handle_search_req( reader_pubkey_bytes: &[u8], conn: &ConnectionState, state: &AppState, + trace_state: Option<&crate::conformance::AbstractState>, ) { // The community-wide channel scope (no #h tag on the filter). `None` means // "no accessible channels and no global access" → EOSE, exactly as the @@ -559,6 +597,44 @@ async fn handle_search_req( } }; + // Conformance read-seam emit (search lane). Same (B) + // projection + missing-lookup guard-rail as the + // non-search path — see `record_read_by_id_rows`. The + // `filter_channel` is `None`: search at the abstract + // level isn't bound to a single channel filter, the + // per-row `channel_id` carries the channel identity + // honestly. + if let Some(state_snap) = trace_state { + let row_channels: Vec> = + events.iter().map(|e| e.channel_id).collect(); + let distinct: Vec = { + let mut s: std::collections::BTreeSet = + std::collections::BTreeSet::new(); + for c in row_channels.iter().flatten() { + s.insert(*c); + } + s.into_iter().collect() + }; + let channel_communities = + match state.db.communities_of_channels(&distinct).await { + Ok(m) => m, + Err(e) => { + warn!( + sub_id = %sub_id, + "conformance row-community lookup failed: {e}" + ); + std::collections::HashMap::new() + } + }; + crate::conformance::record_read_by_id_rows( + &state.tracer, + state_snap, + None, + &row_channels, + &channel_communities, + ); + } + let event_map: std::collections::HashMap<[u8; 32], &buzz_core::StoredEvent> = events .iter()