hub: member-of routing slice B1 — routed notices: for_lct, sealing, per-child share, round-robin fetch

hub/hub-daemon/src/rest.rs

@@ -304,6 +304,17 @@ pub struct SealedNotice {
     /// peer actor but was sealed by the hub.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub sealed_by: Option<Uuid>,
+    /// PRD_MEMBER_OF_ROUTING H3 (Hub B1): set when this notice is queued in a ROUTER's mailbox on
+    /// behalf of member-of child `for_lct` (canonical). The router demuxes by it BEFORE opening
+    /// anything. `None` on every ordinary notice — omitted from the JSON, so legacy notices (and
+    /// their content-derived `notice_id`s) are byte-for-byte unchanged.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub for_lct: Option<String>,
+    /// Who can open `sealed` when it was routed: `"member"` = the child's own key (its hub pin, or
+    /// its current consent key for a pure-LCT child). No routed notice is ever sealed to the router
+    /// (PRD rev 3). `None` on ordinary notices.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sealed_to: Option<String>,
 }
 
 /// Cached projection: a folded `HubState` plus the ledger position it reflects.
@@ -1554,6 +1565,8 @@ async fn queue_sealed_notice(
         pointer_uri: pointer_uri.to_string(),
         queued_at: Utc::now(),
         sealed_by: None,
+        for_lct: None,
+        sealed_to: None,
     }).await {
         tracing::warn!("queue_sealed_notice: enqueue failed for {recipient}: {e}");
     }
@@ -1587,6 +1600,16 @@ async fn enqueue_notice_op(
     s: &RestState, recipient: Uuid, notice: SealedNotice,
     op: Option<(Uuid, &str, &mailbox::SendOp)>,
 ) -> Result<Enqueued, mailbox::MailboxError> {
+    // PRD_MEMBER_OF_ROUTING H3 (Hub B1): a routed notice counts against its child's SHARE of the
+    // router's receipt mailbox, ⌊MAX / N⌋ (min 1), N = the router's current children. Read from the
+    // projection BEFORE the mailbox lock, so no new lock nesting is introduced.
+    let routed_share = match notice.for_lct.as_deref() {
+        Some(_) => {
+            let state = { let ledger = s.ledger.lock().await; s.projected(&ledger) };
+            Some(routing::routed_share(&state, recipient))
+        }
+        None => None,
+    };
     let cutoff = Utc::now() - chrono::Duration::seconds(NOTICE_TTL_SECS);
     // Notices removed here (TTL-expired or cap-evicted) die *upstream* of every
     // receiver: they never fire, never hit a gate, never dead-letter, and the
@@ -1612,6 +1635,15 @@ async fn enqueue_notice_op(
                 "the recipient's receipt mailbox has the maximum number of unacknowledged notices; \
                  no notice accepted — it clears as the recipient acknowledges");
         }
+        if let (Some(share), Some(child)) = (routed_share, notice.for_lct.as_deref()) {
+            let held = record.notices().iter().filter(|n| n.for_lct.as_deref() == Some(child)).count();
+            if held >= share {
+                // That child's share only: its siblings keep receiving (Sprout B3, rev 6).
+                return mailbox::refused(StatusCode::INSUFFICIENT_STORAGE, &format!(
+                    "{child} has {held} unacknowledged routed notices, its share of the router's \
+                     mailbox ({share}); no notice accepted for it — siblings are unaffected"));
+            }
+        }
         if !record.is_receipts() {
             let queue = record.notices_mut();
             let mut kept = Vec::with_capacity(queue.len());
@@ -5145,6 +5177,8 @@ async fn dispatch_channel(
                 pointer_uri,
                 queued_at: Utc::now(),
                 sealed_by: None,
+                for_lct: None,
+                sealed_to: None,
             };
             // The historical helper name reflects its first caller; the mechanism is generic:
             // atomically queue notice + sender-op record, then land the deterministic act.
@@ -5230,6 +5264,8 @@ async fn dispatch_channel(
                 pointer_uri,
                 queued_at: Utc::now(),
                 sealed_by: Some(caller_lct_id),
+                for_lct: None,
+                sealed_to: None,
             };
             send_secret_ordered(s, caller_lct_id, to, notice, act, operation_id.as_deref(), binding).await
         }
@@ -9958,6 +9994,8 @@ mod lct_registry_tests {
             pointer_uri: "hub://act/xyz".into(),
             queued_at: Utc::now(),
             sealed_by: None,
+            for_lct: None,
+            sealed_to: None,
         };
         assert!(enqueue_notice(&state, recipient, notice.clone()).await.unwrap());
 

