From 77bcad7f26f82beb07d57020bf86c32b59576881 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:04:03 -0400 Subject: [PATCH 01/10] perf(runtime): coalesce resync, drain deferred-startup convergence, off-critical-path idempotency fsync Part of the #736 inbound/convergence boundary-contracts campaign (cheap correctness / non-blocking wins). - #637: drain take_pending_convergence_groups() after the deferred-startup replay loop so buffered convergence groups are scheduled before steady state instead of stranded until the next unrelated command/event. - #695: replace the per-event full transport resync with a routes-dirty flag + a single coalesced resync after the batch drains (both drain_pending_session_events and observe_account_device_effects). - #760: collect kind-448 push-gossip ids into a HashSet and retain once after the loop (O(n^2) -> O(n)). - #691: move SendIdempotencyStore::persist_to_disk off the async send hot path via spawn_blocking (at-least-once durability; inline fallback outside a runtime). Co-Authored-By: Claude Opus 4.8 --- crates/agent-connector/src/stream_session.rs | 25 ++++++++++++- crates/marmot-app/src/client/sync.rs | 36 +++++++++++++++---- .../marmot-app/src/runtime/account_worker.rs | 9 +++++ 3 files changed, 62 insertions(+), 8 deletions(-) diff --git a/crates/agent-connector/src/stream_session.rs b/crates/agent-connector/src/stream_session.rs index c2f6868a..dc026e2b 100644 --- a/crates/agent-connector/src/stream_session.rs +++ b/crates/agent-connector/src/stream_session.rs @@ -142,7 +142,30 @@ impl SendIdempotencyStore { inner.order.push_back(key); true }; - if should_persist && let Err(err) = self.persist_to_disk() { + if !should_persist { + return; + } + // #691: persist OFF the async send hot path. The in-memory entry recorded + // above already enforces first-write-wins for the lifetime of this process, + // so the (fs + double-fsync) disk write does not need to block the caller. + // Durability is therefore at-least-once: a crash after the send returns but + // before the spawned write lands can lose the on-disk record, so a retry + // after restart may re-send — acceptable because the in-process dedup covers + // the common retry window and the record only persists after a successful + // send. When called outside a tokio runtime (unit tests) we persist inline. + match tokio::runtime::Handle::try_current() { + Ok(_) => { + let store = self.clone(); + tokio::task::spawn_blocking(move || store.persist_to_disk_logged()); + } + Err(_) => self.persist_to_disk_logged(), + } + } + + /// Persist the current records to disk, logging (not propagating) any error. + /// Used by both the off-hot-path `spawn_blocking` write and the inline fallback. + fn persist_to_disk_logged(&self) { + if let Err(err) = self.persist_to_disk() { tracing::warn!( target: "agent_connector", method = "send_idempotency_persist", diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index 61b2ce6a..ac8a80e9 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/crates/marmot-app/src/client/sync.rs @@ -96,6 +96,7 @@ impl AppClient { // (see `schema_valid_message_ids`). let source_message_id_hex = String::new(); let source_recorded_at = unix_now_seconds(); + let mut routes_dirty = false; for event in &effects.events { let before = self.state.groups.len(); let previous_group = @@ -124,10 +125,16 @@ impl AppClient { &source_message_id_hex, ); if self.state.groups.len() != before { - self.refresh_group_routes()?; - self.sync_runtime_groups().await?; + routes_dirty = true; } } + // #695: coalesce. A batch with N membership-changing events previously ran + // a full transport resync (over ALL group subscriptions) once per such + // event; resync once after the batch drains instead. + if routes_dirty { + self.refresh_group_routes()?; + self.sync_runtime_groups().await?; + } self.app.save_state(&self.state)?; Ok(summary) } @@ -315,6 +322,11 @@ impl AppClient { .account_home() .account(&self.state.label)? .account_id_hex; + let mut routes_dirty = false; + // #760: collect push-gossip ids and strip them from `summary.messages` in + // ONE pass after the loop. The previous per-message `retain` was O(n) per + // gossip event → O(n²) over a batch a relay could flood with kind-448s. + let mut gossip_message_ids: HashSet = HashSet::new(); for event in &effects.events { let before = self.state.groups.len(); let previous_group = @@ -378,9 +390,7 @@ impl AppClient { "ignoring malformed push token gossip: {err}", ); } - summary - .messages - .retain(|candidate| candidate.message_id_hex != message.message_id_hex); + gossip_message_ids.insert(message.message_id_hex.clone()); continue; } if message.kind == MARMOT_APP_EVENT_KIND_CHAT @@ -475,8 +485,7 @@ impl AppClient { summary.projection_updates.push(projection_update); } if self.state.groups.len() != before { - self.refresh_group_routes()?; - self.sync_runtime_groups().await?; + routes_dirty = true; } if let cgka_traits::engine::GroupEvent::GroupStateChanged { group_id, @@ -518,6 +527,19 @@ impl AppClient { .set_group_self_membership(&self.state.label, &group_id_hex, false)?; } } + // #760: strip all collected push-gossip messages in one pass. + if !gossip_message_ids.is_empty() { + summary + .messages + .retain(|candidate| !gossip_message_ids.contains(&candidate.message_id_hex)); + } + // #695: coalesce the per-event full transport resync into one resync after + // the batch drains (was once per membership-changing event, each over ALL + // group subscriptions). + if routes_dirty { + self.refresh_group_routes()?; + self.sync_runtime_groups().await?; + } // Synthesize durable kind-1210 system rows from authenticated state // changes (peer commits, auto-commits, and scheduled convergence). let system_updates = self.project_group_system_rows(&effects.events, source_recorded_at); diff --git a/crates/marmot-app/src/runtime/account_worker.rs b/crates/marmot-app/src/runtime/account_worker.rs index a4b29934..74bb7b90 100644 --- a/crates/marmot-app/src/runtime/account_worker.rs +++ b/crates/marmot-app/src/runtime/account_worker.rs @@ -483,6 +483,15 @@ async fn run_app_runtime_account_worker( } } + // #637: mutations replayed during deferred startup (e.g. a queued SendMessage + // / InviteMembers) can buffer convergence groups. The steady-state arms below + // drain `take_pending_convergence_groups()` after every command/event, but the + // deferred-replay loop above does not — so schedule them here before entering + // the loop, otherwise buffered groups stay stranded until the next unrelated + // command/event (a liveness gap). `schedule_groups` is an idempotent set + // insert, so this is safe even when the loop buffered nothing. + scheduled_convergence.schedule_groups(client.take_pending_convergence_groups()); + let mut reconnect_backoff = AccountWorkerReconnectBackoff::default(); loop { From d509d208738e4b41314453970bee977722a738de Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:04:14 -0400 Subject: [PATCH 02/10] perf(cgka-engine): resolve transport_group_id via an in-memory index, drop pre-auth O(groups) scan (#740) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of the #736 campaign (boundary contract 4: transport routing). Unknown transport_group_ids previously forced a list_groups() + per-group MlsGroup::load scan BEFORE payload authentication — attacker-paced CPU/IO. Add an authoritative in-memory transport_group_id -> GroupId index, populated at hydrate_stable_groups_from_storage and at group create/join (transport_group_id is immutable per group), so resolution is O(1) and a miss returns the direct id with no scan. (engine.rs also introduces the #636 seen_message_ids_hex_cache field used later in distributed_convergence.rs.) Co-Authored-By: Claude Opus 4.8 --- crates/cgka-engine/src/engine.rs | 37 +++++++++++++++++++ crates/cgka-engine/src/group_lifecycle.rs | 18 +++++++++ .../src/message_processor/ingest.rs | 33 +++++++---------- 3 files changed, 68 insertions(+), 20 deletions(-) diff --git a/crates/cgka-engine/src/engine.rs b/crates/cgka-engine/src/engine.rs index 50b0ac61..42d96685 100644 --- a/crates/cgka-engine/src/engine.rs +++ b/crates/cgka-engine/src/engine.rs @@ -180,6 +180,29 @@ pub struct Engine { /// added by [`Self::quarantine_stored_group_on_hydrate`] and removed by a /// successful [`Self::retry_hydrate_quarantined_group`]. pub(crate) quarantined_groups: HashMap, + + /// Authoritative `transport_group_id -> GroupId` resolver for inbound + /// routing (#740). A Nostr-routed group's `transport_group_id` + /// (`nostr_group_id`) differs from its MLS group id, so the direct + /// `get_group` lookup in `group_id_for_transport_group_id` misses on the + /// normal production path; this index answers that in O(1) instead of the + /// former O(groups) scan that deserialized EVERY joined `MlsGroup` — a scan + /// that ran BEFORE payload authentication, so an unauthenticated peer could + /// flood unknown `transport_group_id`s to force attacker-paced CPU + storage + /// I/O. Populated for every group at hydration + /// ([`Self::hydrate_one_stored_group`]) and establishment (`do_create_group` + /// / `do_join_welcome`); a `transport_group_id` is immutable for a group's + /// lifetime (routing-component updates are out of scope), so no per-commit + /// maintenance is needed. If routing rotation is ever implemented, this + /// index MUST be updated at the rotation site. + pub(crate) transport_group_id_index: HashMap, GroupId>, + + /// #636: cached hex-encoded snapshot of `seen_message_ids` for the + /// convergence `CanonicalizationState`, tagged with the seen-set generation + /// it was built from. A convergence drain runs the pass up to 16× and + /// previously re-hex-encoded the whole (up to 100k-entry) set every pass; + /// this rebuilds only when the set actually changed. `None` until first use. + pub(crate) seen_message_ids_hex_cache: Option<(u64, std::collections::BTreeSet)>, } // ── Builder ───────────────────────────────────────────────────────────────── @@ -318,6 +341,8 @@ impl EngineBuilder { audit_operation_counter: 0, current_audit_context: None, quarantined_groups: HashMap::new(), + transport_group_id_index: HashMap::new(), + seen_message_ids_hex_cache: None, }) } } @@ -778,6 +803,18 @@ impl Engine { self.leaving_groups.remove(group_id); } + // #740: record this group's transport routing id so inbound resolution + // is an O(1) index hit rather than a pre-auth O(groups) MlsGroup-load + // scan. `mls_group` is owned here (not borrowing `self`), and the + // `provider` borrow is already released, so this direct field insert is + // disjoint from the storage/crypto borrows above. + if let Ok(transport_group_id) = + crate::app_components::transport_group_id_of_group(&mls_group) + { + self.transport_group_id_index + .insert(transport_group_id, group_id.clone()); + } + self.epoch_manager.set_stable(group_id.clone(), group.epoch); self.audit_group( group_id, diff --git a/crates/cgka-engine/src/group_lifecycle.rs b/crates/cgka-engine/src/group_lifecycle.rs index 5ce465f7..d382a463 100644 --- a/crates/cgka-engine/src/group_lifecycle.rs +++ b/crates/cgka-engine/src/group_lifecycle.rs @@ -223,6 +223,16 @@ impl Engine { required_capabilities: required_caps, }; self.storage.put_group(&group_record)?; + // #740: index this group's transport routing id for O(1) inbound + // resolution (see `Engine::transport_group_id_index`). Best-effort: a + // routing-read failure only forfeits the fast path (inbound would fall + // through to the unknown-group disposition), never fails creation. + if let Ok(transport_group_id) = + crate::app_components::transport_group_id_of_group(&mls_group) + { + self.transport_group_id_index + .insert(transport_group_id, group_id.clone()); + } // 5. Wrap welcomes via the peeler. // @@ -531,6 +541,14 @@ impl Engine { }; mirror_app_components_into_record(&mls_group, &mut group_record); self.storage.put_group(&group_record)?; + // #740: index this joined group's transport routing id for O(1) inbound + // resolution (see `Engine::transport_group_id_index`). + if let Ok(transport_group_id) = + crate::app_components::transport_group_id_of_group(&mls_group) + { + self.transport_group_id_index + .insert(transport_group_id, group_id.clone()); + } // Cache self's capabilities. Other members' capabilities arrive as // we ingest commits that touched their leaves; join-via-welcome diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index e605e983..0260ad9d 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -1362,26 +1362,19 @@ impl Engine { Err(err) => return Err(EngineError::Storage(err)), } - let provider = EngineOpenMlsProvider::::new(&self.crypto, self.storage.mls_storage()); - for group_id in self.storage.list_groups()? { - if group_id.as_slice() == transport_group_id { - return Ok(group_id); - } - let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice()); - let Some(mls_group) = MlsGroup::load( - as openmls_traits::OpenMlsProvider>::storage( - &provider, - ), - &mls_gid, - ) - .map_err(|e| EngineError::Backend(format!("load route candidate: {e:?}")))? - else { - continue; - }; - if crate::app_components::transport_group_id_of_group(&mls_group)? == transport_group_id - { - return Ok(group_id); - } + // #740: for Nostr-routed groups the `transport_group_id` (nostr_group_id) + // differs from the MLS group id, so the direct lookup above misses on the + // normal production path. Resolve via the authoritative in-memory index + // (built at hydration and group establishment; see + // `Engine::transport_group_id_index`) instead of the former O(groups) + // scan that deserialized every joined `MlsGroup`. That scan ran BEFORE + // payload authentication, so a peer flooding kind-445 events with unknown + // `transport_group_id`s forced attacker-paced CPU + storage I/O. A miss + // now costs one hash probe: the id is not one of our groups, so fall + // through to the direct id and let the unknown-group / NotForThisClient + // handling drop it without any per-event scan. + if let Some(group_id) = self.transport_group_id_index.get(transport_group_id) { + return Ok(group_id.clone()); } Ok(direct) From df8db4dfd6767a2ed984a3d305253ded63e83091 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:04:37 -0400 Subject: [PATCH 03/10] fix(recovery): weld the lag-recovery order into one AppEventReplayCursor; bound lag replay (#630 #609 #747) Part of the #736 campaign (boundary contracts 1 & 2: storage ordering + runtime recovery). - #630: introduce AppEventReplayCursor = (recorded_at, message_id_hex, insert_order) in storage-sqlite with a matching APP_EVENT_REPLAY_ORDER_ASC/_DESC SQL fragment, so the recovery query order, the subscription watermark, and recovery_row_is_pre_subscription suppression are ONE definition and cannot drift. The local insert_order is the third field because the cursor is a per-client recovery cut-point (not the cross-client display order) and it is load-bearing for unscoped/all-groups recovery where message_id_hex is unique only per group. insert_order is surfaced on StoredAppMessageRecord / AppMessageRecord (not FFI). The frozen legacy migration reader keeps its own COALESCE order. - #609/#747: bound agent-connector replay_missed_inbound to the newest DELIVERED_INBOUND_CURSOR_CAPACITY rows instead of limit:None (full history). Tests: storage app_messages order == cursor comparator (incl. unscoped cross-group insert_order case); runtime same-second + insert_order suppression. (marmot-app/lib.rs also adds display_name_from_directory_entry used later by #639; account_projection.rs also adds the #762 batched component load.) Co-Authored-By: Claude Opus 4.8 --- crates/agent-connector/src/inbound.rs | 12 +- crates/agent-connector/src/tests.rs | 1 + crates/cli/src/lib.rs | 1 + crates/marmot-app/src/conversions.rs | 1 + crates/marmot-app/src/lib.rs | 22 ++- crates/marmot-app/src/projection.rs | 5 + .../marmot-app/src/runtime/subscriptions.rs | 57 +++--- crates/marmot-app/src/runtime/tests.rs | 66 +++++-- crates/marmot-uniffi/src/conversions/media.rs | 2 + crates/marmot-uniffi/src/conversions/mod.rs | 1 + .../storage-sqlite/src/account_projection.rs | 170 ++++++++++++------ .../src/account_projection/tests.rs | 60 +++++++ crates/storage-sqlite/src/lib.rs | 4 +- 13 files changed, 306 insertions(+), 96 deletions(-) diff --git a/crates/agent-connector/src/inbound.rs b/crates/agent-connector/src/inbound.rs index b294e40b..d794830e 100644 --- a/crates/agent-connector/src/inbound.rs +++ b/crates/agent-connector/src/inbound.rs @@ -236,9 +236,19 @@ impl AgentConnector { }; let mut events = Vec::new(); for account in accounts { + // #609/#747: bound the replay to the newest `DELIVERED_INBOUND_CURSOR_CAPACITY` + // rows instead of `limit: None` (the entire per-account history). Lag + // recovery only needs to look back as far as the delivered cursor can + // dedup against (4096), and the broadcast ring that overflowed is far + // smaller (1024) — so a window of the cursor capacity comfortably covers + // anything that could have been dropped, while keeping replay O(window) + // instead of O(history) on the same task that must keep draining the + // broadcast channel. The `limit` path returns the newest N rows (see + // `SqliteAccountStorage::app_messages`), which is exactly the recent + // window a lagged subscription missed. let query = AppMessageQuery { group_id_hex: group_filter.map(str::to_owned), - limit: None, + limit: Some(crate::DELIVERED_INBOUND_CURSOR_CAPACITY), }; let records = self.runtime.messages_with_query(&account.label, query)?; for record in records { diff --git a/crates/agent-connector/src/tests.rs b/crates/agent-connector/src/tests.rs index aae8428f..cbf00785 100644 --- a/crates/agent-connector/src/tests.rs +++ b/crates/agent-connector/src/tests.rs @@ -2222,6 +2222,7 @@ fn received_chat_record( source_epoch: Some(1), recorded_at: 100, received_at: 100, + insert_order: 0, } } diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index b727fa93..e7c5e579 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -1707,6 +1707,7 @@ mod tests { source_epoch: None, recorded_at: 100 + u64::try_from(index / 2).unwrap(), received_at: 100 + u64::try_from(index / 2).unwrap(), + insert_order: i64::try_from(index).unwrap(), }) .collect::>(); diff --git a/crates/marmot-app/src/conversions.rs b/crates/marmot-app/src/conversions.rs index 59ea68fc..8de0c967 100644 --- a/crates/marmot-app/src/conversions.rs +++ b/crates/marmot-app/src/conversions.rs @@ -215,6 +215,7 @@ pub(crate) fn app_message_record_from_stored(record: StoredAppMessageRecord) -> source_epoch: record.source_epoch, recorded_at: record.recorded_at, received_at: record.received_at, + insert_order: record.insert_order, } } diff --git a/crates/marmot-app/src/lib.rs b/crates/marmot-app/src/lib.rs index 67cc6110..bf1cc165 100644 --- a/crates/marmot-app/src/lib.rs +++ b/crates/marmot-app/src/lib.rs @@ -507,6 +507,12 @@ pub struct AppMessageRecord { pub source_epoch: Option, pub recorded_at: u64, pub received_at: u64, + /// Local `app_events` insert order (rowid). The final LOCAL tiebreak of the + /// raw-event replay cursor used by lag-recovery watermark/suppression (#630); + /// not part of the cross-client display order. `#[serde(default)]` keeps + /// older serialized records readable. + #[serde(default)] + pub insert_order: i64, } #[derive(Clone, Debug, Default, PartialEq, Eq)] @@ -2154,8 +2160,20 @@ impl MarmotApp { &self, account_id_hex: &str, ) -> Result, AppError> { - if let Some(entry) = self.directory_entry_for_account_id(account_id_hex)? - && let Some(name) = display_name_for_profile(entry.profile.as_ref()) + let entry = self.directory_entry_for_account_id(account_id_hex)?; + self.display_name_from_directory_entry(account_id_hex, entry.as_ref()) + } + + /// Resolve a display name from an ALREADY-FETCHED directory entry, falling + /// back to a local account's label. Split out so callers that already hold + /// the entry (e.g. notification building, #639) don't re-query + /// `directory_entry_for_account_id`. + pub(crate) fn display_name_from_directory_entry( + &self, + account_id_hex: &str, + entry: Option<&UserDirectoryRecord>, + ) -> Result, AppError> { + if let Some(name) = display_name_for_profile(entry.and_then(|entry| entry.profile.as_ref())) { return Ok(Some(name)); } diff --git a/crates/marmot-app/src/projection.rs b/crates/marmot-app/src/projection.rs index 31e4ebe8..2c63a557 100644 --- a/crates/marmot-app/src/projection.rs +++ b/crates/marmot-app/src/projection.rs @@ -574,6 +574,11 @@ impl LegacyAccountProjectionDb { .and_then(|value| value.try_into().ok()), recorded_at: recorded_at.try_into().unwrap_or_default(), received_at: received_at.try_into().unwrap_or_default(), + // Frozen legacy migration reader (the old `messages` table): its + // records are re-inserted into `app_events` with fresh rowids by + // the one-time migration and are never used as a replay cursor, so + // the value here is immaterial. + insert_order: 0, }) }; let rows = match (&query.group_id_hex, query.limit) { diff --git a/crates/marmot-app/src/runtime/subscriptions.rs b/crates/marmot-app/src/runtime/subscriptions.rs index e654bbb0..f8f40e56 100644 --- a/crates/marmot-app/src/runtime/subscriptions.rs +++ b/crates/marmot-app/src/runtime/subscriptions.rs @@ -7,6 +7,7 @@ use std::sync::{Arc, Mutex as StdMutex}; use cgka_traits::app_event::MARMOT_APP_EVENT_KIND_AGENT_STREAM_START; use cgka_traits::{GroupId, MessageId}; use serde::{Deserialize, Serialize}; +use storage_sqlite::AppEventReplayCursor; use tokio::sync::{broadcast, mpsc, oneshot, watch}; use transport_quic_stream::AgentTextStreamCrypto; @@ -773,9 +774,12 @@ impl MarmotAppRuntime { // messages worth re-emitting (still deduped via `seen_message_ids`). // `None` (empty snapshot) means no pre-existing history, so recovery // emits everything as before. - let recovery_watermark: Option<(u64, String)> = snapshot - .last() - .map(|message| (message.recorded_at, message.message_id_hex.clone())); + let recovery_watermark: Option = + snapshot.last().map(|message| AppEventReplayCursor { + recorded_at: message.recorded_at, + message_id_hex: message.message_id_hex.clone(), + insert_order: message.insert_order, + }); let (updates_tx, updates_rx) = mpsc::channel(APP_RUNTIME_SUBSCRIPTION_BUFFER); tokio::spawn(async move { loop { @@ -808,7 +812,7 @@ impl MarmotAppRuntime { Ok(updates) => updates, Err(_) => continue, }; - for update in updates { + for (row_cursor, update) in updates { let message = update.message(); // Suppress pre-existing history at or below the // subscription watermark. Recovery reloads the full @@ -816,11 +820,12 @@ impl MarmotAppRuntime { // limited subscriber would receive every older row // as a bogus "live" update on the first lag. Only // messages strictly newer than the watermark are - // genuinely-missed live messages. + // genuinely-missed live messages. The watermark and + // `row_cursor` are the SAME `AppEventReplayCursor` the + // recovery query ordered by, so the cut is exact. if recovery_row_is_pre_subscription( recovery_watermark.as_ref(), - message.recorded_at, - &message.message_id_hex, + &row_cursor, ) { continue; } @@ -1340,25 +1345,26 @@ pub(crate) fn messages_recovery_query(query: &AppMessageQuery) -> AppMessageQuer /// watermark — i.e. it existed at subscription time and must NOT be re-emitted /// as a live update. /// -/// `watermark` is `(recorded_at, message_id_hex)` of the newest message that +/// `watermark` is the [`AppEventReplayCursor`] of the newest message that /// existed when the subscription was created (the last row of the ascending /// snapshot; `None` for an empty snapshot). Because lag recovery drops the /// caller's initial-replay `limit` and reloads the full group history (see /// #180), a limited subscriber would otherwise receive every older row as a -/// bogus live update on the first lag. Comparing on the canonical -/// `(recorded_at, message_id_hex)` key — the same order the store returns — -/// suppresses pre-subscription history while still admitting genuinely-new -/// messages (strictly greater than the watermark). An empty watermark means -/// there was no pre-existing history, so nothing is suppressed. +/// bogus live update on the first lag. Both the watermark and the compared row +/// are the SAME `AppEventReplayCursor` the store orders by (#630, #736 boundary +/// contract 2), so the suppression boundary can never disagree with the recovery +/// query order — even for an unscoped (all-groups) subscription, where the +/// `insert_order` tiebreak distinguishes two groups' rows that share +/// `(recorded_at, message_id_hex)`. A row at or below the watermark is +/// pre-existing history (suppress); strictly-greater rows are genuinely-new +/// (emit). An empty watermark means there was no pre-existing history, so +/// nothing is suppressed. pub(crate) fn recovery_row_is_pre_subscription( - watermark: Option<&(u64, String)>, - recorded_at: u64, - message_id_hex: &str, + watermark: Option<&AppEventReplayCursor>, + row: &AppEventReplayCursor, ) -> bool { match watermark { - Some((watermark_at, watermark_id)) => { - (recorded_at, message_id_hex) <= (*watermark_at, watermark_id.as_str()) - } + Some(watermark) => row <= watermark, None => false, } } @@ -1373,7 +1379,7 @@ fn messages_recovery_updates( account_id_hex: &str, account_label: &str, query: AppMessageQuery, -) -> Result, AppError> { +) -> Result, AppError> { let records = app.messages_with_query(account_label, query)?; let senders = records .iter() @@ -1385,12 +1391,23 @@ fn messages_recovery_updates( let updates = records .into_iter() .filter_map(|record| { + // Capture the replay cursor (incl. the LOCAL `insert_order`) before + // the record is consumed into a `RuntimeMessageUpdate`, so the + // watermark suppression compares the exact order the recovery query + // produced (#630). `insert_order` is deliberately not carried on the + // FFI-facing `ReceivedMessage`. + let cursor = AppEventReplayCursor { + recorded_at: record.recorded_at, + message_id_hex: record.message_id_hex.clone(), + insert_order: record.insert_order, + }; received_message_update_from_record( account_id_hex, account_label, record, &display_names, ) + .map(|update| (cursor, update)) }) .collect(); Ok(updates) diff --git a/crates/marmot-app/src/runtime/tests.rs b/crates/marmot-app/src/runtime/tests.rs index 2d525ed5..a2664d25 100644 --- a/crates/marmot-app/src/runtime/tests.rs +++ b/crates/marmot-app/src/runtime/tests.rs @@ -924,6 +924,7 @@ fn latest_agent_stream_start_accepts_mixed_case_filter() { source_epoch: None, recorded_at: 0, received_at: 0, + insert_order: 0, }], Some(&stream_id_hex.to_uppercase()), ) @@ -967,6 +968,7 @@ fn message_record(message_id_hex: &str, group_id_hex: &str, kind: u64) -> AppMes source_epoch: Some(7), recorded_at: 11, received_at: 12, + insert_order: 0, } } @@ -1095,39 +1097,67 @@ fn limited_subscription_recovery_suppresses_pre_subscription_history() { // reloads ALL five rows. Rows 10-50 are at/below the watermark and must be // suppressed; a genuinely-new row (60) arriving after subscription must be // emitted. - let watermark = Some((50_u64, "id50".to_owned())); + // #630/#736: the watermark and every compared row are the SAME + // `AppEventReplayCursor` the store orders by — `(recorded_at, message_id_hex, + // insert_order)` — so the suppression boundary can never disagree with the + // recovery query order. + use storage_sqlite::AppEventReplayCursor; + fn cur(recorded_at: u64, message_id_hex: &str, insert_order: i64) -> AppEventReplayCursor { + AppEventReplayCursor { + recorded_at, + message_id_hex: message_id_hex.to_owned(), + insert_order, + } + } + let watermark = Some(cur(50, "id50", 5)); + let wm = watermark.as_ref(); // Every pre-subscription row (including the watermark row itself) is - // suppressed — even the ones the limited snapshot never contained (10/20/30). - for (at, id) in [ - (10, "id10"), - (20, "id20"), - (30, "id30"), - (40, "id40"), - (50, "id50"), + // suppressed — even the ones the limited snapshot never contained (10/20/30), + // and a same-second row with a SMALLER id (existed at subscription time). + for row in [ + cur(10, "id10", 1), + cur(20, "id20", 2), + cur(30, "id30", 3), + cur(40, "id40", 4), + cur(50, "id40", 4), + cur(50, "id50", 5), ] { assert!( - recovery_row_is_pre_subscription(watermark.as_ref(), at, id), - "row ({at}, {id}) existed at subscription time and must be suppressed on recovery" + recovery_row_is_pre_subscription(wm, &row), + "row {row:?} at/below the watermark must be suppressed on recovery" ); } - // A genuinely-new post-subscription row is emitted. + // Genuinely-new rows strictly greater than the watermark are emitted: + // a later second, and a same-second row with a greater id. + assert!( + !recovery_row_is_pre_subscription(wm, &cur(60, "id60", 6)), + "a later-second message must be emitted" + ); assert!( - !recovery_row_is_pre_subscription(watermark.as_ref(), 60, "id60"), - "a message newer than the watermark is a real missed live update and must be emitted" + !recovery_row_is_pre_subscription(wm, &cur(50, "id99", 7)), + "same-second row with a greater id sorts after the watermark and must be emitted" ); - // Same-second tie-break: a row at the watermark timestamp but a larger - // message id sorts strictly after the watermark and must be emitted. + // Unscoped (all-groups) case: the same `message_id_hex` can appear in two + // groups at the same second (it is unique only per group). `insert_order` + // then distinguishes them: a later-inserted duplicate (strictly greater + // cursor) is emitted, an earlier one is suppressed. A two-field key could + // not tell these apart. + let dup_watermark = Some(cur(50, "dup", 5)); + assert!( + !recovery_row_is_pre_subscription(dup_watermark.as_ref(), &cur(50, "dup", 8)), + "a later-inserted same-(recorded_at,id) row is genuinely new and must be emitted" + ); assert!( - !recovery_row_is_pre_subscription(watermark.as_ref(), 50, "id99"), - "same-timestamp row with a greater id sorts after the watermark and must be emitted" + recovery_row_is_pre_subscription(dup_watermark.as_ref(), &cur(50, "dup", 3)), + "an earlier-inserted same-(recorded_at,id) row existed already and must be suppressed" ); // An empty snapshot has no watermark, so recovery suppresses nothing // (unchanged behavior for unlimited / empty-history subscriptions). - assert!(!recovery_row_is_pre_subscription(None, 10, "id10")); + assert!(!recovery_row_is_pre_subscription(None, &cur(10, "id10", 1))); } #[test] diff --git a/crates/marmot-uniffi/src/conversions/media.rs b/crates/marmot-uniffi/src/conversions/media.rs index a2f14c9e..7bd81dcb 100644 --- a/crates/marmot-uniffi/src/conversions/media.rs +++ b/crates/marmot-uniffi/src/conversions/media.rs @@ -339,6 +339,7 @@ mod tests { source_epoch: Some(7), recorded_at: 10, received_at: 11, + insert_order: 0, }; let records = media_records_ffi(vec![message]); @@ -451,6 +452,7 @@ mod tests { source_epoch: Some(7), recorded_at: 10, received_at: 11, + insert_order: 0, }; let from_list: Vec = media_records_ffi(vec![message]) diff --git a/crates/marmot-uniffi/src/conversions/mod.rs b/crates/marmot-uniffi/src/conversions/mod.rs index 2d0f3d87..30292a5d 100644 --- a/crates/marmot-uniffi/src/conversions/mod.rs +++ b/crates/marmot-uniffi/src/conversions/mod.rs @@ -341,6 +341,7 @@ mod tests { source_epoch: None, recorded_at: 10, received_at: 11, + insert_order: 0, }; let ffi = AppMessageRecordFfi::from(record); diff --git a/crates/storage-sqlite/src/account_projection.rs b/crates/storage-sqlite/src/account_projection.rs index de60cd39..330c539a 100644 --- a/crates/storage-sqlite/src/account_projection.rs +++ b/crates/storage-sqlite/src/account_projection.rs @@ -60,6 +60,70 @@ pub struct StoredAppMessageRecord { pub source_epoch: Option, pub recorded_at: u64, pub received_at: u64, + /// Local `app_events` insert order (rowid). The final, LOCAL tiebreak of the + /// raw-event replay ordering; see [`AppEventReplayCursor`]. Never used for + /// cross-client display order (that is the materialized-timeline surface). + pub insert_order: i64, +} + +impl StoredAppMessageRecord { + /// The raw-event replay cursor for this row (recovery ordering only). + pub fn replay_cursor(&self) -> AppEventReplayCursor { + AppEventReplayCursor { + recorded_at: self.recorded_at, + message_id_hex: self.message_id_hex.clone(), + insert_order: self.insert_order, + } + } +} + +/// Column list for [`SqliteAccountStorage::app_messages`], ending in +/// `insert_order` (column index 10, read by `app_message_from_row`). +const APP_EVENT_REPLAY_COLUMNS: &str = "message_id_hex, direction, group_id_hex, sender, plaintext, \ + kind, tags_json, source_epoch, recorded_at, received_at, insert_order"; + +/// The ONE ascending order for the raw-event replay surface (recovery / lag +/// replay), shared by [`SqliteAccountStorage::app_messages`] and — via +/// [`AppEventReplayCursor`]'s `Ord` — the runtime recovery watermark and +/// suppression, so the query order and the watermark cut-point can never drift +/// (#630, #736 boundary contract 1). This is the RAW-EVENT surface only: it is +/// NOT the materialized-timeline `(timeline_at, message_id_hex)` display order. +pub(crate) const APP_EVENT_REPLAY_ORDER_ASC: &str = "recorded_at, message_id_hex, insert_order"; +/// Descending variant of [`APP_EVENT_REPLAY_ORDER_ASC`] for the newest-first +/// `LIMIT` window that a bounded replay materializes before re-sorting ascending. +pub(crate) const APP_EVENT_REPLAY_ORDER_DESC: &str = + "recorded_at DESC, message_id_hex DESC, insert_order DESC"; + +/// Total order over the RAW-EVENT replay surface: `(recorded_at, message_id_hex, +/// insert_order)`. `insert_order` is a LOCAL rowid, which is correct here because +/// this cursor is only ever a per-client recovery cut-point (the lag-recovery +/// watermark + suppression), never the cross-client user-visible timeline order. +/// The third field is load-bearing for unscoped (all-groups) recovery: the same +/// `message_id_hex` can appear in two groups (it is unique only per group — e.g. +/// a sender posting identical content to two groups in the same second), so a +/// two-field cut-point could wrongly suppress a genuinely-new same-second row. +/// This is the single canonical comparator behind #630; keep it byte-identical +/// to [`APP_EVENT_REPLAY_ORDER_ASC`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AppEventReplayCursor { + pub recorded_at: u64, + pub message_id_hex: String, + pub insert_order: i64, +} + +impl Ord for AppEventReplayCursor { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.recorded_at + .cmp(&other.recorded_at) + .then_with(|| self.message_id_hex.cmp(&other.message_id_hex)) + .then_with(|| self.insert_order.cmp(&other.insert_order)) + } +} + +impl PartialOrd for AppEventReplayCursor { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } } #[derive(Clone, Debug, PartialEq, Eq)] @@ -208,9 +272,12 @@ impl SqliteAccountStorage { .storage()?; drop(group_statement); + let mut components_by_group = all_account_group_components(&conn)?; let mut groups = Vec::with_capacity(raw_groups.len()); for raw in raw_groups { - let components = account_group_components(&conn, &raw.group_id_hex)?; + let components = components_by_group + .remove(&raw.group_id_hex) + .unwrap_or_default(); groups.push(StoredAccountGroup { group_id_hex: raw.group_id_hex, endpoint: raw.endpoint, @@ -465,48 +532,34 @@ impl SqliteAccountStorage { &self, query: StoredAppMessageQuery, ) -> StorageResult> { + // Single-source the column list + replay ordering so the query order and + // the runtime recovery watermark/suppression (via `AppEventReplayCursor`) + // cannot drift (#630, #736). The limited variants take the newest-first + // `LIMIT` window, then re-sort ascending into replay order. + let cols = APP_EVENT_REPLAY_COLUMNS; + let asc = APP_EVENT_REPLAY_ORDER_ASC; + let desc = APP_EVENT_REPLAY_ORDER_DESC; let sql = match (&query.group_id_hex, query.limit) { - (Some(_), Some(_)) => { - "SELECT message_id_hex, direction, group_id_hex, sender, plaintext, - kind, tags_json, source_epoch, recorded_at, received_at - FROM ( - SELECT insert_order, message_id_hex, direction, group_id_hex, sender, - plaintext, kind, tags_json, source_epoch, recorded_at, received_at - FROM app_events + (Some(_), Some(_)) => format!( + "SELECT {cols} FROM ( + SELECT {cols} FROM app_events WHERE group_id_hex = ?1 - ORDER BY recorded_at DESC, message_id_hex DESC, insert_order DESC - LIMIT ?2 - ) - ORDER BY recorded_at, message_id_hex, insert_order" - } + ORDER BY {desc} LIMIT ?2 + ) ORDER BY {asc}" + ), (Some(_), None) => { - "SELECT message_id_hex, direction, group_id_hex, sender, plaintext, - kind, tags_json, source_epoch, recorded_at, received_at - FROM app_events - WHERE group_id_hex = ?1 - ORDER BY recorded_at, message_id_hex, insert_order" - } - (None, Some(_)) => { - "SELECT message_id_hex, direction, group_id_hex, sender, plaintext, - kind, tags_json, source_epoch, recorded_at, received_at - FROM ( - SELECT insert_order, message_id_hex, direction, group_id_hex, sender, - plaintext, kind, tags_json, source_epoch, recorded_at, received_at - FROM app_events - ORDER BY recorded_at DESC, message_id_hex DESC, insert_order DESC - LIMIT ?1 - ) - ORDER BY recorded_at, message_id_hex, insert_order" - } - (None, None) => { - "SELECT message_id_hex, direction, group_id_hex, sender, plaintext, - kind, tags_json, source_epoch, recorded_at, received_at - FROM app_events - ORDER BY recorded_at, message_id_hex, insert_order" + format!("SELECT {cols} FROM app_events WHERE group_id_hex = ?1 ORDER BY {asc}") } + (None, Some(_)) => format!( + "SELECT {cols} FROM ( + SELECT {cols} FROM app_events + ORDER BY {desc} LIMIT ?1 + ) ORDER BY {asc}" + ), + (None, None) => format!("SELECT {cols} FROM app_events ORDER BY {asc}"), }; let conn = self.lock()?; - let mut statement = conn.prepare(sql).storage()?; + let mut statement = conn.prepare(&sql).storage()?; let rows = match (&query.group_id_hex, query.limit) { (Some(group_id_hex), Some(limit)) => statement .query_map( @@ -1097,29 +1150,39 @@ fn checkpoint_wal_truncate_after_secure_prune(conn: &Connection) -> StorageResul }) } -fn account_group_components( +/// #762: load ALL account-group components in one ordered query, bucketed by +/// group in Rust, instead of an N+1 per-group query during full-projection load. +/// Ordered by `(group_id_hex, component_id)` so each group's components keep +/// `component_id` order (matching the prior per-group `ORDER BY component_id`). +fn all_account_group_components( conn: &rusqlite::Connection, - group_id_hex: &str, -) -> StorageResult> { +) -> StorageResult>> { let mut statement = conn .prepare( - "SELECT component_id, component_name, component_data_hex + "SELECT group_id_hex, component_id, component_name, component_data_hex FROM account_group_app_components - WHERE group_id_hex = ?1 - ORDER BY component_id", + ORDER BY group_id_hex, component_id", ) .storage()?; - statement - .query_map(params![group_id_hex], |row| { - Ok(StoredAccountGroupComponent { - component_id: i64_to_u16(row.get(0)?, 0)?, - component_name: row.get(1)?, - component_data_hex: row.get(2)?, - }) + let rows = statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + StoredAccountGroupComponent { + component_id: i64_to_u16(row.get(1)?, 1)?, + component_name: row.get(2)?, + component_data_hex: row.get(3)?, + }, + )) }) - .storage()? - .collect::, _>>() - .storage() + .storage()?; + let mut by_group: std::collections::HashMap> = + std::collections::HashMap::new(); + for row in rows { + let (group_id_hex, component) = row.storage()?; + by_group.entry(group_id_hex).or_default().push(component); + } + Ok(by_group) } fn delete_stale_group_components( @@ -1198,6 +1261,7 @@ fn app_message_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result(8)?.try_into().unwrap_or_default(), received_at: row.get::<_, i64>(9)?.try_into().unwrap_or_default(), + insert_order: row.get::<_, i64>(10)?, }) } diff --git a/crates/storage-sqlite/src/account_projection/tests.rs b/crates/storage-sqlite/src/account_projection/tests.rs index 5c1a759f..a23863ab 100644 --- a/crates/storage-sqlite/src/account_projection/tests.rs +++ b/crates/storage-sqlite/src/account_projection/tests.rs @@ -1395,3 +1395,63 @@ fn all_row_count(store: &SqliteAccountStorage, table: &str) -> i64 { }) .unwrap() } + +#[test] +fn app_messages_replay_order_matches_cursor_comparator() { + // #630/#736 boundary contract 1: the raw-event replay query order + // (`app_messages`) MUST equal the `AppEventReplayCursor` Rust comparator, so + // the recovery watermark/suppression and the recovery query can never drift. + // Covers the unscoped (all-groups) case where two groups share the same + // `(recorded_at, message_id_hex)` and only the local `insert_order` + // distinguishes them. + let store = SqliteAccountStorage::in_memory().unwrap(); + // Same second, ids inserted in NON-lexical order; plus a cross-group + // duplicate id at the same second; plus a later-second row. + store.record_app_event(&app_event("bbb", "aa", 50)).unwrap(); // insert_order 1 + store.record_app_event(&app_event("aaa", "aa", 50)).unwrap(); // insert_order 2 (smaller id, later insert) + // The two `dup` rows share `message_id_hex` (allowed: the app_events UNIQUE is + // per-(group, message_id)) but carry distinct globally-unique + // `source_message_id_hex` (distinct outer transport events), mirroring a + // sender posting identical content to two groups in the same second. + let mut dup_aa = app_event("dup", "aa", 50); + dup_aa.source_message_id_hex = Some("source-dup-aa".to_owned()); + let mut dup_bb = app_event("dup", "bb", 50); + dup_bb.source_message_id_hex = Some("source-dup-bb".to_owned()); + store.record_app_event(&dup_aa).unwrap(); // insert_order 3 + store.record_app_event(&dup_bb).unwrap(); // insert_order 4 (same id, other group) + store.record_app_event(&app_event("zzz", "aa", 60)).unwrap(); // insert_order 5 (later second) + + let rows = store + .app_messages(StoredAppMessageQuery { + group_id_hex: None, + limit: None, + }) + .unwrap(); + + // The SQL ORDER BY must equal sorting the same rows by the cursor comparator. + let mut by_cursor = rows.clone(); + by_cursor.sort_by_key(|r| r.replay_cursor()); + let key = |r: &StoredAppMessageRecord| (r.message_id_hex.clone(), r.group_id_hex.clone()); + assert_eq!( + rows.iter().map(key).collect::>(), + by_cursor.iter().map(key).collect::>(), + "app_messages SQL order must equal the AppEventReplayCursor comparator" + ); + + // Concrete order: same-second by id (aaa>(), + vec!["aaa", "bbb", "dup", "dup", "zzz"], + ); + let dups: Vec<_> = rows.iter().filter(|r| r.message_id_hex == "dup").collect(); + assert_eq!(dups.len(), 2); + assert!( + dups[0].insert_order < dups[1].insert_order, + "cross-group same-id rows are ordered by the local insert_order tiebreak" + ); + assert_eq!(dups[0].group_id_hex, "aa"); + assert_eq!(dups[1].group_id_hex, "bb"); +} diff --git a/crates/storage-sqlite/src/lib.rs b/crates/storage-sqlite/src/lib.rs index 80a7d9a0..65b9ecca 100644 --- a/crates/storage-sqlite/src/lib.rs +++ b/crates/storage-sqlite/src/lib.rs @@ -18,8 +18,8 @@ mod timeline; pub use account_projection::{ AccountGroupPushToken, AccountNotificationSettings, AccountPushRegistration, - AccountStoredPushRegistration, StoredAccountGroup, StoredAccountGroupComponent, - StoredAccountState, StoredAppMessageQuery, StoredAppMessageRecord, + AccountStoredPushRegistration, AppEventReplayCursor, StoredAccountGroup, + StoredAccountGroupComponent, StoredAccountState, StoredAppMessageQuery, StoredAppMessageRecord, }; pub use chat_list::{ AccountUnreadTotal, ChatListAvatar, ChatListMessagePreview, ChatListQuery, ChatListRow, From 383999706bf9a4de887fb4035d35d69c74cebec2 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:04:53 -0400 Subject: [PATCH 04/10] perf(storage,adapter,engine): incremental/bounded convergence & projection cleanups (#636 #698 #752 #704 #750 #639 #761 #762) Part of the #736 campaign (boundary contracts 1 & 4 + incremental recompute). - #636: BoundedIdSet generation counter + engine seen_message_ids_hex_cache, so the convergence hex snapshot is re-encoded once per change, not per (up to 16) pass. - #698/#752: transport-nostr-adapter by_transport_group index + CanonicalEndpoint (cached parsed RelayUrl), rebuilt at activate/sync_groups/deactivate; routes_for is O(matching groups) with no double-parse and a constant-time miss. - #704: unread_summary_tx derives count + first-unread + mention_count from ONE ordered scan. - #750: chat_list_projection_meta marker (migration 0022) skips the per-group mention recompute on warm-up once counts are reconciled. - #639: notification_user fetches the directory entry once (display_name derived via display_name_from_directory_entry) instead of a duplicate lookup. - #761: public_directory_users batched into two queries (was 2N+1). - #762: account group components loaded in one bucketed query (was N+1). - Timeline display order consolidated into TIMELINE_ORDER_BY_ASC/_DESC, kept distinct from the raw-event replay cursor. Co-Authored-By: Claude Opus 4.8 --- crates/cgka-engine/src/bounded_id_set.rs | 40 ++++++ .../src/distributed_convergence.rs | 30 +++- crates/marmot-app/src/notifications.rs | 11 +- crates/storage-sqlite/src/chat_list.rs | 130 +++++++++--------- crates/storage-sqlite/src/migrations.rs | 7 + .../0022_chat_list_projection_version.rs | 30 ++++ crates/storage-sqlite/src/shared.rs | 71 ++++++++-- crates/storage-sqlite/src/timeline.rs | 22 ++- crates/transport-nostr-adapter/src/lib.rs | 119 +++++++++++++--- 9 files changed, 350 insertions(+), 110 deletions(-) create mode 100644 crates/storage-sqlite/src/migrations/0022_chat_list_projection_version.rs diff --git a/crates/cgka-engine/src/bounded_id_set.rs b/crates/cgka-engine/src/bounded_id_set.rs index 170ac30a..a9b89cb0 100644 --- a/crates/cgka-engine/src/bounded_id_set.rs +++ b/crates/cgka-engine/src/bounded_id_set.rs @@ -41,6 +41,11 @@ pub(crate) struct BoundedIdSet { members: HashSet, order: VecDeque, capacity: usize, + /// Monotonic counter bumped on every membership change (insert or the + /// eviction that an insert triggers). Lets callers cache a derived view of + /// the set (e.g. the convergence hex-encoded snapshot, #636) and rebuild it + /// only when the set actually changed, instead of once per convergence pass. + generation: u64, } impl BoundedIdSet { @@ -54,6 +59,7 @@ impl BoundedIdSet { members: HashSet::new(), order: VecDeque::new(), capacity, + generation: 0, } } @@ -71,6 +77,10 @@ impl BoundedIdSet { if !self.members.insert(value.clone()) { return; } + // Membership changed (new id, plus any eviction below) — bump once so a + // no-op re-insert above leaves the generation (and any derived cache) + // untouched. + self.generation = self.generation.wrapping_add(1); self.order.push_back(value); while self.order.len() > self.capacity { if let Some(evicted) = self.order.pop_front() { @@ -79,6 +89,13 @@ impl BoundedIdSet { } } + /// A token that changes whenever the set's membership changes. A derived + /// snapshot is still current iff the generation matches the one captured + /// when it was built. + pub(crate) fn generation(&self) -> u64 { + self.generation + } + /// Iterates the cached entries in no particular order. pub(crate) fn iter(&self) -> impl Iterator { self.members.iter() @@ -171,4 +188,27 @@ mod tests { fn zero_capacity_panics() { let _ = BoundedIdSet::::with_capacity(0); } + + #[test] + fn generation_bumps_only_on_membership_change() { + // Backs the #636 convergence hex-snapshot cache: the generation must + // change on any real insert (and its eviction) but NOT on a no-op + // re-insert, or the cache would either serve stale data or rebuild + // needlessly. + let mut set = BoundedIdSet::with_capacity(2); + assert_eq!(set.generation(), 0); + set.insert(1u32); + let after_first = set.generation(); + assert_ne!(after_first, 0); + // No-op re-insert: membership unchanged, generation stable. + set.insert(1); + assert_eq!(set.generation(), after_first); + // New id: generation advances. + set.insert(2); + let after_second = set.generation(); + assert_ne!(after_second, after_first); + // Insert past capacity evicts the oldest: still a membership change. + set.insert(3); + assert_ne!(set.generation(), after_second); + } } diff --git a/crates/cgka-engine/src/distributed_convergence.rs b/crates/cgka-engine/src/distributed_convergence.rs index 78d4bcfb..a2ba1e5a 100644 --- a/crates/cgka-engine/src/distributed_convergence.rs +++ b/crates/cgka-engine/src/distributed_convergence.rs @@ -86,6 +86,28 @@ impl Engine { Ok(()) } + /// Hex-encoded `seen_message_ids` snapshot for the convergence + /// `CanonicalizationState`, cached by the seen-set generation (#636). A + /// convergence drain calls `converge_stored_openmls_messages` up to 16 times; + /// re-hex-encoding the whole (up to 100k-entry) set every pass was pure heap + /// churn. The cache is rebuilt only when the set changed since it was last + /// encoded, so a settled multi-pass drain encodes it at most once. + fn seen_message_ids_hex_for_convergence(&mut self) -> std::collections::BTreeSet { + let generation = self.seen_message_ids.generation(); + if let Some((cached_generation, snapshot)) = &self.seen_message_ids_hex_cache + && *cached_generation == generation + { + return snapshot.clone(); + } + let snapshot: std::collections::BTreeSet = self + .seen_message_ids + .iter() + .map(|message_id| hex::encode(message_id.as_slice())) + .collect(); + self.seen_message_ids_hex_cache = Some((generation, snapshot.clone())); + snapshot + } + pub(crate) fn convergence_policy_for_group( &self, group_id: &GroupId, @@ -237,11 +259,9 @@ impl Engine { current_tip_epoch: previous_tip.0, retained_anchor_epoch, last_convergence_relevant_input_ms, - seen_message_ids: self - .seen_message_ids - .iter() - .map(|message_id| hex::encode(message_id.as_slice())) - .collect(), + // #636: reuse the cached hex snapshot across the up-to-16 passes of a + // convergence drain; it is re-encoded only when the seen set changed. + seen_message_ids: self.seen_message_ids_hex_for_convergence(), }; let max_rewind_commits = policy.convergence.max_rewind_commits; diff --git a/crates/marmot-app/src/notifications.rs b/crates/marmot-app/src/notifications.rs index 0ad53da5..ad4f2690 100644 --- a/crates/marmot-app/src/notifications.rs +++ b/crates/marmot-app/src/notifications.rs @@ -1159,11 +1159,16 @@ fn notification_user_from_message( } fn notification_user(app: &MarmotApp, account_id_hex: &str) -> Result { - let profile = app.directory_entry_for_account_id(account_id_hex)?; + // #639: fetch the directory entry ONCE and derive both the display name and + // picture from it, instead of calling `directory_entry_for_account_id` here + // and again inside `display_name_for_account_id` (each re-opens the directory + // caches and fans across every account's SQLCipher cache). + let entry = app.directory_entry_for_account_id(account_id_hex)?; + let display_name = app.display_name_from_directory_entry(account_id_hex, entry.as_ref())?; Ok(NotificationUser { account_id_hex: account_id_hex.to_owned(), - display_name: app.display_name_for_account_id(account_id_hex)?, - picture_url: profile.and_then(|entry| entry.profile.and_then(|profile| profile.picture)), + display_name, + picture_url: entry.and_then(|entry| entry.profile.and_then(|profile| profile.picture)), }) } diff --git a/crates/storage-sqlite/src/chat_list.rs b/crates/storage-sqlite/src/chat_list.rs index 836b2840..460ba16e 100644 --- a/crates/storage-sqlite/src/chat_list.rs +++ b/crates/storage-sqlite/src/chat_list.rs @@ -282,6 +282,10 @@ fn rebuild_all_chat_list_rows_tx( for group in groups { rebuild_chat_list_row_for_group_tx(tx, local_account_id_hex, group, mention_classifier)?; } + // #750: a full rebuild recomputes every mention count, so the projection is + // current — advance the marker so warm-up completeness checks skip the + // per-group recompute. + set_chat_list_mention_counts_version_tx(tx, CHAT_LIST_MENTION_COUNTS_VERSION)?; Ok(()) } @@ -303,6 +307,34 @@ fn refresh_chat_list_row_tx( chat_list_row_tx(tx, group_id_hex) } +/// Current chat-list projection reconciliation version (#750). Once the stored +/// marker reaches this value, the warm-up completeness check trusts the stored +/// `unread_mention_count`s (kept current by incremental refresh) and skips the +/// O(groups × unread) per-group recompute. Bump this if a future migration can +/// again leave the counts stale. +const CHAT_LIST_MENTION_COUNTS_VERSION: i64 = 1; + +fn chat_list_mention_counts_version_tx(tx: &Connection) -> StorageResult { + Ok(tx + .query_row( + "SELECT mention_counts_version FROM chat_list_projection_meta WHERE id = 1", + [], + |row| row.get::<_, i64>(0), + ) + .optional() + .storage()? + .unwrap_or(0)) +} + +fn set_chat_list_mention_counts_version_tx(tx: &Connection, version: i64) -> StorageResult<()> { + tx.execute( + "UPDATE chat_list_projection_meta SET mention_counts_version = ?1 WHERE id = 1", + params![version], + ) + .storage()?; + Ok(()) +} + fn chat_list_projection_complete_tx( tx: &Connection, local_account_id_hex: &str, @@ -450,10 +482,18 @@ fn chat_list_projection_complete_tx( )? { return Ok(false); } + // #750: once the mention counts have been reconciled to the current + // projection version, the per-group recompute below is redundant — + // incremental refresh keeps `unread_mention_count` current — so skip the + // O(groups × unread-per-group) scan (which materializes every unread + // plaintext). The cheap structural checks above still run every warm-up. + if chat_list_mention_counts_version_tx(tx)? >= CHAT_LIST_MENTION_COUNTS_VERSION { + return Ok(true); + } // Mention classification needs the injected closure (not expressible in pure // SQL), so the remaining checks recompute the unread-mention count per group // and compare it to the stored value. This corrects rows upgraded via - // migration 0018 (which defaults the new column to 0) for groups that + // migration 0019 (which defaults the new column to 0) for groups that // actually have unread mentions. for (group_id_hex, stored_mention_count) in chat_list_stored_mention_counts_tx(tx)? { let read_state = read_state_tx(tx, &group_id_hex)?; @@ -468,6 +508,9 @@ fn chat_list_projection_complete_tx( return Ok(false); } } + // Counts verified current — record the version so future warm-ups skip the + // per-group recompute above. + set_chat_list_mention_counts_version_tx(tx, CHAT_LIST_MENTION_COUNTS_VERSION)?; Ok(true) } @@ -640,32 +683,16 @@ fn unread_summary_tx( "", ) }; - let sql = format!( - "SELECT COUNT(*) - FROM message_timeline - WHERE group_id_hex = ?1 - AND kind = ?2 - AND deleted = 0 - AND invalidation_status IS NULL - AND sender != ?3 - AND {where_sql}" - ); - let count = tx - .query_row( - &sql, - params![ - group_id_hex, - u64_to_i64(MARMOT_APP_EVENT_KIND_CHAT)?, - local_account_id_hex, - u64_to_i64(marker_at)?, - marker_id, - ], - |row| row.get::<_, i64>(0), - ) - .storage() - .and_then(i64_to_u64)?; - let first_sql = format!( - "SELECT message_id_hex + // #704: derive count + first-unread id + mention_count from ONE ordered scan + // over the unread window, instead of a separate COUNT(*), a LIMIT 1, and a + // full mention scan over the identical predicate. Ordered by the + // materialized-timeline key `(timeline_at, message_id_hex)` so the first row + // is the first-unread message. Mention classification is injected (the + // storage layer never parses nostr/NIP-21); a tag blob that fails to decode + // is treated as no tags (no mention) so one malformed row never errors the + // whole projection. + let scan_sql = format!( + "SELECT message_id_hex, plaintext, tags_json FROM message_timeline WHERE group_id_hex = ?1 AND kind = ?2 @@ -673,40 +700,10 @@ fn unread_summary_tx( AND invalidation_status IS NULL AND sender != ?3 AND {where_sql} - ORDER BY timeline_at ASC, message_id_hex ASC - LIMIT 1" + ORDER BY timeline_at ASC, message_id_hex ASC" ); - let first_message_id = tx - .query_row( - &first_sql, - params![ - group_id_hex, - u64_to_i64(MARMOT_APP_EVENT_KIND_CHAT)?, - local_account_id_hex, - u64_to_i64(marker_at)?, - marker_id, - ], - |row| row.get::<_, String>(0), - ) - .optional() - .storage()?; - // Recompute the mention count over the SAME unread window. Mention - // classification is injected (the storage layer never parses nostr/NIP-21), - // so iterate the unread rows and apply the predicate to each plaintext/tag - // pair. A tag blob that fails to decode is treated as no tags (no mention) - // so one malformed row never errors the whole projection. - let mention_sql = format!( - "SELECT plaintext, tags_json - FROM message_timeline - WHERE group_id_hex = ?1 - AND kind = ?2 - AND deleted = 0 - AND invalidation_status IS NULL - AND sender != ?3 - AND {where_sql}" - ); - let mut mention_stmt = tx.prepare(&mention_sql).storage()?; - let mut mention_rows = mention_stmt + let mut scan_stmt = tx.prepare(&scan_sql).storage()?; + let mut scan_rows = scan_stmt .query(params![ group_id_hex, u64_to_i64(MARMOT_APP_EVENT_KIND_CHAT)?, @@ -715,10 +712,17 @@ fn unread_summary_tx( marker_id, ]) .storage()?; + let mut count: u64 = 0; let mut mention_count: u64 = 0; - while let Some(row) = mention_rows.next().storage()? { - let plaintext: String = row.get(0).storage()?; - let tags_json: String = row.get(1).storage()?; + let mut first_message_id: Option = None; + while let Some(row) = scan_rows.next().storage()? { + let message_id_hex: String = row.get(0).storage()?; + let plaintext: String = row.get(1).storage()?; + let tags_json: String = row.get(2).storage()?; + if first_message_id.is_none() { + first_message_id = Some(message_id_hex); + } + count += 1; let tags = crate::tags_from_json(tags_json).unwrap_or_default(); if mention_classifier(&plaintext, &tags) { mention_count += 1; diff --git a/crates/storage-sqlite/src/migrations.rs b/crates/storage-sqlite/src/migrations.rs index 9063955f..3ed47ad1 100644 --- a/crates/storage-sqlite/src/migrations.rs +++ b/crates/storage-sqlite/src/migrations.rs @@ -40,6 +40,8 @@ mod migration_0019_chat_list_unread_mention_count; mod migration_0020_message_modifier_edges; #[path = "migrations/0021_push_token_owner_signatures.rs"] mod migration_0021_push_token_owner_signatures; +#[path = "migrations/0022_chat_list_projection_version.rs"] +mod migration_0022_chat_list_projection_version; use crate::SqliteResultExt; use cgka_traits::storage::{StorageError, StorageResult}; @@ -157,6 +159,11 @@ const MIGRATIONS: &[Migration] = &[ name: "0021_push_token_owner_signatures", apply: migration_0021_push_token_owner_signatures::apply, }, + Migration { + version: 22, + name: "0022_chat_list_projection_version", + apply: migration_0022_chat_list_projection_version::apply, + }, ]; pub(crate) fn run_all(connection: &mut Connection) -> StorageResult<()> { diff --git a/crates/storage-sqlite/src/migrations/0022_chat_list_projection_version.rs b/crates/storage-sqlite/src/migrations/0022_chat_list_projection_version.rs new file mode 100644 index 00000000..45db31c6 --- /dev/null +++ b/crates/storage-sqlite/src/migrations/0022_chat_list_projection_version.rs @@ -0,0 +1,30 @@ +//! Add a chat-list projection-version marker so the warm-up completeness check +//! can skip the O(groups × unread-per-group) per-group mention recompute once the +//! stored `unread_mention_count`s are known current (#750). +//! +//! Migration 0019 added `unread_mention_count` defaulted to 0, so +//! `chat_list_projection_complete_tx` re-derives the mention count of every +//! unread message of every group on warm-up to correct those defaulted rows — +//! materializing every unread plaintext each time. This marker records that the +//! counts have been reconciled to the current projection version; once set, the +//! per-group recompute is skipped (incremental refresh keeps the counts current +//! from then on). Seeded at version 0 (not reconciled) so the first warm-up still +//! runs the correction, then advances the marker. + +use crate::SqliteResultExt; +use cgka_traits::storage::StorageResult; +use rusqlite::Transaction; + +pub(crate) fn apply(tx: &Transaction<'_>) -> StorageResult<()> { + tx.execute_batch( + r#" +CREATE TABLE chat_list_projection_meta ( + id INTEGER PRIMARY KEY CHECK (id = 1), + mention_counts_version INTEGER NOT NULL DEFAULT 0 +); +INSERT INTO chat_list_projection_meta (id, mention_counts_version) VALUES (1, 0); +"#, + ) + .storage()?; + Ok(()) +} diff --git a/crates/storage-sqlite/src/shared.rs b/crates/storage-sqlite/src/shared.rs index bed1e310..6be65df7 100644 --- a/crates/storage-sqlite/src/shared.rs +++ b/crates/storage-sqlite/src/shared.rs @@ -283,24 +283,69 @@ CREATE TABLE IF NOT EXISTS directory_search_graph_follows ( } pub fn public_directory_users(&self) -> StorageResult> { - let ids = { - let conn = self.lock(); + // #761: two batched queries + Rust bucketing instead of a 2N+1 (one user + // row query plus one follows query per user). Preserves the full result + // set and ordering (users by account_id_hex; each user's follows by + // position then follow id). + let conn = self.lock(); + let mut follows_by_account: std::collections::HashMap> = + std::collections::HashMap::new(); + { let mut stmt = self .conn_ref(&conn) - .prepare("SELECT account_id_hex FROM directory_users ORDER BY account_id_hex") + .prepare( + "SELECT account_id_hex, follow_account_id_hex FROM directory_user_follows + ORDER BY account_id_hex, position, follow_account_id_hex", + ) .storage()?; - stmt.query_map([], |row| row.get::<_, String>(0)) - .storage()? - .collect::, _>>() - .storage()? - }; - let mut records = Vec::with_capacity(ids.len()); - for id in ids { - if let Some(record) = self.public_directory_user(&id)? { - records.push(record); + let rows = stmt + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .storage()?; + for row in rows { + let (account_id_hex, follow) = row.storage()?; + follows_by_account + .entry(account_id_hex) + .or_default() + .push(follow); } } - Ok(records) + let mut stmt = self + .conn_ref(&conn) + .prepare( + "SELECT account_id_hex, npub, profile_json, relay_lists_json, + key_package_json, event_id_hex, event_kind, event_created_at + FROM directory_users + ORDER BY account_id_hex", + ) + .storage()?; + let records = stmt + .query_map([], |row| { + Ok(PublicDirectoryUserRecord { + account_id_hex: row.get(0)?, + npub: row.get(1)?, + profile_json: row.get(2)?, + relay_lists_json: row.get(3)?, + key_package_json: row.get(4)?, + event_id_hex: row.get(5)?, + event_kind: optional_i64_to_u64(row.get::<_, Option>(6)?), + event_created_at: optional_i64_to_u64(row.get::<_, Option>(7)?), + follows: Vec::new(), + }) + }) + .storage()? + .collect::, _>>() + .storage()?; + Ok(records + .into_iter() + .map(|mut record| { + record.follows = follows_by_account + .remove(&record.account_id_hex) + .unwrap_or_default(); + record + }) + .collect()) } pub fn relay_telemetry_settings(&self) -> StorageResult { diff --git a/crates/storage-sqlite/src/timeline.rs b/crates/storage-sqlite/src/timeline.rs index 1f44806c..894f7c32 100644 --- a/crates/storage-sqlite/src/timeline.rs +++ b/crates/storage-sqlite/src/timeline.rs @@ -16,6 +16,20 @@ use rusqlite::{Connection, OptionalExtension, params, params_from_iter}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +/// The ONE display order for the materialized message-timeline / chat-list +/// surface: `(timeline_at, message_id_hex)` (#736 boundary contract 1). This is +/// the cross-client user-visible order (`timeline_at == recorded_at` at +/// projection). It is a SEPARATE surface from the raw-event replay cursor +/// (`AppEventReplayCursor`, which additionally tie-breaks on the LOCAL +/// `insert_order` and is used only for lag-recovery watermark/suppression). Keep +/// the two distinct: the replay cursor MUST NOT be applied to timeline +/// pagination. Non-aliased `ORDER BY` sites reference these; a few aliased +/// subquery lookups and the keyset-pagination predicates express the SAME order +/// inline (they carry a table alias / bind placeholders that don't fit a plain +/// fragment). +pub(crate) const TIMELINE_ORDER_BY_ASC: &str = "ORDER BY timeline_at ASC, message_id_hex ASC"; +pub(crate) const TIMELINE_ORDER_BY_DESC: &str = "ORDER BY timeline_at DESC, message_id_hex DESC"; + const DEFAULT_TIMELINE_LIMIT: usize = 50; /// Largest number of rows a single timeline cursor query returns. A /// materialized window kept above this cannot be re-fetched in one query, so @@ -1860,7 +1874,7 @@ fn timeline_records_by_ids_tx( deleted, deleted_by_message_id_hex, invalidation_status FROM message_timeline WHERE group_id_hex = ? AND message_id_hex IN ({placeholders}) - ORDER BY timeline_at ASC, message_id_hex ASC" + {TIMELINE_ORDER_BY_ASC}" ); let mut params = Vec::::with_capacity(message_ids.len() + 1); params.push(rusqlite::types::Value::Text(group_id_hex.to_owned())); @@ -1971,10 +1985,8 @@ fn timeline_query_sql( format!("WHERE {}", clauses.join(" AND ")) }; let order_sql = match pagination.direction { - CursorDirection::After => "ORDER BY timeline_at ASC, message_id_hex ASC", - CursorDirection::None | CursorDirection::Before => { - "ORDER BY timeline_at DESC, message_id_hex DESC" - } + CursorDirection::After => TIMELINE_ORDER_BY_ASC, + CursorDirection::None | CursorDirection::Before => TIMELINE_ORDER_BY_DESC, }; Ok(( format!( diff --git a/crates/transport-nostr-adapter/src/lib.rs b/crates/transport-nostr-adapter/src/lib.rs index 389f50c3..3da0e75b 100644 --- a/crates/transport-nostr-adapter/src/lib.rs +++ b/crates/transport-nostr-adapter/src/lib.rs @@ -713,9 +713,59 @@ struct AccountRoutes { groups: Vec, } +/// A stored routing endpoint paired with its parsed/canonical `RelayUrl`, cached +/// once when the route index is built so `routes_for` never re-parses the same +/// verbatim signed endpoint on every inbound event (#698/#752). The verbatim +/// string is preserved untouched (the signed-routing invariant); `parsed` is a +/// read-side accelerator (`None` for a non-relay endpoint that fails to parse). +#[derive(Clone)] +struct CanonicalEndpoint { + verbatim: TransportEndpoint, + parsed: Option, +} + +impl CanonicalEndpoint { + fn new(endpoint: &TransportEndpoint) -> Self { + Self { + parsed: RelayUrl::parse(endpoint.as_str()).ok(), + verbatim: endpoint.clone(), + } + } + + /// Normalization-safe match against an inbound endpoint, reusing this entry's + /// cached parse and the inbound endpoint's single parse. Mirrors + /// [`endpoints_match`] exactly (byte-equality fast path, else canonical + /// `RelayUrl` equality, else no match), but without re-parsing either side. + fn matches(&self, endpoint: &TransportEndpoint, parsed_endpoint: Option<&RelayUrl>) -> bool { + if self.verbatim == *endpoint { + return true; + } + match (self.parsed.as_ref(), parsed_endpoint) { + (Some(stored), Some(inbound)) => stored == inbound, + _ => false, + } + } +} + +/// A group-delivery route resolved by `transport_group_id`, with its endpoints +/// pre-canonicalized. Entries in the [`AdapterState::by_transport_group`] index. +#[derive(Clone)] +struct GroupRouteEntry { + account_id: MemberId, + group_id: GroupId, + endpoints: Vec, +} + #[derive(Default)] struct AdapterState { accounts: HashMap, + /// Derived accelerator for `routes_for` group delivery (#698/#752): maps a + /// `transport_group_id` to its candidate routes, so an inbound group event is + /// resolved in O(matching groups) instead of scanning O(accounts × groups) + /// every event (attacker-floodable). Rebuilt wholesale from `accounts` (the + /// authoritative signed-routing state) at every mutation + /// (activate/sync_groups/deactivate), so it can never drift. + by_transport_group: HashMap, Vec>, metrics: NostrAdapterMetrics, relay_index: RelayIndexRegistry, telemetry: RelayDeliveryTelemetry, @@ -753,6 +803,26 @@ fn endpoints_match(candidate: &TransportEndpoint, endpoint: &TransportEndpoint) } impl AdapterState { + /// Rebuild the derived `by_transport_group` index from the authoritative + /// `accounts` routing state. Called after every mutation so the index cannot + /// drift from the signed-routing source of truth (#698/#752). + fn rebuild_transport_group_index(&mut self) { + let mut index: HashMap, Vec> = HashMap::new(); + for (account_id, routes) in &self.accounts { + for group in &routes.groups { + index + .entry(group.transport_group_id.clone()) + .or_default() + .push(GroupRouteEntry { + account_id: account_id.clone(), + group_id: group.group_id.clone(), + endpoints: group.endpoints.iter().map(CanonicalEndpoint::new).collect(), + }); + } + } + self.by_transport_group = index; + } + fn activate(&mut self, activation: TransportAccountActivation, replaced: usize) { self.metrics.subscriptions_created += 1 + activation.group_subscriptions.len(); self.metrics.subscriptions_removed += replaced; @@ -763,6 +833,7 @@ impl AdapterState { groups: activation.group_subscriptions, }, ); + self.rebuild_transport_group_index(); } fn sync_groups(&mut self, sync: TransportGroupSync, created: usize, removed: usize) { @@ -770,12 +841,14 @@ impl AdapterState { account.groups = sync.group_subscriptions; self.metrics.subscriptions_created += created; self.metrics.subscriptions_removed += removed; + self.rebuild_transport_group_index(); } } fn deactivate(&mut self, account_id: &MemberId, removed_count: usize) { self.accounts.remove(account_id); self.metrics.subscriptions_removed += removed_count; + self.rebuild_transport_group_index(); } fn record_inbound_event(&mut self, delivered: usize) { @@ -846,27 +919,31 @@ impl AdapterState { endpoint: &TransportEndpoint, ) -> Vec { match &message.envelope { - TransportEnvelope::GroupMessage { transport_group_id } => self - .accounts - .iter() - .flat_map(|(account_id, routes)| { - routes - .groups - .iter() - .filter(move |group| { - group.transport_group_id == *transport_group_id - && group - .endpoints - .iter() - .any(|candidate| endpoints_match(candidate, endpoint)) - }) - .map(move |group| DeliveryRoute { - account_id: account_id.clone(), - group_id_hint: Some(group.group_id.clone()), - plane: TransportDeliveryPlane::Group, - }) - }) - .collect(), + TransportEnvelope::GroupMessage { transport_group_id } => { + // #698/#752: O(1) index lookup by transport_group_id, then match + // over ONLY that group's candidate routes — instead of scanning + // every account × group per event. A miss (unknown/attacker id) + // returns empty after a single hash probe. The inbound endpoint is + // parsed once here; stored endpoints are pre-parsed in the index. + let Some(entries) = self.by_transport_group.get(transport_group_id) else { + return Vec::new(); + }; + let parsed_endpoint = RelayUrl::parse(endpoint.as_str()).ok(); + entries + .iter() + .filter(|entry| { + entry + .endpoints + .iter() + .any(|candidate| candidate.matches(endpoint, parsed_endpoint.as_ref())) + }) + .map(|entry| DeliveryRoute { + account_id: entry.account_id.clone(), + group_id_hint: Some(entry.group_id.clone()), + plane: TransportDeliveryPlane::Group, + }) + .collect() + } TransportEnvelope::Welcome { recipient } => self .accounts .iter() From 4742d4bc08c4e513d9b957c2f23c083090fb4d55 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:05:10 -0400 Subject: [PATCH 05/10] feat(peeler): bind outer kind-445 created_at to the inner app event created_at (#630 package E) Part of the #736 campaign (cross-client recorded_at consistency). The outer kind-445 envelope was signed with no custom_created_at, so it defaulted to wrap-time now(): receivers all agreed (shared envelope) but the sender's own copy used the inner event created_at, so a same-second pair could sort differently in the sender's view. Stamp the outer created_at with the inner app event's created_at (already carried in GroupMessageMetadata::Application) so the sender and every receiver record the same recorded_at. PROTOCOL-VISIBLE: changes the outer kind-445 event id (created_at is in the id preimage) and makes identical content broadcast to multiple groups share a timestamp (a minor, accepted correlation trade-off). message_id_hex is unaffected. Co-Authored-By: Claude Opus 4.8 --- crates/traits/src/peeler.rs | 17 +++++++++ crates/transport-nostr-peeler/src/peeler.rs | 41 ++++++++++++++++++++- 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/crates/traits/src/peeler.rs b/crates/traits/src/peeler.rs index d413b61a..01bc7c54 100644 --- a/crates/traits/src/peeler.rs +++ b/crates/traits/src/peeler.rs @@ -47,6 +47,23 @@ impl GroupMessageMetadata { Self::CommitOrProposal } + /// The `created_at` to stamp on the OUTER transport envelope so the sender + /// and every receiver agree on the message's timestamp (#630 cross-client / + /// package E). Only application messages carry a sender-authenticated inner + /// `created_at`; commits/proposals have none, so `None` here means the + /// transport default (wrap time) stands. Binding the outer `created_at` to + /// the inner one changes the outer transport event id (it is part of the id + /// preimage) and makes broadcasts of identical content to multiple groups + /// share a timestamp — an accepted trade-off for cross-client ordering. + pub fn outer_created_at(&self) -> Option { + match self { + Self::Application { + inner_created_at, .. + } => Some(*inner_created_at), + Self::CommitOrProposal => None, + } + } + /// Compute the transport-level expiration timestamp, if any. pub fn expiration_timestamp(&self) -> Result, GroupMessageMetadataError> { let Self::Application { diff --git a/crates/transport-nostr-peeler/src/peeler.rs b/crates/transport-nostr-peeler/src/peeler.rs index 83f43d90..d3716298 100644 --- a/crates/transport-nostr-peeler/src/peeler.rs +++ b/crates/transport-nostr-peeler/src/peeler.rs @@ -154,8 +154,17 @@ impl NostrMlsPeeler { [expiration.to_string()], )); } - let signed = EventBuilder::new(Kind::Custom(KIND_MARMOT_GROUP_MESSAGE as u16), content) - .tags(tags) + // Package E (#630 cross-client): bind the outer kind-445 `created_at` to + // the inner app event's sender-authenticated `created_at` so the sender + // and every receiver record the same `recorded_at`. Without this the + // builder defaults to wrap-time `now()`, which only receivers agree on. + // Commits/proposals carry no inner timestamp, so they keep the default. + let mut builder = + EventBuilder::new(Kind::Custom(KIND_MARMOT_GROUP_MESSAGE as u16), content).tags(tags); + if let Some(created_at) = metadata.and_then(GroupMessageMetadata::outer_created_at) { + builder = builder.custom_created_at(nostr::Timestamp::from_secs(created_at)); + } + let signed = builder .sign_with_keys(&ephemeral) .map_err(|e| PeelerError::WrapFailed(format!("ephemeral kind-445 sign: {e}")))?; let event = NostrTransportEvent::from_nostr_event(&signed).map_err(to_peeler_error)?; @@ -595,6 +604,34 @@ mod tests { assert_eq!(event.tag_values(EXPIRATION_TAG).len(), 1); } + #[tokio::test] + async fn group_wrap_binds_outer_created_at_to_inner_for_app_messages() { + // Package E (#630 cross-client): the outer kind-445 `created_at` is bound + // to the inner app event's sender-authenticated `created_at`, so the + // sender and every receiver record the same `recorded_at` for a message. + let group_id = vec![0x99; 32]; + let ctx = GroupContextSnapshot::new( + EpochId(9), + HashMap::from([(DEFAULT_EXPORTER_LABEL.to_string(), vec![0x7a; 32])]), + Some(group_id), + ); + let inner_created_at = 1_700_000_123; + let wrapped = NostrMlsPeeler::default() + .wrap_group_message_with_metadata( + &EncryptedPayload { + ciphertext: b"inner mls bytes".to_vec(), + aad: vec![], + }, + &ctx, + &GroupMessageMetadata::application(inner_created_at, None), + ) + .await + .expect("wrap succeeds"); + + let event = NostrTransportEvent::from_transport_message(&wrapped).expect("payload parses"); + assert_eq!(event.created_at, inner_created_at); + } + #[tokio::test] async fn group_wrap_metadata_omits_expiration_for_control_or_disabled_retention() { let ctx = GroupContextSnapshot::new( From 79e5636e088845fb57dd030cc5c71c2479d646c2 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Tue, 30 Jun 2026 22:05:10 -0400 Subject: [PATCH 06/10] docs(architecture): describe the five inbound/convergence boundary contracts (#736) Name the canonical derived-state owners (storage ordering surfaces, runtime recovery, convergence scheduling, transport routing, critical sections) in the app-runtime overview, keep the raw-event replay cursor distinct from the materialized-timeline display order, and note recorded_at is cross-client-stable once package E binds the outer created_at. Cross-link from distributed-convergence.md. Co-Authored-By: Claude Opus 4.8 --- .../distributed-convergence.md | 5 +++ .../overview/marmot-app-runtime.md | 42 ++++++++++++++++++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/docs/marmot-architecture/distributed-convergence.md b/docs/marmot-architecture/distributed-convergence.md index c66ef38b..139d48b7 100644 --- a/docs/marmot-architecture/distributed-convergence.md +++ b/docs/marmot-architecture/distributed-convergence.md @@ -35,6 +35,11 @@ input quiesces, then canonicalizes stored commits, proposals, and app-message wi `MlsGroup`. Accepted application messages and invalidation records are then available for the application layer to render, hide, mark, or otherwise handle. +The surrounding inbound plumbing — storage ordering surfaces, runtime lag-recovery, convergence scheduling, transport +routing, and off-critical-path I/O — is governed by the five boundary contracts in +[`overview/marmot-app-runtime.md`](./overview/marmot-app-runtime.md) ("Inbound / Convergence Boundary Contracts"). +Branch selection itself (below) is unchanged by those. + ## Shape The engine keeps a bounded candidate-state graph. The live `MlsGroup` is a materialized view of the selected branch. diff --git a/docs/marmot-architecture/overview/marmot-app-runtime.md b/docs/marmot-architecture/overview/marmot-app-runtime.md index ba8d285b..3b5aaed1 100644 --- a/docs/marmot-architecture/overview/marmot-app-runtime.md +++ b/docs/marmot-architecture/overview/marmot-app-runtime.md @@ -1,7 +1,7 @@ --- title: "Marmot App Runtime Shape" created: 2026-05-19 -updated: 2026-05-20 +updated: 2026-06-30 tags: [marmot, overview, app-runtime, daemon, tui] status: implemented-first-slice --- @@ -67,6 +67,46 @@ subscriptions and relay policy: The adapter is transport plumbing. The relay plane is app-runtime orchestration. +## Inbound / Convergence Boundary Contracts + +The receive/convergence path is governed by **five boundary contracts** (tracking issue #736), each owned locally +rather than by one monolithic pipeline. The engine's live MLS roster/epoch state stays in the engine; these contracts +add the incremental, bounded, non-blocking, single-source properties around it so new code inherits them. + +1. **Storage ordering surfaces** (`storage-sqlite`). Two DISTINCT, separately-named orders that MUST NOT be conflated: + - **Raw-event replay cursor** — `AppEventReplayCursor` = `(recorded_at, message_id_hex, insert_order)` over + `app_events`, with a matching `APP_EVENT_REPLAY_ORDER_ASC/_DESC` SQL fragment. `insert_order` is a LOCAL rowid, + correct here because this cursor is only a per-client lag-recovery cut-point (never cross-client display). The + third field is load-bearing for unscoped (all-groups) recovery, where the same `message_id_hex` can appear in two + groups. + - **Materialized-timeline order** — `TIMELINE_ORDER_BY_ASC/_DESC` = `(timeline_at, message_id_hex)` over + `message_timeline`/chat-list; the cross-client user-visible display + pagination order (`timeline_at == recorded_at` + at projection). The replay cursor MUST NOT be applied to timeline pagination. +2. **Runtime recovery** (`marmot-app` runtime). The lag-recovery watermark capture and `recovery_row_is_pre_subscription` + suppression are the SAME `AppEventReplayCursor` the recovery query orders by, so the suppression boundary can never + drift from the query order. Lag replay reads a bounded window (broadcast-depth/watermark-keyed), never the full + history. +3. **Engine convergence scheduling** (`marmot-app` account worker). Pending convergence groups are drained via + `take_pending_convergence_groups()` → `ScheduledConvergence::schedule_groups` at EVERY worker loop entry, including + the deferred-startup replay loop, so buffered convergence work is never stranded. +4. **Transport routing** (`cgka-engine` + `transport-nostr-adapter`). Both layers resolve a `transport_group_id` through + an in-memory index built from authoritative state (engine: at hydration + group create/join; adapter: rebuilt at + activate/sync_groups/deactivate, with canonicalized relay endpoints cached), so per-event routing is O(1)-ish and no + unauthenticated peer can force an O(groups) pre-auth scan. +5. **Daemon/connector critical sections** (`agent-connector`; the CLI daemon lock is tracked separately). Relay I/O, + full resync, and idempotency `fsync` run OFF the per-event/per-command critical path (coalesced resync, `spawn_blocking` + persistence) so one slow item cannot head-of-line-block unrelated events. + +Canonical derived-state owners: timeline display order + raw-event replay cursor = `storage-sqlite` (the two helpers +above); lag watermark/suppression = the subscription runtime, typed on `AppEventReplayCursor`; pending-convergence +scheduling = `ScheduledConvergence`; transport routing = the engine + adapter indexes; message dedup = the engine's +bounded `seen_message_ids` set (borrowed, not re-serialized, per convergence pass). `recorded_at` is cross-client-stable: +the outer transport envelope's `created_at` is bound to the inner app event's `created_at` at wrap time, so a sender and +every receiver record the same value. + +The convergence branch-selection model is unchanged; see [`../distributed-convergence.md`](../distributed-convergence.md) +and [`../cgka-engine-canonicalization-contract.md`](../cgka-engine-canonicalization-contract.md). + ## Daemon Boundary `dmd` should host one `MarmotAppRuntime`. It should accept socket requests, pass intents into the runtime, and stream From 9c68a21561136d981971d8940cce9c835ef7f7f7 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:15:30 -0400 Subject: [PATCH 07/10] fix(connector,engine,docs): address CodeRabbit review on #768 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - inbound: when broadcast lag exceeds the bounded replay window (DELIVERED_INBOUND_CURSOR_CAPACITY), emit resync_required instead of a necessarily-partial newest-window replay, so older missed messages are not silently omitted (CodeRabbit Major, data integrity). - engine: strengthen the transport_group_id_index invariant doc — the engine has no group-deletion path today and a stale entry is self-correcting, but a future deleter/routing-rotation site MUST update the index (CodeRabbit). - docs: point the raw-event replay cursor at the `app_events` table queried via SqliteAccountStorage::app_messages (CodeRabbit). Co-Authored-By: Claude Opus 4.8 --- crates/agent-connector/src/inbound.rs | 40 ++++++++++++++++--- crates/cgka-engine/src/engine.rs | 9 ++++- .../overview/marmot-app-runtime.md | 5 ++- 3 files changed, 44 insertions(+), 10 deletions(-) diff --git a/crates/agent-connector/src/inbound.rs b/crates/agent-connector/src/inbound.rs index d794830e..5b23856f 100644 --- a/crates/agent-connector/src/inbound.rs +++ b/crates/agent-connector/src/inbound.rs @@ -112,15 +112,43 @@ impl AgentConnector { ) .map(|event| (event, replay_id)) } + Err(broadcast::error::RecvError::Lagged(dropped)) + if dropped > crate::DELIVERED_INBOUND_CURSOR_CAPACITY as u64 => + { + // The lag exceeded the bounded storage-replay window: the + // newest-window replay (capped at DELIVERED_INBOUND_CURSOR_CAPACITY + // rows) cannot be guaranteed to include every dropped message — + // missed messages older than that window fall outside it. Emit + // resync_required so the agent performs a full reconcile rather than + // resuming with older messages silently omitted, instead of treating + // a necessarily-partial replay as complete. + tracing::warn!( + target: "agent_connector", + method = "stream_inbound_events", + dropped_events = dropped, + replay_window = crate::DELIVERED_INBOUND_CURSOR_CAPACITY, + "inbound broadcast lag exceeded the bounded replay window; emitting resync_required" + ); + Some(( + resync_required_event( + account_id_hex.as_deref(), + group_id_hex.as_deref(), + dropped, + ), + None, + )) + } Err(broadcast::error::RecvError::Lagged(dropped)) => { // The broadcast channel overflowed: `dropped` events were evicted // before we could deliver them and are gone from the channel for - // good (catch-up never re-emits already-broadcast messages). Recover - // the missed inbound messages from durable storage and re-deliver - // them on the existing InboundMessage path the consumer already - // handles, so a reconnect backlog never silently loses user - // messages. The delivered-id cursor guarantees we re-deliver only - // messages this subscription has not already emitted. + // good (catch-up never re-emits already-broadcast messages). The + // drop count is within the bounded replay window (checked above), so + // the newest-window storage replay fully covers the gap. Recover the + // missed inbound messages from durable storage and re-deliver them on + // the existing InboundMessage path the consumer already handles, so a + // reconnect backlog never silently loses user messages. The + // delivered-id cursor guarantees we re-deliver only messages this + // subscription has not already emitted. match self.replay_missed_inbound( account_id_hex.as_deref(), group_id_hex.as_deref(), diff --git a/crates/cgka-engine/src/engine.rs b/crates/cgka-engine/src/engine.rs index 42d96685..d90f847a 100644 --- a/crates/cgka-engine/src/engine.rs +++ b/crates/cgka-engine/src/engine.rs @@ -193,8 +193,13 @@ pub struct Engine { /// ([`Self::hydrate_one_stored_group`]) and establishment (`do_create_group` /// / `do_join_welcome`); a `transport_group_id` is immutable for a group's /// lifetime (routing-component updates are out of scope), so no per-commit - /// maintenance is needed. If routing rotation is ever implemented, this - /// index MUST be updated at the rotation site. + /// maintenance is needed. The engine has no group-deletion path today + /// (`StorageProvider::delete_group` is never called from engine code; a left + /// group's record is retained), so an entry cannot outlive its group here — + /// and even a hypothetical stale entry is self-correcting, since the + /// resolved `GroupId` is loaded by the caller and a missing group is dropped + /// as unknown. If engine-side group deletion OR routing rotation is ever + /// added, that site MUST remove/update the corresponding index entry. pub(crate) transport_group_id_index: HashMap, GroupId>, /// #636: cached hex-encoded snapshot of `seen_message_ids` for the diff --git a/docs/marmot-architecture/overview/marmot-app-runtime.md b/docs/marmot-architecture/overview/marmot-app-runtime.md index 3b5aaed1..e5530752 100644 --- a/docs/marmot-architecture/overview/marmot-app-runtime.md +++ b/docs/marmot-architecture/overview/marmot-app-runtime.md @@ -74,8 +74,9 @@ rather than by one monolithic pipeline. The engine's live MLS roster/epoch state add the incremental, bounded, non-blocking, single-source properties around it so new code inherits them. 1. **Storage ordering surfaces** (`storage-sqlite`). Two DISTINCT, separately-named orders that MUST NOT be conflated: - - **Raw-event replay cursor** — `AppEventReplayCursor` = `(recorded_at, message_id_hex, insert_order)` over - `app_events`, with a matching `APP_EVENT_REPLAY_ORDER_ASC/_DESC` SQL fragment. `insert_order` is a LOCAL rowid, + - **Raw-event replay cursor** — `AppEventReplayCursor` = `(recorded_at, message_id_hex, insert_order)` over the + `app_events` table (queried via `SqliteAccountStorage::app_messages`), with a matching + `APP_EVENT_REPLAY_ORDER_ASC/_DESC` SQL fragment. `insert_order` is a LOCAL rowid, correct here because this cursor is only a per-client lag-recovery cut-point (never cross-client display). The third field is load-bearing for unscoped (all-groups) recovery, where the same `message_id_hex` can appear in two groups. From d206519f73c4c04b98b428568ea5d98f82aac19b Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 1 Jul 2026 14:54:18 -0400 Subject: [PATCH 08/10] fix(engine,app): support nostr_group_id/relay rotation end-to-end (adversarial review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two P1 findings: the transport routing index and the app-side subscription refresh could not handle an in-place Nostr routing-component rotation, which the protocol allows (local UpdateAppComponents and remote admin commits both reach routing via the generic app-component update path). Engine (Finding 1 — resolver went stale, a regression vs the old live-scan): - Add Engine::reindex_transport_group_id, called at every commit-apply site (do_confirm_published, remote-commit ingest, convergence apply) to additively re-insert the group's CURRENT transport_group_id. The prior id is left mapped so inbound messages still addressed to the pre-rotation route during the overlap window keep resolving (the map is intentionally many-to-one, matching the spec's "publish to the prior routing address" model). Runs on commit application only, not per app message, so the extra MlsGroup::load is cheap. - Regression test: after a rotation X->Y is applied, a later commit addressed to the NEW nostr_group_id resolves to the group instead of a phantom direct id. App (Finding 2 — pre-existing; the app never resubscribed on a routing change): - AppTransportRouting::add_group now replaces by group_id (was early-return on a duplicate) and reports whether the route set changed. - refresh_group_routes returns whether any route was added/modified; the receive path runs it once per batch and resyncs when the membership count OR a route changed (a routing/relay rotation emits no GroupStateChanged event, so the count-only trigger missed it). - Unit test: add_group replaces (not duplicates) a group's route on rotation. Co-Authored-By: Claude Opus 4.8 --- .../src/distributed_convergence.rs | 5 + crates/cgka-engine/src/engine.rs | 56 ++++++-- .../src/message_processor/ingest.rs | 5 + crates/cgka-engine/src/publish.rs | 6 + crates/cgka-engine/tests/update_group_data.rs | 122 +++++++++++++++++- crates/marmot-app/src/client/mod.rs | 19 ++- crates/marmot-app/src/client/sync.rs | 25 ++-- crates/marmot-app/src/lib.rs | 20 ++- crates/marmot-app/src/tests.rs | 43 ++++++ 9 files changed, 273 insertions(+), 28 deletions(-) diff --git a/crates/cgka-engine/src/distributed_convergence.rs b/crates/cgka-engine/src/distributed_convergence.rs index a2ba1e5a..207cfdd6 100644 --- a/crates/cgka-engine/src/distributed_convergence.rs +++ b/crates/cgka-engine/src/distributed_convergence.rs @@ -388,6 +388,11 @@ impl Engine { &result, max_retained_anchor_rewind, )?; + // #740 rotation: a routing-component update commit applied through + // convergence may have changed this group's nostr_group_id; additively + // refresh the transport-id index so it resolves on the inbound path + // (prior id retained for the overlap window). + self.reindex_transport_group_id(group_id); let origin_commit_id = single_accepted_commit_id(&result); let origin_commit_actor = single_accepted_commit_actor(&observations, origin_commit_id.as_ref()); diff --git a/crates/cgka-engine/src/engine.rs b/crates/cgka-engine/src/engine.rs index d90f847a..3c6f2108 100644 --- a/crates/cgka-engine/src/engine.rs +++ b/crates/cgka-engine/src/engine.rs @@ -192,14 +192,17 @@ pub struct Engine { /// I/O. Populated for every group at hydration /// ([`Self::hydrate_one_stored_group`]) and establishment (`do_create_group` /// / `do_join_welcome`); a `transport_group_id` is immutable for a group's - /// lifetime (routing-component updates are out of scope), so no per-commit - /// maintenance is needed. The engine has no group-deletion path today - /// (`StorageProvider::delete_group` is never called from engine code; a left - /// group's record is retained), so an entry cannot outlive its group here — - /// and even a hypothetical stale entry is self-correcting, since the - /// resolved `GroupId` is loaded by the caller and a missing group is dropped - /// as unknown. If engine-side group deletion OR routing rotation is ever - /// added, that site MUST remove/update the corresponding index entry. + /// maintenance is needed for a static route. A `transport_group_id` CAN + /// change via a Nostr routing-component update commit (rotation); every + /// commit-apply site therefore calls [`Self::reindex_transport_group_id`], + /// which additively inserts the new id while leaving the prior id in place + /// for the rotation overlap window (this map is intentionally many-to-one). + /// The engine has no group-deletion path today (`StorageProvider::delete_group` + /// is never called from engine code; a left group's record is retained), so an + /// entry cannot outlive its group — and even a hypothetical stale entry is + /// self-correcting, since the resolved `GroupId` is loaded by the caller and a + /// missing group is dropped as unknown. If engine-side group deletion is ever + /// added, that site MUST remove the corresponding index entries. pub(crate) transport_group_id_index: HashMap, GroupId>, /// #636: cached hex-encoded snapshot of `seen_message_ids` for the @@ -665,6 +668,43 @@ impl Engine { /// actually joined — but clearing a Remove/SelfRemove is not. We therefore /// scope crash-recovery to staged commits that remove no members, matching /// the prior (pre-recovery) behaviour for removal-bearing commits. + /// Additively refresh a group's `transport_group_id_index` entry from live + /// MLS state after a commit that may have changed the Nostr routing + /// component (#740 rotation). Inserts the group's CURRENT `transport_group_id`; + /// any prior id is left in place so inbound messages still addressed to the + /// pre-rotation route during the overlap window keep resolving (the map is + /// many-to-one, matching the spec's "publish to the prior routing address" + /// overlap model). Called at the commit-apply sites (`do_confirm_published` + /// for local rotation, remote-commit ingest, and convergence apply); these + /// only fire on commit application, not per app message, so the extra + /// `MlsGroup::load` is cost-appropriate. Best-effort: a load / routing-read + /// failure just forfeits the fast path for this group (inbound would fall to + /// the unknown-group disposition), never fails the merge. + pub(crate) fn reindex_transport_group_id(&mut self, group_id: &GroupId) { + let mls_group = { + let provider = crate::provider::EngineOpenMlsProvider::::new( + &self.crypto, + self.storage.mls_storage(), + ); + let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice()); + match openmls::group::MlsGroup::load( + as openmls_traits::OpenMlsProvider>::storage( + &provider, + ), + &mls_gid, + ) { + Ok(Some(group)) => group, + _ => return, + } + }; + if let Ok(transport_group_id) = + crate::app_components::transport_group_id_of_group(&mls_group) + { + self.transport_group_id_index + .insert(transport_group_id, group_id.clone()); + } + } + pub fn hydrate_stable_groups_from_storage(&mut self) -> Result<(), EngineError> { for group_id in self.storage.list_groups()? { if let Err(reason) = self.hydrate_one_stored_group(&group_id) { diff --git a/crates/cgka-engine/src/message_processor/ingest.rs b/crates/cgka-engine/src/message_processor/ingest.rs index 0260ad9d..8ed63c3f 100644 --- a/crates/cgka-engine/src/message_processor/ingest.rs +++ b/crates/cgka-engine/src/message_processor/ingest.rs @@ -768,6 +768,11 @@ impl Engine { ); self.storage.put_group(&g)?; } + // #740 rotation: a peer's applied commit may have changed the + // Nostr routing component; additively refresh the transport-id + // index so the new nostr_group_id resolves (prior id retained + // for the overlap window). No-op when routing was unchanged. + self.reindex_transport_group_id(&group_id); // Refresh self-cache since our own leaf may have been // updated by the commit's path. crate::capability_manager::cache_self_capabilities( diff --git a/crates/cgka-engine/src/publish.rs b/crates/cgka-engine/src/publish.rs index a123811a..2ba84db3 100644 --- a/crates/cgka-engine/src/publish.rs +++ b/crates/cgka-engine/src/publish.rs @@ -146,6 +146,12 @@ impl Engine { // Post-merge fork-recovery anchor (idempotent, own transaction). self.retain_current_epoch_snapshot_for_group(&group_id)?; + // #740 rotation: a confirmed local UpdateAppComponents commit may have + // changed this group's Nostr routing component; additively refresh the + // transport-id index so the new nostr_group_id resolves on the inbound + // path (the prior id stays mapped for the overlap window). + self.reindex_transport_group_id(&group_id); + // Durable state has committed. From here on the steps are in-memory // state-machine transitions (idempotent-unsafe but infallible w.r.t. the // backend) plus best-effort cleanup. Kind discriminates create (always diff --git a/crates/cgka-engine/tests/update_group_data.rs b/crates/cgka-engine/tests/update_group_data.rs index c6acd617..2df749bc 100644 --- a/crates/cgka-engine/tests/update_group_data.rs +++ b/crates/cgka-engine/tests/update_group_data.rs @@ -18,7 +18,8 @@ use cgka_engine::{Engine, EngineBuilder}; use cgka_traits::EngineError; use cgka_traits::app_components::{ AppComponentData, GROUP_ADMIN_POLICY_COMPONENT_ID, GROUP_AVATAR_URL_COMPONENT_ID, - GROUP_MESSAGE_RETENTION_COMPONENT_ID, GroupAvatarUrlV1, encode_group_avatar_url_v1, + GROUP_MESSAGE_RETENTION_COMPONENT_ID, GroupAvatarUrlV1, NOSTR_ROUTING_COMPONENT_ID, + NostrRoutingV1, default_group_components, encode_group_avatar_url_v1, encode_nostr_routing_v1, }; use cgka_traits::capabilities::{Capability, CapabilityRequirement, Feature, RequirementLevel}; use cgka_traits::engine::{CgkaEngine, CreateGroupRequest, SendIntent, SendResult}; @@ -207,6 +208,125 @@ fn converge_buffered_commit(engine: &mut Engine, group_id: assert_eq!(result.convergence_status, ConvergenceStatus::Settled); } +/// Engine that supports the Nostr routing component (the default component set +/// does not), so a group can carry — and later rotate — a `nostr_group_id`. +fn build_with_routing(id: &[u8]) -> Engine { + let mut components: Vec<_> = default_group_components().into_iter().collect(); + components.push(NOSTR_ROUTING_COMPONENT_ID); + EngineBuilder::new(SqliteAccountStorage::in_memory().unwrap()) + .identity(pad32(id)) + .account_identity_proof_signer(proof_signer(id)) + .feature_registry(registry()) + .supported_app_components(components) + .peeler(Box::new(MockPeeler)) + .build() + .unwrap() +} + +/// #740 rotation regression: after a Nostr routing-component update commit is +/// applied, the engine's `transport_group_id_index` must self-heal so inbound +/// messages addressed to the NEW `nostr_group_id` still resolve to the group. +/// Before the fix the index was populated only at hydrate/create/join and went +/// stale on rotation, stranding the group until restart. Exercises the +/// convergence-apply reindex site on the recipient; `confirm_published` and the +/// direct remote-commit-apply path call the same `reindex_transport_group_id`. +#[tokio::test] +async fn routing_rotation_reindexes_inbound_transport_group_id() { + let mut alice = build_with_routing(b"alice"); + let mut bob = build_with_routing(b"bob"); + let bob_kp = bob.fresh_key_package().await.unwrap(); + + // Create with routing X and invite bob. + let routing_x = NostrRoutingV1::new([0x41; 32], vec!["wss://x.example".into()]).unwrap(); + let (gid, create) = alice + .create_group(CreateGroupRequest { + name: "orig".into(), + description: "d".into(), + members: vec![bob_kp], + required_features: vec![], + app_components: vec![AppComponentData { + component_id: NOSTR_ROUTING_COMPONENT_ID, + data: encode_nostr_routing_v1(&routing_x).unwrap(), + }], + initial_admins: vec![], + }) + .await + .unwrap(); + let (pending, welcomes) = match create { + SendResult::GroupCreated { pending, welcomes } => (pending, welcomes), + other => panic!("expected GroupCreated, got {other:?}"), + }; + alice.confirm_published(pending).await.unwrap(); + bob.join_welcome(welcomes.into_iter().next().unwrap()) + .await + .unwrap(); + + // Alice rotates the routing X -> Y and confirms locally. + let routing_y = NostrRoutingV1::new([0x59; 32], vec!["wss://y.example".into()]).unwrap(); + let res = alice + .send(SendIntent::UpdateAppComponents { + group_id: gid.clone(), + updates: vec![AppComponentData { + component_id: NOSTR_ROUTING_COMPONENT_ID, + data: encode_nostr_routing_v1(&routing_y).unwrap(), + }], + }) + .await + .unwrap(); + let (rotate_commit, rotate_pending) = match res { + SendResult::GroupEvolution { msg, pending, .. } => (msg, pending), + other => panic!("expected GroupEvolution, got {other:?}"), + }; + alice.confirm_published(rotate_pending).await.unwrap(); + + // Bob receives the rotation commit. Per the overlap model it is published to + // the PRIOR routing address X (peers still address the old id until they + // apply the rotation); bob resolves X (indexed at join) and applies it. + let routed_rotation = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: routing_x.nostr_group_id.to_vec(), + }, + ..rotate_commit + }; + bob.ingest(routed_rotation).await.unwrap(); + converge_buffered_commit(&mut bob, &gid); + + // Alice now renames the group; this commit is published to the NEW routing + // address Y. Bob can only apply it if his transport-id index self-healed to + // map Y -> group when he applied the rotation. A stale index would resolve Y + // to a phantom direct GroupId and drop the commit as unknown. + let res = alice + .send(SendIntent::UpdateGroupData { + group_id: gid.clone(), + name: Some("after-rotation".into()), + description: None, + }) + .await + .unwrap(); + let (rename_commit, rename_pending) = match res { + SendResult::GroupEvolution { msg, pending, .. } => (msg, pending), + other => panic!("expected GroupEvolution, got {other:?}"), + }; + alice.confirm_published(rename_pending).await.unwrap(); + + let routed_rename = TransportMessage { + envelope: TransportEnvelope::GroupMessage { + transport_group_id: routing_y.nostr_group_id.to_vec(), + }, + ..rename_commit + }; + // Resolves to bob's real group (buffered for convergence), not dropped. + assert!(matches!( + bob.ingest(routed_rename).await.unwrap(), + cgka_traits::ingest::IngestOutcome::Buffered { .. } + )); + converge_buffered_commit(&mut bob, &gid); + + // Bob followed the post-rotation commit addressed to the NEW nostr_group_id. + assert_eq!(bob.group_record(&gid).unwrap().name, "after-rotation"); + assert_eq!(bob.epoch(&gid).unwrap().0, alice.epoch(&gid).unwrap().0); +} + fn malicious_app_component_commit( storage: &SqliteAccountStorage, sender: &MemberId, diff --git a/crates/marmot-app/src/client/mod.rs b/crates/marmot-app/src/client/mod.rs index bd40eaa6..bdb5f0e1 100644 --- a/crates/marmot-app/src/client/mod.rs +++ b/crates/marmot-app/src/client/mod.rs @@ -1941,13 +1941,24 @@ impl AppClient { AppGroupEncryptedMediaComponent::new(policy)?.to_app_component_data() } - pub(crate) fn refresh_group_routes(&mut self) -> Result<(), AppError> { + /// Upsert every group's current transport subscription into the routing + /// state, returning whether any route was added or modified. A modified + /// route means an in-place `nostr_group_id` / relay rotation on an existing + /// group (Finding 2) — `add_group` now replaces by `group_id` instead of + /// early-returning — so callers can trigger a single resync when routes + /// actually change rather than only on a membership-count change. + pub(crate) fn refresh_group_routes(&mut self) -> Result { + let mut changed = false; for group in &self.state.groups { let group_id = GroupId::new(hex::decode(&group.group_id_hex)?); - self.routing - .add_group(group.nostr_routing.subscription(&group_id)?); + if self + .routing + .add_group(group.nostr_routing.subscription(&group_id)?) + { + changed = true; + } } - Ok(()) + Ok(changed) } fn refresh_routing(&mut self) -> Result<(), AppError> { diff --git a/crates/marmot-app/src/client/sync.rs b/crates/marmot-app/src/client/sync.rs index c79a7833..3d15cc55 100644 --- a/crates/marmot-app/src/client/sync.rs +++ b/crates/marmot-app/src/client/sync.rs @@ -128,11 +128,14 @@ impl AppClient { routes_dirty = true; } } - // #695: coalesce. A batch with N membership-changing events previously ran - // a full transport resync (over ALL group subscriptions) once per such - // event; resync once after the batch drains instead. - if routes_dirty { - self.refresh_group_routes()?; + // #695 + rotation: reconcile transport routes once after the batch drains + // instead of per membership-changing event. refresh_group_routes upserts + // every group's current subscription — picking up a join's new route AND + // an in-place nostr_group_id / relay rotation on an existing group + // (Finding 2) — and reports whether anything changed; a membership count + // change (join/leave) also forces the single resync. + let routes_changed = self.refresh_group_routes()?; + if routes_dirty || routes_changed { self.sync_runtime_groups().await?; } self.app.save_state(&self.state)?; @@ -538,11 +541,13 @@ impl AppClient { .messages .retain(|candidate| !gossip_message_ids.contains(&candidate.message_id_hex)); } - // #695: coalesce the per-event full transport resync into one resync after - // the batch drains (was once per membership-changing event, each over ALL - // group subscriptions). - if routes_dirty { - self.refresh_group_routes()?; + // #695 + rotation: reconcile transport routes once after the batch drains. + // refresh_group_routes upserts every group's current subscription (a + // join's new route AND an in-place nostr_group_id / relay rotation on an + // existing group, Finding 2) and reports whether anything changed; a + // membership count change (join/leave) also forces the single resync. + let routes_changed = self.refresh_group_routes()?; + if routes_dirty || routes_changed { self.sync_runtime_groups().await?; } // Synthesize durable kind-1210 system rows from authenticated state diff --git a/crates/marmot-app/src/lib.rs b/crates/marmot-app/src/lib.rs index 16d78de8..3fd2270e 100644 --- a/crates/marmot-app/src/lib.rs +++ b/crates/marmot-app/src/lib.rs @@ -2785,16 +2785,26 @@ impl AppTransportRouting { } } - fn add_group(&self, group: TransportGroupSubscription) { + /// Insert or replace the route for a group, returning whether the route set + /// changed. Replaces by `group_id` (rather than early-returning on a + /// duplicate) so an in-place `nostr_group_id` / relay rotation on an existing + /// group actually switches the subscription (Finding 2). Returns `false` + /// when the group's subscription is already present and identical. + fn add_group(&self, group: TransportGroupSubscription) -> bool { let mut state = self.write(); - if state + if let Some(existing) = state .group_routes - .iter() - .any(|existing| existing.group_id == group.group_id) + .iter_mut() + .find(|existing| existing.group_id == group.group_id) { - return; + if *existing == group { + return false; + } + *existing = group; + return true; } state.group_routes.push(group); + true } fn snapshot(&self) -> AppRoutingState { diff --git a/crates/marmot-app/src/tests.rs b/crates/marmot-app/src/tests.rs index 0f0fdd12..ead49aa7 100644 --- a/crates/marmot-app/src/tests.rs +++ b/crates/marmot-app/src/tests.rs @@ -2029,3 +2029,46 @@ fn secure_prune_account_app_events_before_returns_media_hashes_above_storage_lay .is_none() ); } + +// Finding 2: an in-place nostr_group_id / relay rotation on an existing group +// must switch the transport subscription. `add_group` previously early-returned +// on a duplicate group_id, so a rotated route never took effect; it now replaces +// by group_id and reports whether the route set changed. +#[test] +fn transport_add_group_replaces_route_for_same_group_on_rotation() { + let routing = AppTransportRouting::new(AppRoutingState { + local_inbox_endpoints: Vec::new(), + key_package_endpoints: Vec::new(), + inbox_routes: HashMap::new(), + group_routes: Vec::new(), + required_acks: 0, + }); + let group_id = GroupId::new(vec![0xAB; 16]); + let sub_x = TransportGroupSubscription { + group_id: group_id.clone(), + transport_group_id: vec![0x41; 32], + endpoints: vec![TransportEndpoint("wss://x.example".to_owned())], + }; + // First insert of a new group route reports a change. + assert!(routing.add_group(sub_x.clone())); + // Re-adding the identical subscription is a no-op. + assert!(!routing.add_group(sub_x.clone())); + + // Rotation: same group_id, new transport_group_id + relay. Must replace the + // existing route, report a change, and NOT leave a duplicate for the group. + let sub_y = TransportGroupSubscription { + group_id: group_id.clone(), + transport_group_id: vec![0x59; 32], + endpoints: vec![TransportEndpoint("wss://y.example".to_owned())], + }; + assert!(routing.add_group(sub_y.clone())); + + let snapshot = routing.snapshot(); + let routes: Vec<_> = snapshot + .group_routes + .iter() + .filter(|route| route.group_id == group_id) + .collect(); + assert_eq!(routes.len(), 1, "rotation must replace, not duplicate"); + assert_eq!(routes[0], &sub_y); +} From 2f745ea417ec9ed2c4d6f878226835d825c944f9 Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:31:36 -0400 Subject: [PATCH 09/10] fix(adapter,storage): finish #698 inbox endpoints + #761 directory bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the two partials flagged in review so their scope is fully covered: #698 / #752 — the group-message routing arm already used the cached-canonical `by_transport_group` index, but the Welcome/inbox arm still called `endpoints_match`, re-parsing both stored inbox endpoints AND the inbound endpoint on every event. Store inbox endpoints as pre-parsed `CanonicalEndpoint`s (built once at activate) and match via `CanonicalEndpoint::matches` with the inbound endpoint parsed once per event, mirroring the group arm. The now-unused `endpoints_match` free function is removed (its rationale moved onto `CanonicalEndpoint::matches`); the existing `welcome_event_routes_despite_trailing_slash_mismatch` regression test still guards normalization. #761 — `public_directory_users` collapsed the 2N+1 into two batched queries but still loaded the entire (network-populated) directory cache. Add a defensive `PUBLIC_DIRECTORY_USERS_MAX` (10_000) cap: the user query is `LIMIT`-bounded and the follows query is scoped to the same capped set via a matching subquery LIMIT, so neither materializes the whole cache. The cap sits well above the app-layer search reachability (`USER_DIRECTORY_SEARCH_MAX_VISITED` == 8192) so it never truncates realistic results, and a privacy-safe warn fires if it is ever hit (no silent cap). Extracted a `public_directory_users_capped(max)` helper so the bound is unit-testable without inserting 10k rows. Co-Authored-By: Claude Opus 4.8 --- crates/storage-sqlite/src/shared.rs | 87 ++++++++++++++++++-- crates/transport-nostr-adapter/src/lib.rs | 98 +++++++++++------------ 2 files changed, 127 insertions(+), 58 deletions(-) diff --git a/crates/storage-sqlite/src/shared.rs b/crates/storage-sqlite/src/shared.rs index 6be65df7..106fbae3 100644 --- a/crates/storage-sqlite/src/shared.rs +++ b/crates/storage-sqlite/src/shared.rs @@ -11,6 +11,15 @@ use rusqlite::{OptionalExtension, params}; const SHARED_BUSY_TIMEOUT_MS: u64 = 5_000; +/// Defensive upper bound on how many public-directory users +/// [`SqliteSharedStorage::public_directory_users`] materializes at once (#761). +/// The public directory is populated from network data, so without a cap a +/// hostile or simply large cache could be loaded unboundedly into memory. Set +/// well above the app-layer directory-search reachability +/// (`USER_DIRECTORY_SEARCH_MAX_VISITED` == 8192) so it never truncates a +/// realistic result set; a warn fires if it is ever hit. +const PUBLIC_DIRECTORY_USERS_MAX: usize = 10_000; + #[derive(Clone, Debug)] pub struct SqliteSharedStorage { conn: Arc>, @@ -283,10 +292,23 @@ CREATE TABLE IF NOT EXISTS directory_search_graph_follows ( } pub fn public_directory_users(&self) -> StorageResult> { + self.public_directory_users_capped(PUBLIC_DIRECTORY_USERS_MAX) + } + + fn public_directory_users_capped( + &self, + max: usize, + ) -> StorageResult> { // #761: two batched queries + Rust bucketing instead of a 2N+1 (one user - // row query plus one follows query per user). Preserves the full result - // set and ordering (users by account_id_hex; each user's follows by - // position then follow id). + // row query plus one follows query per user), AND a defensive `max` cap + // so a large network-populated directory cache cannot be materialized + // unboundedly into memory. The cap sits well above the app-layer + // directory-search reachability, so it does not truncate realistic + // results; a warn fires if it ever bites. The follows query is scoped to + // the SAME capped user set (matching subquery LIMIT) so neither query + // loads the whole cache. Ordering preserved (users by account_id_hex; + // each user's follows by position then follow id). + let cap = i64::try_from(max).unwrap_or(i64::MAX); let conn = self.lock(); let mut follows_by_account: std::collections::HashMap> = std::collections::HashMap::new(); @@ -295,11 +317,15 @@ CREATE TABLE IF NOT EXISTS directory_search_graph_follows ( .conn_ref(&conn) .prepare( "SELECT account_id_hex, follow_account_id_hex FROM directory_user_follows + WHERE account_id_hex IN ( + SELECT account_id_hex FROM directory_users + ORDER BY account_id_hex LIMIT ?1 + ) ORDER BY account_id_hex, position, follow_account_id_hex", ) .storage()?; let rows = stmt - .query_map([], |row| { + .query_map(params![cap], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) }) .storage()?; @@ -317,11 +343,12 @@ CREATE TABLE IF NOT EXISTS directory_search_graph_follows ( "SELECT account_id_hex, npub, profile_json, relay_lists_json, key_package_json, event_id_hex, event_kind, event_created_at FROM directory_users - ORDER BY account_id_hex", + ORDER BY account_id_hex + LIMIT ?1", ) .storage()?; let records = stmt - .query_map([], |row| { + .query_map(params![cap], |row| { Ok(PublicDirectoryUserRecord { account_id_hex: row.get(0)?, npub: row.get(1)?, @@ -337,6 +364,16 @@ CREATE TABLE IF NOT EXISTS directory_search_graph_follows ( .storage()? .collect::, _>>() .storage()?; + if records.len() >= max { + // no silent caps: surface (aggregate count only, privacy-safe) that + // the listing was truncated so an operator can tell the bound bit. + tracing::warn!( + target: "storage_sqlite::shared", + method = "public_directory_users", + cap = max, + "public directory listing hit the defensive cap; results truncated", + ); + } Ok(records .into_iter() .map(|mut record| { @@ -578,6 +615,44 @@ mod tests { ); } + // #761: the batched listing is defensively bounded. The uncapped path still + // returns every user; the cap bounds the materialized set to the + // lowest-ordered users and attaches only their (subquery-scoped) follows, + // never loading the whole network-populated cache. + #[test] + fn public_directory_users_capped_bounds_result_and_scopes_follows() { + let storage = SqliteSharedStorage::in_memory().unwrap(); + for (i, prefix) in ["01", "02", "03"].iter().enumerate() { + storage + .put_public_directory_user(&PublicDirectoryUserRecord { + account_id_hex: prefix.repeat(32), + npub: format!("npub{i}"), + profile_json: None, + relay_lists_json: "{}".to_owned(), + key_package_json: None, + event_id_hex: None, + event_kind: None, + event_created_at: None, + follows: vec!["ff".repeat(32)], + }) + .unwrap(); + } + + // Uncapped path returns every user with its follows. + assert_eq!(storage.public_directory_users().unwrap().len(), 3); + + // Capped at 2: only the two lowest-ordered users, each with its follows. + let capped = storage.public_directory_users_capped(2).unwrap(); + assert_eq!(capped.len(), 2); + assert_eq!(capped[0].account_id_hex, "01".repeat(32)); + assert_eq!(capped[1].account_id_hex, "02".repeat(32)); + assert!( + capped + .iter() + .all(|user| user.follows == vec!["ff".repeat(32)]) + ); + } + #[test] fn relay_telemetry_settings_default_and_persist() { let storage = SqliteSharedStorage::in_memory().unwrap(); diff --git a/crates/transport-nostr-adapter/src/lib.rs b/crates/transport-nostr-adapter/src/lib.rs index 3da0e75b..30e7d1c8 100644 --- a/crates/transport-nostr-adapter/src/lib.rs +++ b/crates/transport-nostr-adapter/src/lib.rs @@ -709,7 +709,10 @@ struct DeliveryRoute { #[derive(Clone, Default)] struct AccountRoutes { - inbox_endpoints: Vec, + /// Pre-canonicalized so the Welcome/inbox arm of `routes_for` never re-parses + /// the same stored inbox endpoint per event (#698/#752), mirroring the group + /// arm's `by_transport_group` entries. + inbox_endpoints: Vec, groups: Vec, } @@ -732,10 +735,23 @@ impl CanonicalEndpoint { } } - /// Normalization-safe match against an inbound endpoint, reusing this entry's - /// cached parse and the inbound endpoint's single parse. Mirrors - /// [`endpoints_match`] exactly (byte-equality fast path, else canonical - /// `RelayUrl` equality, else no match), but without re-parsing either side. + /// Compare this stored route endpoint against the endpoint an inbound event + /// arrived on, normalization-safe and without re-parsing either side. + /// + /// Routing must not depend on callers pre-canonicalizing relay URLs. Inbound + /// endpoints are built from a parsed nostr `RelayUrl` (`sdk_client`), while + /// stored group/inbox endpoints carry the verbatim signed routing strings + /// (`marmot.transport.nostr.routing.v1`), which are intentionally never + /// rewritten. A raw `==` therefore drops events whenever the two differ only + /// by a `url`-canonicalizable detail (trailing slash, host case, default + /// port, percent-encoding) — see darkmatter#482. + /// + /// Fast path is byte equality against the verbatim string. Otherwise both + /// sides are compared by their cached/parsed `RelayUrl` value, folding those + /// canonicalization differences together. If either side has no parse (e.g. a + /// non-Nostr transport endpoint), fall back to byte inequality so behavior is + /// never looser than exact match. Read-side only; the verbatim stored string + /// is left untouched, preserving the signed-routing invariant. fn matches(&self, endpoint: &TransportEndpoint, parsed_endpoint: Option<&RelayUrl>) -> bool { if self.verbatim == *endpoint { return true; @@ -772,36 +788,6 @@ struct AdapterState { sync: RelaySyncTelemetry, } -/// Compare a stored route endpoint against the endpoint an inbound event -/// arrived on, normalization-safe. -/// -/// Routing must not depend on callers pre-canonicalizing relay URLs. Inbound -/// endpoints are built from a parsed nostr `RelayUrl` (`sdk_client`), while -/// stored group/inbox endpoints carry the verbatim signed routing strings -/// (`marmot.transport.nostr.routing.v1`), which are intentionally never -/// rewritten. A raw `==` therefore drops events whenever the two differ only -/// by a `url`-canonicalizable detail (trailing slash, host case, default -/// port, percent-encoding) — see darkmatter#482. -/// -/// Fast path is byte equality. Otherwise both sides are parsed as `RelayUrl` -/// and compared by value, which folds those canonicalization differences -/// together. If either side fails to parse as a relay URL (e.g. a non-Nostr -/// transport endpoint), fall back to the byte comparison so behavior is never -/// looser than exact match. This is a read-side comparison only; stored -/// endpoints are left untouched, preserving the signed-routing invariant. -fn endpoints_match(candidate: &TransportEndpoint, endpoint: &TransportEndpoint) -> bool { - if candidate == endpoint { - return true; - } - match ( - RelayUrl::parse(candidate.as_str()), - RelayUrl::parse(endpoint.as_str()), - ) { - (Ok(candidate), Ok(endpoint)) => candidate == endpoint, - _ => false, - } -} - impl AdapterState { /// Rebuild the derived `by_transport_group` index from the authoritative /// `accounts` routing state. Called after every mutation so the index cannot @@ -829,7 +815,11 @@ impl AdapterState { self.accounts.insert( activation.account_id, AccountRoutes { - inbox_endpoints: activation.inbox_endpoints, + inbox_endpoints: activation + .inbox_endpoints + .iter() + .map(CanonicalEndpoint::new) + .collect(), groups: activation.group_subscriptions, }, ); @@ -944,22 +934,26 @@ impl AdapterState { }) .collect() } - TransportEnvelope::Welcome { recipient } => self - .accounts - .iter() - .filter(|(account_id, routes)| { - *account_id == recipient - && routes - .inbox_endpoints - .iter() - .any(|candidate| endpoints_match(candidate, endpoint)) - }) - .map(|(account_id, _)| DeliveryRoute { - account_id: account_id.clone(), - group_id_hint: None, - plane: TransportDeliveryPlane::AccountInbox, - }) - .collect(), + TransportEnvelope::Welcome { recipient } => { + // #698/#752: parse the inbound endpoint once (not per candidate, + // not per account) and match against pre-canonicalized inbox + // endpoints, mirroring the group arm above. + let parsed_endpoint = RelayUrl::parse(endpoint.as_str()).ok(); + self.accounts + .iter() + .filter(|(account_id, routes)| { + *account_id == recipient + && routes.inbox_endpoints.iter().any(|candidate| { + candidate.matches(endpoint, parsed_endpoint.as_ref()) + }) + }) + .map(|(account_id, _)| DeliveryRoute { + account_id: account_id.clone(), + group_id_hint: None, + plane: TransportDeliveryPlane::AccountInbox, + }) + .collect() + } } } } From 2ccf26577694c5bfd0e0489387ad789ccac7484e Mon Sep 17 00:00:00 2001 From: Jeff Gardner <202880+erskingardner@users.noreply.github.com> Date: Wed, 1 Jul 2026 17:33:27 -0400 Subject: [PATCH 10/10] perf(marmot-app): memoize notification lookups across the wake batch (#639) Introduce a per-batch `NotificationResolver` that memoizes the notification- building lookups that repeat across the events drained in one `collect_notifications_after_wake` pass: account notification settings (keyed by account_label), the group record (keyed by account_label + group_id_hex), and directory-derived users (keyed by account_id_hex, shared by the receiver and every sender). Each backing lookup re-opens the per-account SQLCipher / directory caches and fans across accounts, so a busy multi-message batch previously re-paid that per message. - The resolver is created once and shared across the wake-drain loop; the live per-event notification stream keeps a throwaway resolver per event (via the unchanged `notification_update_from_event` wrapper), so it never serves stale settings/group/user across events. - Cached values are message-agnostic: `user` stores the raw directory result and the per-message sender-display-name fallback is applied to the returned clone in `notification_user_from_message`, never written back into the cache. - `reaction_target` stays per-message (keyed by a distinct reacted-to id). Completes the previously-partial #639 (the duplicate sender lookup was already folded). Adversarially reviewed for transparency, fallback isolation, staleness, and cross-account keying. Test: sender-display-name fallback is per-message, not cached. Closes #639 Co-Authored-By: Claude Opus 4.8 --- crates/marmot-app/src/notifications.rs | 119 +++++++++++++++++-- crates/marmot-app/src/notifications/tests.rs | 53 ++++++++- crates/marmot-app/src/runtime/mod.rs | 16 ++- 3 files changed, 170 insertions(+), 18 deletions(-) diff --git a/crates/marmot-app/src/notifications.rs b/crates/marmot-app/src/notifications.rs index ad4f2690..f99f709e 100644 --- a/crates/marmot-app/src/notifications.rs +++ b/crates/marmot-app/src/notifications.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::time::{SystemTime, UNIX_EPOCH}; use chacha20poly1305::{ @@ -908,18 +908,104 @@ pub(crate) fn is_push_gossip_kind(kind: u64) -> bool { ) } +/// Per-batch memoization of the notification-building lookups that repeat across +/// the events drained in one `collect_notifications_after_wake` pass (#639): +/// account notification settings (keyed by account_label), the group record +/// (keyed by account_label + group_id_hex), and directory-derived users (keyed by +/// account_id_hex — shared by the receiver and by every sender). Each backing +/// lookup re-opens the per-account SQLCipher / directory caches and fans across +/// accounts, so a busy multi-message batch would otherwise re-pay that per +/// message. `reaction_target` is intentionally NOT cached — it is keyed by a +/// distinct reacted-to message id per reaction, so there is nothing to reuse. +/// +/// Cached values are message-agnostic: `user` stores the raw directory result, +/// and the per-message sender-display-name fallback is applied to the returned +/// clone by `notification_user_from_message`, never written back into the cache. +#[derive(Default)] +pub(crate) struct NotificationResolver { + settings: HashMap, + groups: HashMap<(String, String), Option>, + users: HashMap, +} + +impl NotificationResolver { + fn settings( + &mut self, + app: &MarmotApp, + account_label: &str, + ) -> Result { + if let Some(settings) = self.settings.get(account_label) { + return Ok(settings.clone()); + } + let settings = app.notification_settings(account_label)?; + self.settings + .insert(account_label.to_owned(), settings.clone()); + Ok(settings) + } + + fn group( + &mut self, + app: &MarmotApp, + account_label: &str, + group_id_hex: &str, + ) -> Result, AppError> { + let key = (account_label.to_owned(), group_id_hex.to_owned()); + if let Some(group) = self.groups.get(&key) { + return Ok(group.clone()); + } + let group = app.group(account_label, group_id_hex)?; + self.groups.insert(key, group.clone()); + Ok(group) + } + + fn user( + &mut self, + app: &MarmotApp, + account_id_hex: &str, + ) -> Result { + if let Some(user) = self.users.get(account_id_hex) { + return Ok(user.clone()); + } + let user = notification_user(app, account_id_hex)?; + self.users.insert(account_id_hex.to_owned(), user.clone()); + Ok(user) + } +} + +/// Build a notification for a single event with a throwaway resolver. Used by the +/// live per-event notification stream, where events arrive one at a time and +/// there is nothing to memoize across. pub(crate) fn notification_update_from_event( app: &MarmotApp, event: &MarmotAppEvent, +) -> Result, AppError> { + let mut resolver = NotificationResolver::default(); + notification_update_from_event_cached(app, &mut resolver, event) +} + +/// Build a notification for one event, reusing `resolver`'s memoized +/// settings/group/user lookups across a batch of events (#639). +pub(crate) fn notification_update_from_event_cached( + app: &MarmotApp, + resolver: &mut NotificationResolver, + event: &MarmotAppEvent, ) -> Result, AppError> { match event { - MarmotAppEvent::MessageReceived(message) => notification_update_from_message(app, message), + MarmotAppEvent::MessageReceived(message) => { + notification_update_from_message(app, resolver, message) + } MarmotAppEvent::GroupJoined { account_id_hex, account_label, group_id, - } => notification_update_from_group_join(app, account_label, account_id_hex, group_id) - .map(Some), + } => notification_update_from_group_join( + app, + resolver, + account_label, + account_id_hex, + group_id, + ) + .map(Some), MarmotAppEvent::GroupStateUpdated { .. } | MarmotAppEvent::ProjectionUpdated(_) | MarmotAppEvent::AgentStreamStarted(_) @@ -937,9 +1023,10 @@ fn is_notifiable_message_kind(kind: u64) -> bool { fn notification_update_from_message( app: &MarmotApp, + resolver: &mut NotificationResolver, event: &RuntimeMessageReceived, ) -> Result, AppError> { - let settings = app.notification_settings(&event.account_label)?; + let settings = resolver.settings(app, &event.account_label)?; if !settings.local_notifications_enabled { return Err(AppError::NotificationsDisabled); } @@ -951,9 +1038,9 @@ fn notification_update_from_message( return Ok(None); } let group_id_hex = hex::encode(event.message.group_id.as_slice()); - let group = app.group(&event.account_label, &group_id_hex)?; - let receiver = notification_user(app, &event.account_id_hex)?; - let sender = notification_user_from_message(app, &event.message)?; + let group = resolver.group(app, &event.account_label, &group_id_hex)?; + let receiver = resolver.user(app, &event.account_id_hex)?; + let sender = notification_user_from_message(app, resolver, &event.message)?; let is_from_self = event.message.sender == event.account_id_hex; // Resolve the reacted-to row from the materialized timeline by id (not raw // app_events): the timeline reflects deletion/invalidation and never carries @@ -1006,22 +1093,23 @@ fn notification_update_from_message( fn notification_update_from_group_join( app: &MarmotApp, + resolver: &mut NotificationResolver, account_label: &str, account_id_hex: &str, group_id: &cgka_traits::GroupId, ) -> Result { - let settings = app.notification_settings(account_label)?; + let settings = resolver.settings(app, account_label)?; if !settings.local_notifications_enabled { return Err(AppError::NotificationsDisabled); } let group_id_hex = hex::encode(group_id.as_slice()); - let group = app.group(account_label, &group_id_hex)?; - let receiver = notification_user(app, account_id_hex)?; + let group = resolver.group(app, account_label, &group_id_hex)?; + let receiver = resolver.user(app, account_id_hex)?; let sender_id = group .as_ref() .and_then(|group| group.welcomer_account_id_hex.clone()) .unwrap_or_else(|| account_id_hex.to_owned()); - let sender = notification_user(app, &sender_id)?; + let sender = resolver.user(app, &sender_id)?; let invite_ref = group .as_ref() .and_then(|group| group.via_welcome_message_id_hex.clone()) @@ -1149,9 +1237,14 @@ fn reaction_notification_fields( fn notification_user_from_message( app: &MarmotApp, + resolver: &mut NotificationResolver, message: &ReceivedMessage, ) -> Result { - let mut user = notification_user(app, &message.sender)?; + // Start from the cached directory-derived user, then apply THIS message's + // sender-display-name fallback to the returned clone — never written back + // into the resolver cache, so a later message from the same sender still + // applies its own fallback. + let mut user = resolver.user(app, &message.sender)?; if user.display_name.is_none() { user.display_name = message.sender_display_name.clone(); } diff --git a/crates/marmot-app/src/notifications/tests.rs b/crates/marmot-app/src/notifications/tests.rs index 5beff4b5..8efcb12b 100644 --- a/crates/marmot-app/src/notifications/tests.rs +++ b/crates/marmot-app/src/notifications/tests.rs @@ -497,14 +497,61 @@ fn group_invite_notification_is_not_a_mention() { let app = MarmotApp::with_relay(dir.path(), "wss://relay.example"); let group_id = cgka_traits::GroupId::new(vec![0xEE; 32]); - let update = - notification_update_from_group_join(&app, "alice", &account.account_id_hex, &group_id) - .unwrap(); + let update = notification_update_from_group_join( + &app, + &mut NotificationResolver::default(), + "alice", + &account.account_id_hex, + &group_id, + ) + .unwrap(); assert!(matches!(update.trigger, NotificationTrigger::GroupInvite)); assert!(!update.is_mention); } +// #639: the per-batch NotificationResolver caches the raw directory-derived user +// (display_name from the directory, or None). The per-message sender-display-name +// fallback must apply to the RETURNED clone only, never mutate the cache — so two +// messages from the same sender (whose directory entry has no name) each get +// their OWN fallback rather than the first message's leaking to the second. +#[test] +fn resolver_sender_display_name_fallback_is_per_message_not_cached() { + let dir = tempfile::tempdir().unwrap(); + let app = MarmotApp::with_relay(dir.path(), "wss://relay.example"); + let sender = "cc".repeat(32); + + let mut resolver = NotificationResolver::default(); + // Pre-seed the cache as an absent directory entry would (no display name), so + // the resolver serves this cached user and never queries `app`. + resolver.users.insert( + sender.clone(), + NotificationUser { + account_id_hex: sender.clone(), + display_name: None, + picture_url: None, + }, + ); + + let mut first = received_chat("hi", vec![]); + first.sender = sender.clone(); + first.sender_display_name = Some("Name From First".to_owned()); + let user_first = notification_user_from_message(&app, &mut resolver, &first).unwrap(); + assert_eq!(user_first.display_name.as_deref(), Some("Name From First")); + + let mut second = received_chat("yo", vec![]); + second.sender = sender.clone(); + second.sender_display_name = Some("Name From Second".to_owned()); + let user_second = notification_user_from_message(&app, &mut resolver, &second).unwrap(); + // Gets its OWN fallback, not the first message's. + assert_eq!( + user_second.display_name.as_deref(), + Some("Name From Second") + ); + // The cached user is untouched (still no display name). + assert_eq!(resolver.users[&sender].display_name, None); +} + #[test] fn reaction_message_carries_emoji_and_resolved_target_preview() { let target = timeline_target( diff --git a/crates/marmot-app/src/runtime/mod.rs b/crates/marmot-app/src/runtime/mod.rs index caf8bce9..afa5a25d 100644 --- a/crates/marmot-app/src/runtime/mod.rs +++ b/crates/marmot-app/src/runtime/mod.rs @@ -950,11 +950,21 @@ impl MarmotAppRuntime { } let app = self.accounts.app.clone(); + // #639: one resolver shared across the drained batch so repeated + // settings/group/directory-user lookups (same account, group, or sender + // across many events) are memoized instead of re-opening the SQLCipher / + // directory caches per event. + let mut resolver = notifications::NotificationResolver::default(); let drain_until = Instant::now() + remaining; loop { match events.try_recv() { Ok(event) => { - collect_notification_update_from_event(&app, &event, &mut notifications); + collect_notification_update_from_event( + &app, + &mut resolver, + &event, + &mut notifications, + ); } Err(broadcast::error::TryRecvError::Empty) => { if Instant::now() >= drain_until { @@ -969,6 +979,7 @@ impl MarmotAppRuntime { Ok(Ok(event)) => { collect_notification_update_from_event( &app, + &mut resolver, &event, &mut notifications, ); @@ -3132,10 +3143,11 @@ where fn collect_notification_update_from_event( app: &MarmotApp, + resolver: &mut notifications::NotificationResolver, event: &MarmotAppEvent, notifications: &mut Vec, ) { - match notifications::notification_update_from_event(app, event) { + match notifications::notification_update_from_event_cached(app, resolver, event) { Ok(Some(update)) => notifications.push(update), Ok(None) | Err(AppError::NotificationsDisabled) => {} Err(_) => {