From a25944c8c768f92820ccb074a086050dd849f41e Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 12:51:31 +0000 Subject: [PATCH 1/4] docs(events): document Event::SentFrame and the with_stats -> with_observers rename whatsapp-rust#1260 added Event::SentFrame / Client::acquire_sent_frame_forwarding() as the outbound counterpart of RawNode, and renamed NoiseSocket::with_stats to with_observers (now taking a SendObservers struct instead of Option>), with the same change threading through do_handshake's last parameter. - concepts/events.mdx: add SentFrame to the Event enum listing and a new "Sent frame events" section modeled on DecryptedPayload's. - api/client.mdx: add acquire_sent_frame_forwarding, modeled on acquire_decrypted_payload_forwarding. - advanced/websocket-handling.mdx: document SendObservers, update the with_observers/sender_task signatures, and note the breaking rename. --- advanced/websocket-handling.mdx | 56 ++++++++++++++++++++++++++++----- api/client.mdx | 33 +++++++++++++++++++ concepts/events.mdx | 44 ++++++++++++++++++++++++++ 3 files changed, 126 insertions(+), 7 deletions(-) diff --git a/advanced/websocket-handling.mdx b/advanced/websocket-handling.mdx index 75a8028..4038743 100644 --- a/advanced/websocket-handling.mdx +++ b/advanced/websocket-handling.mdx @@ -91,9 +91,14 @@ pub async fn do_handshake( ik_handshake_failures: &AtomicU32, transport: Arc, transport_events: &mut async_channel::Receiver, + observers: SendObservers, ) -> Result> ``` + +**Breaking change:** `do_handshake`'s last parameter used to be `stats: Option>`. It is now `observers: SendObservers`, the same struct [`NoiseSocket::with_observers`](#dedicated-sender-task) takes — see below. A caller that only wants wire-byte accounting passes `SendObservers::with_stats(stats)`; a caller that wants neither observer passes `SendObservers::default()`. + + **Step-by-step process for XX:** 1. **Prepare client payload:** @@ -329,13 +334,31 @@ Location: `src/socket/noise_socket.rs:50-66` The socket uses a dedicated task for sending to ensure frame ordering: ```rust +/// What a socket reports its sends to. Both halves belong to the `Client`; a +/// VoIP relay socket and most tests pass `Default`, reporting to neither. +#[derive(Default, Clone)] +pub struct SendObservers { + /// Wire-byte accounting, recorded after the transport write. + stats: Option>, + /// Publisher for the plaintext of each frame that reached the transport. + sent_frames: Option>, +} + +impl SendObservers { + /// Report wire bytes into `stats` and nothing else. + pub fn with_stats(stats: Arc) -> Self { /* ... */ } + + /// Also publish each sent frame's plaintext through `tap`. + pub(crate) fn with_sent_frames(mut self, tap: Arc) -> Self { /* ... */ } +} + impl NoiseSocket { - pub fn with_stats( + pub fn with_observers( runtime: Arc, transport: Arc, write_key: NoiseCipher, read_key: NoiseCipher, - stats: Option>, + observers: SendObservers, ) -> Self { let write_key = Arc::new(write_key); let read_key = Arc::new(read_key); @@ -350,7 +373,7 @@ impl NoiseSocket { transport.clone(), write_key.clone(), send_job_rx, - stats, + observers, ))); Self { @@ -363,6 +386,12 @@ impl NoiseSocket { } ``` +`NoiseSocket::new` (used by the illustrative handshake walkthrough above) is unchanged: it's a thin wrapper that calls `with_observers` with `SendObservers::default()`, for the callers — mainly tests — that want neither observer. + + +**Breaking change:** `with_stats(..., stats: Option>)` is now `with_observers(..., observers: SendObservers)`. `SendObservers` is one struct rather than one parameter per observer, so the next thing that wants to watch sends — [`SentFrame`](/concepts/events#sentframe) was the first — plugs in there instead of widening this constructor (and `do_handshake`'s) again. A caller that only wants what `with_stats` gave it passes `SendObservers::with_stats(stats)`; the main WA session socket also chains `.with_sent_frames(client.sent_frame_tap.clone())` to wire up sent-frame forwarding, gated by [`Client::acquire_sent_frame_forwarding()`](/api/client#acquire_sent_frame_forwarding). VoIP relay sockets and most tests pass `SendObservers::default()`, reporting to neither — same as passing `None` before. + + **Why a dedicated task?** 1. **Ordering guarantee**: Frames must be sent with sequential counters @@ -382,14 +411,19 @@ async fn sender_task( transport: Arc, write_key: Arc, send_job_rx: async_channel::Receiver, - stats: Option>, + observers: SendObservers, ) { + let SendObservers { stats, sent_frames } = observers; let mut write_counter: u32 = 0; let mut enc_buf = Vec::with_capacity(4096); let mut out_buf = BytesMut::with_capacity(4096); let mut poisoned = false; let mut waiters: Vec<(oneshot::Sender, usize)> = Vec::new(); let mut carry_over: Option = None; + // Plaintexts of this batch's frames, held only while a consumer is + // watching; empty (unallocated) otherwise. Kept until after the write so + // what is published is what the transport actually accepted. + let mut observed: Vec = Vec::new(); loop { let job = match carry_over.take() { @@ -404,10 +438,18 @@ async fn sender_task( continue; } - // Encrypt everything already queued into out_buf. Stop at the - // MAX_BATCH_FRAMES / MAX_BATCH_WIRE_BYTES ceiling, or when try_recv - // finds nothing more waiting. + // Encrypt everything already queued into out_buf, cloning each + // plaintext into `observed` first when `sent_frames` is enabled. Stop + // at the MAX_BATCH_FRAMES / MAX_BATCH_WIRE_BYTES ceiling, or when + // try_recv finds nothing more waiting. // ... encrypt_frame_into loop, then a single transport.send(out_buf) ... + + // After the write succeeds: stats.record_frame_sent for each wire + // frame, then — re-checking sent_frames.enabled() rather than trusting + // the read at capture time, so a batch that outlived its last lease + // stays quiet — tap.publish(plaintext) for each entry in `observed`. + // A write that fails clears `observed` instead: a frame the transport + // refused is not reported as sent. } } ``` diff --git a/api/client.mdx b/api/client.mdx index 7f2c7c5..cc7e7d0 100644 --- a/api/client.mdx +++ b/api/client.mdx @@ -2038,6 +2038,39 @@ while let Ok(event) = event_rx.recv().await { // Drop `_lease` to stop forwarding. ``` +### acquire_sent_frame_forwarding + +```rust +pub fn acquire_sent_frame_forwarding(self: &Arc) -> SentFrameLease +``` + +Acquire a lease that keeps [`Event::SentFrame`](/concepts/events#sentframe) enabled for one consumer. The lease is necessary but not sufficient: a handler that has narrowed its `interest()` away from the default `EventInterest::ALL` also needs `EventKind::SentFrame` added back in, or it won't see the event even while a lease is held. This is the outbound counterpart of [`acquire_decrypted_payload_forwarding`](#acquire_decrypted_payload_forwarding): the event carries the marshaled plaintext of every frame the transport accepted, and — unlike [`wait_for_sent_node`](#wait_for_sent_node) — it is neither filtered nor one-shot, and covers every send path, including acks, delivery receipts, and direct-encoded IQs that never build a `Node` at all. + + + RAII lease. `Event::SentFrame` stays enabled until every acquired lease is dropped — hold it for as long as you want the event forwarded. The lease holds only a weak client reference, so it cannot keep the client alive. + + + +While no lease is held, nothing is emitted and nothing is cloned — the path costs one relaxed atomic load on the noise sender task. Under a lease, the forwarded frame is the same `bytes::Bytes` the caller handed to the socket, so forwarding it is a refcount bump rather than a copy. + + +**Example:** +```rust +use wacore::types::events::{ChannelEventHandler, Event}; + +let _lease = client.acquire_sent_frame_forwarding(); + +let (handler, event_rx) = ChannelEventHandler::new(); +client.register_handler(handler); + +while let Ok(event) = event_rx.recv().await { + if let Event::SentFrame(frame) = &*event { + println!("sent {} bytes", frame.plaintext.len()); + } +} +// Drop `_lease` to stop forwarding. +``` + --- ## Call management diff --git a/concepts/events.mdx b/concepts/events.mdx index 8be5364..9a88bf7 100644 --- a/concepts/events.mdx +++ b/concepts/events.mdx @@ -215,6 +215,9 @@ pub enum Event { // Decrypted payload (opt-in) DecryptedPayload(DecryptedPayload), + + // Sent frame (opt-in) + SentFrame(SentFrame), } ``` @@ -2700,6 +2703,47 @@ Event::DecryptedPayload(payload) => { See [`acquire_decrypted_payload_forwarding`](/api/client#acquire_decrypted_payload_forwarding) for the lease API. +## Sent frame events + +### `SentFrame` + +**Emitted:** One marshaled stanza, exactly as it was handed to the noise frame encryption, after the transport accepts the write. To receive this event, hold a lease from `client.acquire_sent_frame_forwarding()` and include `EventKind::SentFrame` in your handler's `interest()`. While no lease is held, nothing is emitted and nothing is cloned. + +```rust +#[derive(Debug, Clone, Serialize, bon::Builder)] +#[non_exhaustive] +pub struct SentFrame { + #[serde(skip)] + pub plaintext: Bytes, +} + +Event::SentFrame(SentFrame) +``` + +**Fields:** +- `plaintext` — The marshaled stanza, keeping the leading format byte the binary protocol writes, so decoding it is `wacore_binary::marshal::unmarshal_packed_ref(&plaintext)`. This is the plaintext handed to noise encryption, not a transport frame — the length prefix and the AEAD tag are added after it, and only `SessionStats` (see [Observability](/advanced/observability)) accounts for those. A `Bytes`, so forwarding it is a refcount bump, not a copy. + +This is the outbound counterpart of [`RawNode`](#rawnode) and a library extension with no WhatsApp Web equivalent. Before it existed, the send side had no observer at all: `wait_for_sent_node` is a filtered one-shot waiter for a single expected stanza, and it only ever sees stanzas that pass through the marshal-and-send path, so the paths that hand pre-marshaled bytes straight to the socket — acks, delivery receipts, direct-encoded IQs — were invisible even to that. `SentFrame` is emitted from the noise sender task, the single point every send path crosses (including the paths above and the coalesced-burst writes the ack and receipt workers use), so it covers all of them from one place. It exists for recording a session for replay, asserting the wire form in an integration test, and reading back a stanza the server rejected without rebuilding with logging on. + +Handshake frames are pre-noise and are not covered, and neither are VoIP relay sockets, which — like `SessionStats` — receive no observers at all. + +**Example:** +```rust +let _lease = client.acquire_sent_frame_forwarding(); + +// In your event handler: +Event::SentFrame(frame) => { + println!("sent {} bytes", frame.plaintext.len()); +} +// Drop `_lease` to stop forwarding. +``` + + +`plaintext` is skipped during serialization (`#[serde(skip)]`) — like `RawNode` and `DecryptedPayload`, `Serialize` on an event is for diagnostics, and no text format carries raw bytes without an encoding choice this type has no business making. + + +See [`acquire_sent_frame_forwarding`](/api/client#acquire_sent_frame_forwarding) for the lease API, and [WebSocket & Noise Protocol Handling](/advanced/websocket-handling#noisesocket) for how `SendObservers` wires it into the noise sender. + ## Event handler patterns ### Bot builder pattern From c2a3d0935d4ae5c4ef9ab6ee0517f37f20b87682 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 12:55:06 +0000 Subject: [PATCH 2/4] docs: fix stale do_handshake walkthrough call and SessionStats cross-reference Per Greptile review on PR #502: - The Connect walkthrough's do_handshake call still showed the pre-SendObservers argument list; updated it to match the real call site in src/client/lifecycle.rs. - SentFrame's SessionStats reference pointed at advanced/observability.mdx, which doesn't document SessionStats; point it at the websocket-handling page that actually covers it. --- advanced/websocket-handling.mdx | 9 ++++++--- concepts/events.mdx | 2 +- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/advanced/websocket-handling.mdx b/advanced/websocket-handling.mdx index 4038743..fcfa4b8 100644 --- a/advanced/websocket-handling.mdx +++ b/advanced/websocket-handling.mdx @@ -811,11 +811,14 @@ impl Client { ).await??; // Perform Noise handshake - let device = self.persistence_manager.get_device_snapshot(); let noise_socket = do_handshake( - &device, + self.runtime.clone(), + &self.persistence_manager, + &self.ik_handshake_failures, transport, - &mut events + &mut events, + SendObservers::with_stats(self.stats.clone()) + .with_sent_frames(self.sent_frame_tap.clone()), ).await?; // Store socket and start receivers diff --git a/concepts/events.mdx b/concepts/events.mdx index 9a88bf7..f6e820b 100644 --- a/concepts/events.mdx +++ b/concepts/events.mdx @@ -2721,7 +2721,7 @@ Event::SentFrame(SentFrame) ``` **Fields:** -- `plaintext` — The marshaled stanza, keeping the leading format byte the binary protocol writes, so decoding it is `wacore_binary::marshal::unmarshal_packed_ref(&plaintext)`. This is the plaintext handed to noise encryption, not a transport frame — the length prefix and the AEAD tag are added after it, and only `SessionStats` (see [Observability](/advanced/observability)) accounts for those. A `Bytes`, so forwarding it is a refcount bump, not a copy. +- `plaintext` — The marshaled stanza, keeping the leading format byte the binary protocol writes, so decoding it is `wacore_binary::marshal::unmarshal_packed_ref(&plaintext)`. This is the plaintext handed to noise encryption, not a transport frame — the length prefix and the AEAD tag are added after it, and only `SessionStats` (see [WebSocket & Noise Protocol Handling](/advanced/websocket-handling#dedicated-sender-task)) accounts for those. A `Bytes`, so forwarding it is a refcount bump, not a copy. This is the outbound counterpart of [`RawNode`](#rawnode) and a library extension with no WhatsApp Web equivalent. Before it existed, the send side had no observer at all: `wait_for_sent_node` is a filtered one-shot waiter for a single expected stanza, and it only ever sees stanzas that pass through the marshal-and-send path, so the paths that hand pre-marshaled bytes straight to the socket — acks, delivery receipts, direct-encoded IQs — were invisible even to that. `SentFrame` is emitted from the noise sender task, the single point every send path crosses (including the paths above and the coalesced-burst writes the ack and receipt workers use), so it covers all of them from one place. It exists for recording a session for replay, asserting the wire form in an integration test, and reading back a stanza the server rejected without rebuilding with logging on. From 9ddbddbec4a57bf88a1ff216e308d5301884647a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 12:59:32 +0000 Subject: [PATCH 3/4] docs(client): register the handler before acquiring the sent-frame lease Per Codex review on PR #502: acquiring the lease first activates forwarding immediately, so a frame sent between the lease call and register_handler would dispatch to no one. Swap the order in the example. --- api/client.mdx | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/api/client.mdx b/api/client.mdx index cc7e7d0..7b8d0f5 100644 --- a/api/client.mdx +++ b/api/client.mdx @@ -2058,11 +2058,14 @@ While no lease is held, nothing is emitted and nothing is cloned — the path co ```rust use wacore::types::events::{ChannelEventHandler, Event}; -let _lease = client.acquire_sent_frame_forwarding(); - let (handler, event_rx) = ChannelEventHandler::new(); client.register_handler(handler); +// Register the handler before acquiring the lease: forwarding activates +// immediately, so a frame sent between the two calls would otherwise dispatch +// to no one. +let _lease = client.acquire_sent_frame_forwarding(); + while let Ok(event) = event_rx.recv().await { if let Event::SentFrame(frame) = &*event { println!("sent {} bytes", frame.plaintext.len()); From f803956ffc6851f43c3623686ef0d8c0cf3971d5 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 8 Aug 2026 13:03:11 +0000 Subject: [PATCH 4/4] docs(websocket): clarify with_sent_frames is crate-internal wiring Per cubic review on PR #502: the breaking-change note described chaining .with_sent_frames(...) without flagging that it (and sent_frame_tap) are pub(crate), so a reader outside the crate could try to call it and hit a privacy error. Spell out the pub/pub(crate) split and point at Client::acquire_sent_frame_forwarding() as the actual public entry point. --- advanced/websocket-handling.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/advanced/websocket-handling.mdx b/advanced/websocket-handling.mdx index fcfa4b8..bb6962c 100644 --- a/advanced/websocket-handling.mdx +++ b/advanced/websocket-handling.mdx @@ -389,7 +389,7 @@ impl NoiseSocket { `NoiseSocket::new` (used by the illustrative handshake walkthrough above) is unchanged: it's a thin wrapper that calls `with_observers` with `SendObservers::default()`, for the callers — mainly tests — that want neither observer. -**Breaking change:** `with_stats(..., stats: Option>)` is now `with_observers(..., observers: SendObservers)`. `SendObservers` is one struct rather than one parameter per observer, so the next thing that wants to watch sends — [`SentFrame`](/concepts/events#sentframe) was the first — plugs in there instead of widening this constructor (and `do_handshake`'s) again. A caller that only wants what `with_stats` gave it passes `SendObservers::with_stats(stats)`; the main WA session socket also chains `.with_sent_frames(client.sent_frame_tap.clone())` to wire up sent-frame forwarding, gated by [`Client::acquire_sent_frame_forwarding()`](/api/client#acquire_sent_frame_forwarding). VoIP relay sockets and most tests pass `SendObservers::default()`, reporting to neither — same as passing `None` before. +**Breaking change:** `with_stats(..., stats: Option>)` is now `with_observers(..., observers: SendObservers)`. `SendObservers` is one struct rather than one parameter per observer, so the next thing that wants to watch sends — [`SentFrame`](/concepts/events#sentframe) was the first — plugs in there instead of widening this constructor (and `do_handshake`'s) again. A caller that only wants what `with_stats` gave it passes `SendObservers::with_stats(stats)`, which is `pub` and usable from any crate. `with_sent_frames` and the client's `sent_frame_tap` field, by contrast, are `pub(crate)` — internal wiring the client itself uses to chain `.with_sent_frames(client.sent_frame_tap.clone())` when it builds its own socket, not something callable from outside `whatsapp-rust`. An embedder enables sent-frame forwarding the same way any consumer does: through [`Client::acquire_sent_frame_forwarding()`](/api/client#acquire_sent_frame_forwarding), which wires the tap internally. VoIP relay sockets and most tests pass `SendObservers::default()`, reporting to neither — same as passing `None` before. **Why a dedicated task?**