hub/hub-daemon/src/rest/mailbox.rs

@@ -196,6 +196,33 @@ impl Mailbox {
     }
 }
 
+/// One receipt-fetch page. An ordinary mailbox is oldest-first, as always. A router mailbox (any
+/// notice carries `for_lct`) is filled ROUND-ROBIN across children, oldest-first within each, so a
+/// child whose notices are held un-ACKed takes one turn per page instead of hiding every sibling's
+/// mail behind it (PRD_MEMBER_OF_ROUTING H3, Sprout B3). Notices without `for_lct` form one group.
+pub(super) fn fetch_page(notices: &[SealedNotice], limit: usize) -> Vec<&SealedNotice> {
+    if !notices.iter().any(|n| n.for_lct.is_some()) {
+        return notices.iter().take(limit).collect();
+    }
+    let mut order: Vec<Option<&str>> = Vec::new();
+    let mut groups: std::collections::HashMap<Option<&str>, std::collections::VecDeque<&SealedNotice>> = Default::default();
+    for n in notices {
+        let k = n.for_lct.as_deref();
+        if !groups.contains_key(&k) { order.push(k); }
+        groups.entry(k).or_default().push_back(n);
+    }
+    let mut page = Vec::with_capacity(limit.min(notices.len()));
+    while page.len() < limit {
+        let mut took = false;
+        for k in &order {
+            if page.len() == limit { break; }
+            if let Some(n) = groups.get_mut(k).and_then(|q| q.pop_front()) { page.push(n); took = true; }
+        }
+        if !took { break; }
+    }
+    page
+}
+
 pub(super) fn notice_id(recipient: Uuid, notice: &SealedNotice) -> String {
     // Fixed struct field order, domain separated, recipient bound. Includes the
     // committed timestamp: two distinct queue entries never share an ACK by pair_id.
@@ -321,7 +348,7 @@ impl RestState {
             return refused(StatusCode::CONFLICT,
                 "receipt delivery is not enabled for this member; enrollment is an operator act");
         }
-        let notices: Vec<_> = record.notices().iter().take(limit).map(|notice|
+        let notices: Vec<_> = fetch_page(record.notices(), limit).into_iter().map(|notice|
             serde_json::json!({"id": notice_id(recipient, notice), "notice": notice})).collect();
         Ok(serde_json::json!({"protocol": RECEIVE_PROTOCOL, "notifications": notices,
             "remaining": record.notices().len().saturating_sub(limit)}))
@@ -408,7 +435,8 @@ mod tests {
     }
     fn notice() -> SealedNotice {
         SealedNotice { pair_id: Uuid::new_v4(), from: Uuid::new_v4(), sealed: "opaque".into(),
-            kind: "test".into(), pointer_uri: "test".into(), queued_at: Utc::now(), sealed_by: None }
+            kind: "test".into(), pointer_uri: "test".into(), queued_at: Utc::now(), sealed_by: None,
+            for_lct: None, sealed_to: None }
     }
     fn sql(tmp: &tempfile::TempDir, query: &str) {
         rusqlite::Connection::open(tmp.path().join("hub/hub.db")).unwrap().execute_batch(query).unwrap();
@@ -432,6 +460,113 @@ mod tests {
         Ok(serde_json::from_slice(&plain).unwrap())
     }
 
+    // ---- PRD_MEMBER_OF_ROUTING Hub B1: routed notices in a router's receipt mailbox ----
+
+    fn routed(for_lct: Option<&str>, tag: &str) -> SealedNotice {
+        SealedNotice { pointer_uri: tag.into(), for_lct: for_lct.map(str::to_string),
+            sealed_to: for_lct.map(|_| "member".to_string()), ..notice() }
+    }
+
+    /// The new fields are invisible on every ordinary notice: absent from the JSON, so a legacy
+    /// notice serializes — and therefore derives its `notice_id` — exactly as before B1.
+    #[test]
+    fn ordinary_notices_serialize_and_identify_exactly_as_before() {
+        let n = notice();
+        let json = serde_json::to_string(&n).unwrap();
+        assert!(!json.contains("for_lct") && !json.contains("sealed_to"), "{json}");
+        let back: SealedNotice = serde_json::from_str(&json).unwrap();
+        assert_eq!(notice_id(Uuid::nil(), &back), notice_id(Uuid::nil(), &n));
+        let r = routed(Some("lct:web4:mb32:bx"), "p");
+        assert_ne!(notice_id(Uuid::nil(), &r), notice_id(Uuid::nil(), &SealedNotice { for_lct: None, sealed_to: None, ..r.clone() }),
+            "a routed notice's destination is part of what its id commits to");
+    }
+
+    /// One held child cannot hide its siblings: a router page is round-robin across `for_lct`,
+    /// oldest first within each; an ordinary mailbox keeps plain oldest-first order.
+    #[test]
+    fn a_router_fetch_page_is_round_robin_across_children() {
+        let (a, b, c) = (Some("lct:a"), Some("lct:b"), Some("lct:c"));
+        let q: Vec<SealedNotice> = [(a,"a1"),(a,"a2"),(a,"a3"),(a,"a4"),(b,"b1"),(a,"a5"),(b,"b2"),(c,"c1")]
+            .into_iter().map(|(k, t)| routed(k, t)).collect();
+        let page: Vec<&str> = fetch_page(&q, 5).iter().map(|n| n.pointer_uri.as_str()).collect();
+        assert_eq!(page, ["a1", "b1", "c1", "a2", "b2"]);
+        let page: Vec<&str> = fetch_page(&q, 100).iter().map(|n| n.pointer_uri.as_str()).collect();
+        assert_eq!(page.len(), q.len(), "a large page returns everything");
+        let plain: Vec<SealedNotice> = (0..3).map(|i| routed(None, &format!("n{i}"))).collect();
+        let page: Vec<&str> = fetch_page(&plain, 2).iter().map(|n| n.pointer_uri.as_str()).collect();
+        assert_eq!(page, ["n0", "n1"], "ordinary mailbox: unchanged oldest-first");
+    }
+
+    /// Set up a router (member `m`, LCT `y`) with children, straight on the projection: these tests
+    /// are about the mailbox, not intake (Hub A's tests own intake).
+    async fn router_with_children(state: &RestState, m: Uuid, children: &[(&web4_core::crypto::KeyPair, &str)])
+        -> (hub_lib::ids::CanonicalLctId, Vec<hub_lib::ids::CanonicalLctId>) {
+        let y = hub_lib::ids::CanonicalLctId::derive(&web4_core::crypto::KeyPair::generate().verifying_key());
+        state.mailbox_enable_receipts(m).await.unwrap();
+        witness_event(state, HubEvent::RouterRegistered { router_lct: y.to_string(), member: m, issued_at: 1 }).await.unwrap();
+        let mut xs = Vec::new();
+        for (k, _) in children {
+            let x = hub_lib::ids::CanonicalLctId::derive(&k.verifying_key());
+            witness_event(state, HubEvent::MemberOfReported { member: x.to_string(), of: y.to_string(),
+                consent: hub_lib::routing::Consent { pubkey_hex: k.verifying_key().to_hex(),
+                    signature_hex: String::new(), issued_at: Utc::now().to_rfc3339() } }).await.unwrap();
+            xs.push(x);
+        }
+        (y, xs)
+    }
+
+    /// The per-child share: ⌊MAX / N⌋. A child at its share is refused (507) — for that child only;
+    /// its sibling still gets through. Ordinary sends to the router member are unaffected.
+    #[tokio::test]
+    async fn a_child_at_its_share_is_refused_and_its_sibling_is_not() {
+        let (_tmp, state, _) = fixture(true).await;
+        let m = Uuid::new_v4();
+        let (k1, k2) = (web4_core::crypto::KeyPair::generate(), web4_core::crypto::KeyPair::generate());
+        let (_y, xs) = router_with_children(&state, m, &[(&k1, "x"), (&k2, "z")]).await;
+        let (x, z) = (xs[0].to_string(), xs[1].to_string());
+        let share = { let l = state.ledger.lock().await; routing::routed_share(&state.projected(&l), m) };
+        assert_eq!(share, MAX_NOTICES_PER_MEMBER / 2, "two children: half each");
+
+        let seeded = Mailbox::Receipts(ReceiptMailbox { protocol: RECEIVE_PROTOCOL.into(),
+            notices: (0..share).map(|i| routed(Some(&x), &format!("x{i}"))).collect(), acked: vec![] });
+        state.open_store().await.unwrap().mailbox_put(m, &serde_json::to_vec(&seeded).unwrap()).await.unwrap();
+        state.notifications.lock().await.remove(&m);
+
+        let e = enqueue_notice(&state, m, routed(Some(&x), "x-over")).await.err().expect("x is at its share");
+        assert_eq!(status(Err::<(), _>(e)), StatusCode::INSUFFICIENT_STORAGE);
+        assert!(enqueue_notice(&state, m, routed(Some(&z), "z1")).await.unwrap(), "the sibling is unaffected");
+        assert!(enqueue_notice(&state, m, routed(None, "plain")).await.unwrap(), "ordinary mail to the router member too");
+    }
+
+    /// Sealing (PRD H3, rev 3): a pure-LCT child's mail is sealed to its CONSENT key, a member
+    /// child's to its hub PIN — never to the router. The router cannot open it; the child can.
+    #[tokio::test]
+    async fn routed_mail_is_sealed_to_the_child_and_the_router_cannot_open_it() {
+        let (_tmp, state, _) = fixture(true).await;
+        let (router_member, router_key) = (Uuid::new_v4(), web4_core::crypto::KeyPair::generate());
+        let child = web4_core::crypto::KeyPair::generate();
+        let (_y, xs) = router_with_children(&state, router_member, &[(&child, "x")]).await;
+        let st = { let l = state.ledger.lock().await; state.projected(&l) };
+
+        let (key, to) = routing::seal_key_for(&st, None, &xs[0]).expect("pure-LCT child: its consent key");
+        assert_eq!((key.to_hex(), to), (child.verifying_key().to_hex(), "member"));
+        let pair = Uuid::new_v4();
+        let sealed = pair_channel::Sealed::from_base64(&state.signer.channel_seal(&key, pair, b"for x only").unwrap()).unwrap();
+        let hub = state.signer.public_key().unwrap();
+        assert_eq!(pair_channel::open(&child, &hub, pair, &sealed).unwrap(), b"for x only");
+        assert!(pair_channel::open(&router_key, &hub, pair, &sealed).is_err(), "the router relays what it cannot read");
+
+        // A member child: its hub pin wins over the consent key.
+        let (member, pin) = (Uuid::new_v4(), web4_core::crypto::KeyPair::generate());
+        witness_event(&state, HubEvent::MemberAdded { member_lct_id: member, added_by: Uuid::nil(), member_name: None,
+            member_pubkey_hex: Some(pin.verifying_key().to_hex()), anchor_level: None, trust_ceiling: None }).await.unwrap();
+        let st = { let l = state.ledger.lock().await; state.projected(&l) };
+        assert_eq!(routing::seal_key_for(&st, Some(member), &xs[0]).unwrap().0.to_hex(), pin.verifying_key().to_hex());
+        // No pin and no edge: nothing to seal to — the caller must refuse, never seal to the router.
+        let stranger = hub_lib::ids::CanonicalLctId::derive(&web4_core::crypto::KeyPair::generate().verifying_key());
+        assert!(routing::seal_key_for(&st, None, &stranger).is_none());
+    }
+
     #[tokio::test]
     async fn receipt_send_secret_never_accepts_failed_store_without_optin() {
         let (tmp, state, sov) = fixture(true).await;

hub/hub-daemon/src/rest/routing.rs

@@ -333,3 +333,26 @@ pub(super) async fn admin_routers(
         .collect();
     Ok(Json(serde_json::json!({ "routers": routers })))
 }
+
+/// A child's share of its router's receipt mailbox (PRD H3, rev 6): ⌊MAX / N⌋, minimum 1, where N is
+/// the number of current member-of children of the router `router_member` serves. share × N never
+/// exceeds the mailbox total, so dead children cannot reach the router-wide 507.
+pub(super) fn routed_share(state: &HubState, router_member: Uuid) -> usize {
+    let lcts: Vec<&CanonicalLctId> = state.routers.iter()
+        .filter(|(_, r)| r.member.as_uuid() == router_member).map(|(l, _)| l).collect();
+    let n = state.member_of.values().filter(|p| lcts.contains(p)).count().max(1);
+    (MAX_NOTICES_PER_MEMBER / n).max(1)
+}
+
+/// The key a hub-sealed notice for child `lct` is sealed to (PRD H3 Sealing, Hub B1), and the
+/// `sealed_to` label. The child's own hub pin when it is a member (`member` given and pinned),
+/// else its current edge's consent key (a pure-LCT child). NEVER the router's key: the router
+/// relays ciphertext it cannot open. `None` when neither exists — the caller must refuse (409),
+/// never fall back to sealing to the router.
+pub(super) fn seal_key_for(state: &HubState, member: Option<Uuid>, lct: &CanonicalLctId)
+    -> Option<(web4_core::crypto::PublicKey, &'static str)> {
+    let hex_key = member.and_then(|m| state.member_pubkeys.get(&m).or_else(|| state.council_pubkeys.get(&m)))
+        .or_else(|| state.member_of_consent_key.get(lct))?;
+    let bytes: [u8; 32] = hex::decode(hex_key).ok()?.try_into().ok()?;
+    Some((web4_core::crypto::PublicKey::from_bytes(&bytes).ok()?, "member"))
+}