fix: cap channel-owner hints per verification to bound the shared event-id budget

An Ethereum verification for an address owning many channels emitted one
ChannelOwnerChangeHint per owned channel, unbounded. Each hint draws an id from
the block-scoped 14-bit HubEvent sequence (16384 events/block) that the hook
shares with the merge loop and the mandatory BlockConfirmed commit. Verifications
sort to priority 1, so a single verification against an address owning ~16k
channels (accumulated over prior blocks) could exhaust the block's event-id
budget: later message merges then fail via commit_transaction and drop from the
shard root (deterministic censorship), and BlockConfirmed's commit unwrap panics
on the overflow, halting nodes on replay.

Cap the fan-out at MAX_CHANNEL_OWNER_HINTS_PER_VERIFICATION = 256 (~1.5% of the
block budget, far above any plausible single-owner channel count). On exceeding
the cap the scan warns once and stops in ascending key order, so which channels
survive truncation is deterministic across nodes. Truncation is contract-legal:
a shard is already not guaranteed to carry every hint, and consumers reconcile
the full owner set via GetChannelOwner. The public hook becomes a thin wrapper
over a cap-parameterized seam so tests can drive the cap at a small value; the
merge-loop call site and the swallow-all-errors contract are unchanged. The
pre-existing BlockConfirmed unwrap-on-overflow is left as a separate follow-up.

Also folds in review hardening on the same hook: a decode helper dedup in the
tests, a malformed/None/wrong-variant body no-panic test, and doc-comment
accuracy fixes. New tests pin cap truncation, its trie-inertness, and that a cap
above the owned count still emits every hint.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: bond00729

src/storage/store/engine.rs

@@ -803,20 +803,51 @@ impl ShardEngine {
         }
     }
 
+    /// Per-verification cap on channel-owner hints. An unbounded fan-out could let a
+    /// single verification exhaust the shared 16384/block HubEvent id budget (see the
+    /// hook's SHARED-BUDGET COUPLING note). One verification's hints must stay a small
+    /// fraction of that budget so a single message can never approach exhaustion; 256
+    /// is ~1.5% of the budget, far above any plausible single-owner channel count, and
+    /// well below the block-safety threshold. An address owning more than this has its
+    /// hint set truncated in ascending key order (deterministic across nodes);
+    /// consumers reconcile the full set via `GetChannelOwner`.
+    const MAX_CHANNEL_OWNER_HINTS_PER_VERIFICATION: usize = 256;
+
     /// Emit `ChannelOwnerChangeHint` HubEvents for a just-merged Ethereum
     /// verification whose verified address owns one or more channels, by scanning
     /// THIS shard's own ByOwnerAddress replica (the trie-free ownership index built
     /// by the increment-7 block-event fold). Called only from the user-message
     /// merge loop's `Ok(merge_events)` arm — the hottest path this feature touches.
     ///
     /// STRUCTURAL SAFETY CONTRACT (the rule this increment lives or dies by): this
-    /// hook is unable to fail, alter, or panic a message merge. Its ONLY effect on
-    /// the caller is the `Vec<HubEvent>` it returns, which the caller
-    /// `events.extend(...)`s — it never sees the merge result, `validation_errors`,
-    /// or the trie. Every error — a missing/malformed body, a replica read/decode
-    /// error, an event-persist error — is swallowed with a `warn!` and yields fewer
-    /// hints, never an `Err`, never a panic, never a mutation of the merge. Do NOT
-    /// add a `?`, an `Err` early-return, or an unwrap on message/body data here.
+    /// hook never sees the merge result, `validation_errors`, or the trie, so it
+    /// cannot change what merged or the shard root *directly*. Its effects are: (1) the
+    /// `Vec<HubEvent>` it returns, which the caller `events.extend(...)`s, and (2) the
+    /// trie-free hint events it commits into the caller's `txn_batch` via
+    /// `commit_transaction`. Every error — a missing/malformed body, a replica
+    /// read/decode error, an event-persist error — is swallowed with a `warn!` and
+    /// yields fewer hints, never an `Err`, never a panic on message/body data. Do NOT
+    /// add a `?`, an `Err` early-return, or an unwrap on message/body data here. (The
+    /// one non-message panic path, `commit_transaction`'s poisoned-mutex
+    /// `lock().unwrap()`, is shared with the whole merge loop and would already have
+    /// aborted upstream — this hook adds no new panic surface.)
+    ///
+    /// SHARED-BUDGET COUPLING — bounded by a per-verification cap. Effect (2) advances
+    /// the block-scoped, 14-bit `HubEventIdGenerator` sequence (`SEQUENCE_BITS`, 16384
+    /// events/block) that this hook SHARES with the merge loop and the mandatory
+    /// `BlockConfirmed` commit. An
+    /// unbounded fan-out (one verification → N hints for N owned channels) would let a
+    /// single verification against an address owning ~16k channels exhaust that shared
+    /// budget within the block — failing later merges (`commit_transaction(...)?`) and
+    /// tripping `BlockConfirmed`'s `.unwrap()`. So the fan-out is capped at
+    /// `MAX_CHANNEL_OWNER_HINTS_PER_VERIFICATION`: one verification can never draw more
+    /// than a small fraction of the block budget, restoring the property that flooding
+    /// requires many independently rate-limited messages. Truncation is contract-legal
+    /// — the amended `hub_event.proto` already tells consumers a shard may not carry
+    /// every hint; they reconcile via `GetChannelOwner`. Hints emit in ascending key
+    /// order, so WHICH channels survive truncation is deterministic across all nodes.
+    /// (The pre-existing `BlockConfirmed` `.unwrap()` fragility is a separate,
+    /// out-of-scope follow-up.)
     ///
     /// Hints are deliberately NOT trie-indexed. The sibling `merge_events` flow
     /// through the `update_trie` loop at the call site; these hints are appended to
