test(send): pin that a warm group send encodes nothing per recipient

wacore/src/send/tests.rs

@@ -5107,3 +5107,311 @@ mod local_identity_change_on_send {
         }
     }
 }
+
+/// A warm group send — one with no sender-key distribution, which is what a
+/// group in ordinary conversation does for every message between topology
+/// changes — carries nothing per recipient. Profiling a client reported the
+/// encoder growing with group size and named the recipient list as the thing
+/// being serialized per message; on the warm path it is not, and these tests
+/// pin that so it cannot quietly become true.
+///
+/// The distinction that makes it work: `<participants>` (one pairwise-encrypted
+/// `<enc>` per device) is built only inside the `distribution_list` branch. A
+/// warm send leaves that `None`, the phash comes from a memo as a fixed-length
+/// hash, and `stale_users_for` returns empty without walking anything. What is
+/// left is `<enc type="skmsg">` — one ciphertext for the whole group — plus a
+/// reporting token, and neither knows how many members there are.
+///
+/// The distributing send *is* linear, and inherently so: each device needs its
+/// own copy of the sender key under its own ratcheting session, so there is no
+/// cache to add. That side is covered by `mark_full_distribution_list`.
+mod warm_group_send_encoding_scale {
+    use super::*;
+    use crate::libsignal::protocol::{
+        Direction, IdentityChange, IdentityKey, IdentityKeyStore, PreKeyId, PreKeyRecord,
+        PreKeyStore, ProtocolAddress, SenderKeyRecord, SessionRecord, SessionStore, SignedPreKeyId,
+        SignedPreKeyRecord, SignedPreKeyStore,
+    };
+    use crate::runtime::{AbortHandle, Runtime};
+    use crate::types::jid::{JidExt, make_sender_key_name};
+    use crate::types::message::AddressingMode;
+    use std::future::Future;
+    use std::pin::Pin;
+    use std::time::Duration;
+    use wacore_binary::marshal::marshal;
+    use wacore_binary::node::NodeContent;
+
+    type SigResult<T> = crate::libsignal::protocol::error::Result<T>;
+
+    /// A warm send touches no pairwise session, so every store below except the
+    /// sender-key one exists only to satisfy `SignalStores`. `unreachable!()`
+    /// rather than a stub answer: if the warm path ever starts reaching for a
+    /// session, that is the regression these tests are here to catch, and it
+    /// should fail loudly instead of being absorbed.
+    #[derive(Clone, Default)]
+    struct UnusedSessionStore;
+    #[async_trait::async_trait]
+    impl SessionStore for UnusedSessionStore {
+        async fn load_session(&self, _: &ProtocolAddress) -> SigResult<Option<SessionRecord>> {
+            unreachable!("warm group send must not load a pairwise session")
+        }
+        async fn has_session(&self, _: &ProtocolAddress) -> SigResult<bool> {
+            unreachable!("warm group send must not probe for a pairwise session")
+        }
+        async fn store_session(&mut self, _: &ProtocolAddress, _: SessionRecord) -> SigResult<()> {
+            unreachable!("warm group send must not write a pairwise session")
+        }
+    }
+
+    #[derive(Clone)]
+    struct UnusedIdentityStore;
+    #[async_trait::async_trait]
+    impl IdentityKeyStore for UnusedIdentityStore {
+        async fn get_identity_key_pair(&self) -> SigResult<IdentityKeyPair> {
+            unreachable!()
+        }
+        async fn get_local_registration_id(&self) -> SigResult<u32> {
+            unreachable!()
+        }
+        async fn save_identity(
+            &mut self,
+            _: &ProtocolAddress,
+            _: &IdentityKey,
+        ) -> SigResult<IdentityChange> {
+            unreachable!()
+        }
+        async fn is_trusted_identity(
+            &self,
+            _: &ProtocolAddress,
+            _: &IdentityKey,
+            _: Direction,
+        ) -> SigResult<bool> {
+            unreachable!()
+        }
+        async fn get_identity(&self, _: &ProtocolAddress) -> SigResult<Option<IdentityKey>> {
+            unreachable!()
+        }
+    }
+
+    struct UnusedPreKeys;
+    #[async_trait::async_trait]
+    impl PreKeyStore for UnusedPreKeys {
+        async fn get_pre_key(&self, _: PreKeyId) -> SigResult<PreKeyRecord> {
+            unreachable!()
+        }
+        async fn save_pre_key(&mut self, _: PreKeyId, _: &PreKeyRecord) -> SigResult<()> {
+            unreachable!()
+        }
+        async fn remove_pre_key(&mut self, _: PreKeyId) -> SigResult<()> {
+            unreachable!()
+        }
+    }
+    struct UnusedSignedPreKeys;
+    #[async_trait::async_trait]
+    impl SignedPreKeyStore for UnusedSignedPreKeys {
+        async fn get_signed_pre_key(&self, _: SignedPreKeyId) -> SigResult<SignedPreKeyRecord> {
+            unreachable!()
+        }
+        async fn save_signed_pre_key(
+            &mut self,
+            _: SignedPreKeyId,
+            _: &SignedPreKeyRecord,
+        ) -> SigResult<()> {
+            unreachable!()
+        }
+    }
+    #[derive(Default)]
+    struct MemSenderKeyStore(HashMap<SenderKeyName, SenderKeyRecord>);
+    #[async_trait::async_trait]
+    impl SenderKeyStore for MemSenderKeyStore {
+        async fn store_sender_key(
+            &mut self,
+            n: &SenderKeyName,
+            r: SenderKeyRecord,
+        ) -> SigResult<()> {
+            self.0.insert(n.clone(), r);
+            Ok(())
+        }
+        async fn load_sender_key(&self, n: &SenderKeyName) -> SigResult<Option<SenderKeyRecord>> {
+            Ok(self.0.get(n).cloned())
+        }
+    }
+
+    struct TestRuntime;
+    #[async_trait::async_trait]
+    impl Runtime for TestRuntime {
+        fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) -> AbortHandle {
+            let handle = tokio::spawn(future);
+            AbortHandle::new(move || handle.abort())
+        }
+        fn sleep(&self, _d: Duration) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+            Box::pin(async {})
+        }
+        fn spawn_blocking(
+            &self,
+            f: Box<dyn FnOnce() + Send + 'static>,
+        ) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+            Box::pin(async move {
+                let _ = tokio::task::spawn_blocking(f).await;
+            })
+        }
+        fn yield_now(&self) -> Option<Pin<Box<dyn Future<Output = ()> + Send>>> {
+            None
+        }
+    }
+
+    fn member(i: usize) -> Jid {
+        // Fictitious, fixed-width user parts: a varying digit count would change
+        // the encoded length of the *participant list*, which is exactly the
+        // quantity under test, and only in the distributing case does it reach
+        // the wire at all.
+        format!("10000000000{:04}@s.whatsapp.net", i)
+            .parse()
+            .unwrap()
+    }
+
+    /// Byte length of the marshalled stanza with the skmsg ciphertext removed.
+    ///
+    /// The ciphertext cannot be compared directly: `pad_with_context_from_encoded`
+    /// appends a random 1..16-byte pad, so two encodes of the same message differ
+    /// in length by design. Everything else in the stanza is deterministic, and
+    /// everything else is what "does the recipient list reach the wire" asks about.
+    fn stanza_size_without_ciphertext(node: &Node) -> usize {
+        let enc = node
+            .get_optional_child("enc")
+            .expect("a group send always carries <enc>");
+        let payload = match &enc.content {
+            Some(NodeContent::Bytes(b)) => b.len(),
+            other => panic!("<enc> must carry bytes, got {other:?}"),
+        };
+        marshal(node).expect("stanza must marshal").len() - payload
+    }
+
+    async fn warm_group_stanza(member_count: usize) -> Node {
+        let own: Jid = "12025550100:3@s.whatsapp.net".parse().unwrap();
+        let group: Jid = "120363000000000001@g.us".parse().unwrap();
+        let members: Vec<Jid> = (0..member_count).map(member).collect();
+
+        // Seed the chain the warm path expects to already exist: distribution is
+        // what would otherwise create it, and a warm send by definition skips it.
+        let mut rng = rand::make_rng::<rand::rngs::StdRng>();
+        let kp = KeyPair::generate(&mut rng);
+        let mut record = SenderKeyRecord::new_empty();
+        record
+            .add_sender_key_state(3, 1, 0, &[7u8; 32], kp.public_key, Some(kp.private_key))
+            .expect("valid sender key state");
+        let name = make_sender_key_name(&group, &own.to_protocol_address());
+        let mut sender_keys = MemSenderKeyStore::default();
+        sender_keys.0.insert(name, record);
+
+        let mut sessions = UnusedSessionStore;
+        let mut identities = UnusedIdentityStore;
+        let mut prekeys = UnusedPreKeys;
+        let signed_prekeys = UnusedSignedPreKeys;
+        let mut stores = SignalStores {
+            sender_key_store: &mut sender_keys,
+            session_store: &mut sessions,
+            identity_store: &mut identities,
+            prekey_store: &mut prekeys,
+            signed_prekey_store: &signed_prekeys,
+        };
+
+        let group_info = GroupInfo::new(members.clone(), AddressingMode::Pn);
+        let resolved = std::sync::Arc::new(ResolvedGroupDevices::new(members));
+        // Warm the phash memo in setup, exactly as `setup_group_send` does in
+        // the benchmark and as production does on the first send after a
+        // topology change. Left cold, the `OnceLock` would make the *first*
+        // send recompute an O(member_count) hash inside the very path these
+        // tests claim is warm — measuring the cold path under a warm name, and
+        // leaving a regression that recomputed it per send undetectable.
+        resolved.phash(&own).expect("phash must warm in setup");
+        let message = wa::Message {
+            conversation: Some("same text regardless of group size".into()),
+            ..Default::default()
+        };
+        let account = wa::ADVSignedDeviceIdentity::default();
+
+        prepare_group_stanza(
+            &TestRuntime,
+            &mut stores,
+            &MockSendContextResolver::new(),
+            GroupStanzaRequest {
+                group: &group_info,
+                own_jid: &own,
+                own_lid: &own,
+                account: Some(&account),
+                to: &group,
+                message: &message,
+                message_id: "WARM-SCALE-1",
+                force_distribution: false,
+                distribution_targets: None,
+                distribution_policy: SenderKeyDistributionPolicy::BestEffort,
+                phash_devices: Some(&resolved),
+                edit: None,
+                extra_nodes: &[],
+                pre_encoded: None,
+            },
+        )
+        .await
+        .expect("warm group send must succeed")
+        .node
+    }
+
+    /// The headline: 8 members and 512 members produce a byte-identical stanza
+    /// once the randomly padded ciphertext is discounted. If a future change
+    /// puts anything per-recipient back on a warm send, this is what fails.
+    #[tokio::test]
+    async fn warm_send_stanza_size_is_independent_of_group_size() {
+        let mut sizes = Vec::new();
+        for n in [8usize, 32, 128, 512] {
+            let node = warm_group_stanza(n).await;
+
+            assert!(
+                node.get_optional_child("participants").is_none(),
+                "a warm send distributes no sender key, so it must emit no \
+                 <participants> fan-out (group size {n})"
+            );
+            assert!(
+                node.get_optional_child("device-identity").is_none(),
+                "<device-identity> rides along with a pkmsg in the fan-out, and \
+                 there is no fan-out here (group size {n})"
+            );
+            sizes.push((n, stanza_size_without_ciphertext(&node)));
+        }
+
+        let (_, first) = sizes[0];
+        assert!(
+            sizes.iter().all(|&(_, s)| s == first),
+            "warm group stanza must not grow with the participant count; \
+             got {sizes:?} (size excludes the randomly padded skmsg ciphertext)"
+        );
+    }
+
+    /// The phash is the one input that *is* derived from every participant, so
+    /// it is the obvious candidate for smuggling O(N) bytes onto the wire. It
+    /// does not: it is a fixed-width hash, present and identical in width at
+    /// every group size, and different between sizes because the set differs.
+    #[tokio::test]
+    async fn phash_is_present_and_fixed_width_at_every_group_size() {
+        let mut seen: Vec<(usize, String)> = Vec::new();
+        for n in [8usize, 512] {
+            let node = warm_group_stanza(n).await;
+            let phash = node
+                .attrs()
+                .optional_string("phash")
+                .unwrap_or_else(|| panic!("group send carries a phash on every send (size {n})"))
+                .to_string();
+            seen.push((n, phash));
+        }
+        assert_eq!(
+            seen[0].1.len(),
+            seen[1].1.len(),
+            "phash width must not depend on the member count: {seen:?}"
+        );
+        assert_ne!(
+            seen[0].1, seen[1].1,
+            "different participant sets must hash differently, or the \
+             fixed width above would be proving nothing"
+        );
+    }
+}