Make inbound/convergence processing incremental, bounded, non-blocking, single-source (#736)

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(),
@@ -236,9 +264,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 {

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",

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,
     }
 }
 

crates/cgka-engine/src/bounded_id_set.rs

@@ -41,6 +41,11 @@ pub(crate) struct BoundedIdSet<T: Clone + Eq + Hash> {
     members: HashSet<T>,
     order: VecDeque<T>,
     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<T: Clone + Eq + Hash> BoundedIdSet<T> {
@@ -54,6 +59,7 @@ impl<T: Clone + Eq + Hash> BoundedIdSet<T> {
             members: HashSet::new(),
             order: VecDeque::new(),
             capacity,
+            generation: 0,
         }
     }
 
@@ -71,6 +77,10 @@ impl<T: Clone + Eq + Hash> BoundedIdSet<T> {
         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<T: Clone + Eq + Hash> BoundedIdSet<T> {
         }
     }
 
+    /// 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<Item = &T> {
         self.members.iter()
@@ -171,4 +188,27 @@ mod tests {
     fn zero_capacity_panics() {
         let _ = BoundedIdSet::<u32>::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);
+    }
 }

crates/cgka-engine/src/distributed_convergence.rs

@@ -86,6 +86,28 @@ impl<S: StorageProvider> Engine<S> {
         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<String> {
+        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<String> = 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<S: StorageProvider> Engine<S> {
             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;
@@ -368,6 +388,11 @@ impl<S: StorageProvider> Engine<S> {
             &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());

crates/cgka-engine/src/engine.rs

@@ -180,6 +180,37 @@ pub struct Engine<S: StorageProvider> {
     /// added by [`Self::quarantine_stored_group_on_hydrate`] and removed by a
     /// successful [`Self::retry_hydrate_quarantined_group`].
     pub(crate) quarantined_groups: HashMap<GroupId, GroupHydrationQuarantineReason>,
+
+    /// 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
+    /// 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<Vec<u8>, 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<String>)>,
 }
 
 // ── Builder ─────────────────────────────────────────────────────────────────
@@ -318,6 +349,8 @@ impl<S: StorageProvider> EngineBuilder<S> {
             audit_operation_counter: 0,
             current_audit_context: None,
             quarantined_groups: HashMap::new(),
+            transport_group_id_index: HashMap::new(),
+            seen_message_ids_hex_cache: None,
         })
     }
 }
@@ -635,6 +668,43 @@ impl<S: StorageProvider> Engine<S> {
     /// 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::<S>::new(
+                &self.crypto,
+                self.storage.mls_storage(),
+            );
+            let mls_gid = openmls::group::GroupId::from_slice(group_id.as_slice());
+            match openmls::group::MlsGroup::load(
+                <crate::provider::EngineOpenMlsProvider<'_, S> 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) {
@@ -778,6 +848,18 @@ impl<S: StorageProvider> Engine<S> {
             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,

crates/cgka-engine/src/group_lifecycle.rs

@@ -223,6 +223,16 @@ impl<S: StorageProvider> Engine<S> {
             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<S: StorageProvider> Engine<S> {
         };
         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