@@ -837,6 +868,27 @@ impl ShardEngine {
         msg: &proto::Message,
         txn_batch: &mut RocksDbTransactionBatch,
         version: EngineVersion,
+    ) -> Vec<HubEvent> {
+        self.emit_channel_owner_hints_for_verification_capped(
+            msg,
+            txn_batch,
+            version,
+            Self::MAX_CHANNEL_OWNER_HINTS_PER_VERIFICATION,
+        )
+    }
+
+    /// Cap-parameterized body of [`Self::emit_channel_owner_hints_for_verification`].
+    /// `max` is `MAX_CHANNEL_OWNER_HINTS_PER_VERIFICATION` in the single production
+    /// call (the public wrapper above); it is a parameter, and this is `pub(crate)`,
+    /// ONLY so a test can drive the truncation path at a small `max` without
+    /// registering hundreds of channels. Do not call this from production with any
+    /// other value.
+    pub(crate) fn emit_channel_owner_hints_for_verification_capped(
+        &self,
+        msg: &proto::Message,
+        txn_batch: &mut RocksDbTransactionBatch,
+        version: EngineVersion,
+        max: usize,
     ) -> Vec<HubEvent> {
         if !version.is_enabled(ProtocolFeature::ChannelOwnershipEvents) {
             return vec![];
@@ -888,15 +940,16 @@ impl ShardEngine {
             _ => return vec![],
         };
 
-        // Scan this shard's own ByOwnerAddress replica to exhaustion in ascending
-        // key order — the pinned deterministic hint order. One hint per owned
-        // channel. N is unbounded but deterministic and economically bounded by
-        // registry fees (no cap/dedup by design). Every read/persist error is
-        // swallowed with a `warn!`: the merge has already committed and stays
-        // untouchable.
+        // Scan this shard's own ByOwnerAddress replica in ascending key order — the
+        // pinned deterministic hint order — emitting at most `max` hints. The cap
+        // bounds the fan-out so one verification can never approach the shared
+        // 16384/block event-id budget; truncation is contract-legal and, because
+        // emission is ascending-key, deterministic across nodes. Every read/persist
+        // error is swallowed with a `warn!`: the merge has
+        // already committed and stays untouchable.
         let mut hints = vec![];
         let mut page_token: Option<Vec<u8>> = None;
-        loop {
+        'scan: loop {
             let page_options = PageOptions {
                 page_token: page_token.take(),
                 ..PageOptions::default()
@@ -919,6 +972,22 @@ impl ShardEngine {
             };
 
             for channel_key in channel_keys {
+                // Cap the fan-out. `hints.len()` counts hints ACTUALLY emitted — a
+                // persist failure doesn't consume an event id, so it doesn't count
+                // toward the budget the cap protects. On reaching the cap, `warn!`
+                // once and stop the scan; the merge is unaffected and consumers
+                // reconcile the untruncated ownership set via `GetChannelOwner`.
+                if hints.len() >= max {
+                    warn!(
+                        fid = msg.fid(),
+                        hash = msg.hex_hash(),
+                        owner_address = hex::encode(&address),
+                        cap = max,
+                        "Truncating channel-owner hints: verified address owns more channels than the per-verification cap"
+                    );
+                    break 'scan;
+                }
+
                 let mut hint = HubEvent::new_event(
                     HubEventType::ChannelOwnerChangeHint,
                     hub_event::Body::ChannelOwnerChangeHintBody(

src/storage/store/engine_tests.rs

@@ -5090,18 +5090,25 @@ mod tests {
                 .collect()
         }
 
+        // Decode a hint's body. Every event reaching this is already filtered to
+        // `type == ChannelOwnerChangeHint` (by `owner_change_hints`), so the body variant is
+        // guaranteed; a mismatch is a test-harness bug and panics loudly.
+        fn hint_body(hint: &HubEvent) -> &proto::ChannelOwnerChangeHintBody {
+            match hint.body.as_ref().unwrap() {
+                proto::hub_event::Body::ChannelOwnerChangeHintBody(body) => body,
+                other => panic!("expected ChannelOwnerChangeHintBody, got {:?}", other),
+            }
+        }
+
         // Only VERIFICATION_* hints, in emission (id) order — filters out the REGISTER/TRANSFER
         // hints that the block-event fold emits while seeding the replica.
         fn verification_hints(engine: &ShardEngine) -> Vec<HubEvent> {
             owner_change_hints(engine)
                 .into_iter()
-                .filter(|hint| match hint.body.as_ref() {
-                    Some(proto::hub_event::Body::ChannelOwnerChangeHintBody(body)) => {
-                        body.cause == proto::ChannelOwnerChangeCause::VerificationAdd as i32
-                            || body.cause
-                                == proto::ChannelOwnerChangeCause::VerificationRemove as i32
-                    }
-                    _ => false,
+                .filter(|hint| {
+                    let cause = hint_body(hint).cause;
+                    cause == proto::ChannelOwnerChangeCause::VerificationAdd as i32
+                        || cause == proto::ChannelOwnerChangeCause::VerificationRemove as i32
                 })
                 .collect()
         }
@@ -5112,10 +5119,7 @@ mod tests {
             owner_address: &[u8],
             cause: proto::ChannelOwnerChangeCause,
         ) {
-            let body = match hint.body.as_ref().unwrap() {
-                proto::hub_event::Body::ChannelOwnerChangeHintBody(body) => body,
-                other => panic!("expected ChannelOwnerChangeHintBody, got {:?}", other),
-            };
+            let body = hint_body(hint);
             assert_eq!(body.channel_key, channel_key);
             assert_eq!(body.owner_address, owner_address);
             assert_eq!(body.cause, cause as i32);
@@ -5393,6 +5397,163 @@ mod tests {
             );
         }
 
+        #[tokio::test]
+        async fn test_malformed_verification_body_skips_without_panic() {
+            // Adversarial-input pin (plan risk 1): the message body is untrusted network input.
+            // A message whose type passes the filter but whose body is missing or the WRONG
+            // variant — plus one with no `data` at all — must hit the warn-and-skip path: no
+            // hint, and crucially no panic (no unwrap on body/message data). The replica is
+            // populated for this address, so a well-formed body WOULD emit; the empties below are
+            // the malformed body's doing, not an empty replica.
+            let (mut engine, _tmp) = new_verifier_engine().await;
+            register_channel(&mut engine, "pets", verified_address()).await;
+
+            // Sanity: a well-formed add for this owner emits one hint (so "empty" below is meaningful).
+            let mut txn = RocksDbTransactionBatch::new();
+            assert_eq!(
+                engine
+                    .emit_channel_owner_hints_for_verification(
+                        &verification_add(messages_factory::farcaster_time()),
+                        &mut txn,
+                        EngineVersion::V20,
+                    )
+                    .len(),
+                1,
+                "sanity: a well-formed add for this owner emits one hint"
+            );
+
+            // 1. type == VerificationAddEthAddress but body = None.
+            let mut no_body = verification_add(messages_factory::farcaster_time());
+            no_body.data.as_mut().unwrap().body = None;
+
+            // 2. type == VerificationAddEthAddress but body is a DIFFERENT (Remove) variant.
+            let mut wrong_variant = verification_add(messages_factory::farcaster_time());
+            wrong_variant.data.as_mut().unwrap().body = Some(
+                proto::message_data::Body::VerificationRemoveBody(proto::VerificationRemoveBody {
+                    address: verified_address(),
+                    protocol: proto::Protocol::Ethereum as i32,
+                }),
+            );
+
+            // 3. No `data` at all — msg_type() falls back to None, so the hook takes no hint.
+            let mut no_data = verification_add(messages_factory::farcaster_time());
+            no_data.data = None;
+
+            for (label, msg) in [
+                ("body = None", &no_body),
+                ("wrong body variant", &wrong_variant),
+                ("data = None", &no_data),
+            ] {
+                let mut txn = RocksDbTransactionBatch::new();
+                let hints = engine.emit_channel_owner_hints_for_verification(
+                    msg,
+                    &mut txn,
+                    EngineVersion::V20,
+                );
+                assert!(
+                    hints.is_empty(),
+                    "malformed message ({label}) must emit no hint and not panic"
+                );
+            }
+        }
+
+        #[tokio::test]
+        async fn test_hint_fan_out_is_capped() {
+            // One verification for an address owning many channels must not emit an
+            // unbounded number of hints — that would draw down the shared 16384/block
+            // event-id budget. Drive the cap at a small value via the `_capped` seam:
+            // 5 owned channels registered in non-sorted order, cap 3 → exactly the 3
+            // LOWEST channel_keys, in ascending order (truncation is order-deterministic,
+            // not insertion-ordered). Production uses the 256 constant.
+            let (mut engine, _tmp) = new_verifier_engine().await;
+            for key in ["e", "b", "d", "a", "c"] {
+                register_channel(&mut engine, key, verified_address()).await;
+            }
+
+            let add = verification_add(messages_factory::farcaster_time());
+            let mut txn = RocksDbTransactionBatch::new();
+            let hints = engine.emit_channel_owner_hints_for_verification_capped(
+                &add,
+                &mut txn,
+                EngineVersion::V20,
+                3,
+            );
+
+            assert_eq!(hints.len(), 3, "fan-out truncated to the cap");
+            for (hint, channel_key) in hints.iter().zip(["a", "b", "c"]) {
+                assert_hint(
+                    hint,
+                    channel_key,
+                    &verified_address(),
+                    proto::ChannelOwnerChangeCause::VerificationAdd,
+                );
+            }
+        }
+
+        #[tokio::test]
+        async fn test_truncated_hints_do_not_touch_the_trie() {
+            // Truncation must be as trie-inert as normal emission: capping the fan-out
+            // mutates zero trie state, so it can never move the shard root (and thus can
+            // never affect the merge). Mirrors `test_hints_do_not_touch_the_trie`, but on
+            // the truncation path. (That a REAL merge survives ANY hook outcome — error,
+            // empty, or this truncation, which is strictly milder than the scan error
+            // tested there — is pinned by `test_verification_merge_survives_corrupt_replica`.)
+            let (mut engine, _tmp) = new_verifier_engine().await;
+            for key in ["a", "b", "c", "d", "e"] {
+                register_channel(&mut engine, key, verified_address()).await;
+            }
+
+            let root_before = engine.trie_root_hash();
+
+            let add = verification_add(messages_factory::farcaster_time());
+            let mut txn = RocksDbTransactionBatch::new();
+            let hints = engine.emit_channel_owner_hints_for_verification_capped(
+                &add,
+                &mut txn,
+                EngineVersion::V20,
+                3,
+            );
+            assert_eq!(hints.len(), 3, "the cap truncated to 3 hints");
+            engine
+                .get_stores()
+                .onchain_event_store
+                .db
+                .commit(txn)
+                .unwrap();
+
+            assert_eq!(
+                to_hex(&root_before),
+                to_hex(&engine.trie_root_hash()),
+                "emitting truncated hints must not touch the shard root"
+            );
+        }
+
+        #[tokio::test]
+        async fn test_cap_above_owned_count_emits_every_hint() {
+            // Regression guard: the cap only truncates when owned channels EXCEED it.
+            // With the cap well above the owned count, every owned channel still gets a
+            // hint — the normal path is unchanged by the cap.
+            let (mut engine, _tmp) = new_verifier_engine().await;
+            for key in ["a", "b", "c"] {
+                register_channel(&mut engine, key, verified_address()).await;
+            }
+
+            let add = verification_add(messages_factory::farcaster_time());
+            let mut txn = RocksDbTransactionBatch::new();
+            let hints = engine.emit_channel_owner_hints_for_verification_capped(
+                &add,
+                &mut txn,
+                EngineVersion::V20,
+                256,
+            );
+
+            assert_eq!(
+                hints.len(),
+                3,
+                "a cap above the owned count emits one hint per channel"
+            );
+        }
+
         #[tokio::test]
         async fn test_end_to_end_ownership_lifecycle() {
             // Capstone: a channel is registered by one address, transferred to the verified
@@ -5444,13 +5605,7 @@ mod tests {
 
             // The full hint sequence across both paths, in emission order.
             let hints = owner_change_hints(&engine);
-            let causes: Vec<i32> = hints
-                .iter()
-                .map(|hint| match hint.body.as_ref().unwrap() {
-                    proto::hub_event::Body::ChannelOwnerChangeHintBody(body) => body.cause,
-                    other => panic!("expected ChannelOwnerChangeHintBody, got {:?}", other),
-                })
-                .collect();
+            let causes: Vec<i32> = hints.iter().map(|hint| hint_body(hint).cause).collect();
             assert_eq!(
                 causes,
                 vec![