> ## Documentation Index
> Fetch the complete documentation index at: https://whatsapp-rust.jlucaso.com/llms.txt
> Use this file to discover all available pages before exploring further.

# WebSocket & Noise Protocol

> NoiseSocket, connection management, handshake protocol, and frame handling in whatsapp-rust

## Overview

whatsapp-rust uses WebSocket for transport and the Noise Protocol for encryption. All messages are encrypted at the transport layer before being sent over the network.

## Architecture

The WebSocket handling system has several layers:

```
src/
├── handshake.rs           # Noise protocol handshake
├── socket/
│   ├── noise_socket.rs    # Encrypted communication layer
│   ├── mod.rs            # Re-exports
│   └── error.rs          # Socket errors
├── transport/            # WebSocket transport abstraction
└── client.rs            # High-level client orchestration
```

## Noise protocol handshake

The handshake establishes an encrypted channel using the Noise Protocol. whatsapp-rust supports two interactive patterns to match WhatsApp Web's behavior:

* **Noise XX** — three-message mutual authentication used on cold start, after pairing, and as a fallback when IK fails. The server's static public key is unknown ahead of time and is delivered (and verified against a cert chain) inside the `ServerHello`.
* **Noise IK** — a single round-trip resumed handshake used when a previously verified server static key is cached. This saves a server round trip on reconnect by pre-encrypting the client static and the 0-RTT `ClientPayload` against the cached server static.
* **Noise XXfallback** — a recovery pattern the server transparently triggers when the cached server static used by an IK attempt no longer matches its current key. The transcript pivots from IK to XX without restarting the connection.

A 20-second timeout (`NOISE_HANDSHAKE_RESPONSE_TIMEOUT`) is applied when waiting for the server's handshake response, ensuring the client does not hang indefinitely if the server is unresponsive.

### Pattern selection

`do_handshake` picks the pattern based on cached state:

```rust theme={null}
// src/handshake.rs
enum HandshakePattern {
    /// Cold start / pairing / forced fallback after an earlier IK failure.
    Xx,
    /// Cached server static + valid cert chain available; attempt IK.
    Ik([u8; 32]),
}
```

A handshake uses **IK** only when **all** of the following are true:

* The device is registered (paired).
* A cached `server_cert_chain` is present in `Device`.
* Both the leaf and intermediate certificates are inside their `not_before`/`not_after` validity window for the current wall-clock time.
* The process has not already observed an IK failure this session — the client allows at most `IK_FAILURE_THRESHOLD = 1` IK failure per process before forcing XX on subsequent connects.

Otherwise, the client falls back to **XX**. A successful XX (or XXfallback) handshake refreshes the cached `server_cert_chain` so that the next reconnect can attempt IK again.

### Handshake state types

The `wacore_noise` crate exposes one state machine per pattern:

| Type                       | Pattern    | Used by                                       |
| -------------------------- | ---------- | --------------------------------------------- |
| `XxHandshakeState`         | XX         | `run_xx_handshake`                            |
| `IkHandshakeState`         | IK         | `run_ik_handshake`                            |
| `XxFallbackHandshakeState` | XXfallback | Reentry from `IkServerHelloOutcome::Fallback` |

```rust theme={null}
// wacore/noise/src/handshake.rs
pub struct XxHandshakeState { /* ... */ }
pub struct IkHandshakeState { /* ... */ }
pub struct XxFallbackHandshakeState { /* ... */ }

pub enum IkServerHelloOutcome {
    /// Server accepted the cached static — IK can complete in one round trip.
    Continue(Box<IkHandshakeOutcome>),
    /// Server rejected the cached static; pivot the transcript into XXfallback.
    Fallback(Box<IkFallbackInputs>),
}
```

The `pattern` strings are `"Noise_XX_25519_AESGCM_SHA256\0\0\0\0"` for XX and XXfallback (the longer `"Noise_XXfallback_25519_AESGCM_SHA256"` name is hashed to derive `h0` because it exceeds `HASHLEN`).

### XX handshake (cold start / fallback)

```rust theme={null}
// src/handshake.rs
pub async fn do_handshake(
    runtime: Arc<dyn Runtime>,
    persistence_manager: &PersistenceManager,
    ik_handshake_failures: &AtomicU32,
    transport: Arc<dyn Transport>,
    transport_events: &mut async_channel::Receiver<TransportEvent>,
) -> Result<Arc<NoiseSocket>>
```

**Step-by-step process for XX:**

1. **Prepare client payload:**
   ```rust theme={null}
   let client_payload = device.get_client_payload().encode_to_vec();
   ```
   The payload contains client version, platform, and device details.

2. **Initialize XX state:**
   ```rust theme={null}
   let mut handshake_state = XxHandshakeState::new(
       device.noise_key.clone(),
       client_payload,
       &WA_CONN_HEADER,      // [0x57, 0x41] ("WA")
   )?;
   ```

3. **Send ClientHello:**
   ```rust theme={null}
   let client_hello_bytes = handshake_state.build_client_hello()?;

   // Optionally include edge routing for faster reconnection
   let (header, used_edge_routing) =
       build_handshake_header(device.edge_routing_info.as_deref());

   let framed = wacore::framing::encode_frame(
       &client_hello_bytes,
       Some(&header)
   )?;
   transport.send(framed).await?;
   ```

4. **Receive ServerHello** (with 20s timeout):
   ```rust theme={null}
   const NOISE_HANDSHAKE_RESPONSE_TIMEOUT: Duration = Duration::from_secs(20);

   let resp_frame = loop {
       match timeout(NOISE_HANDSHAKE_RESPONSE_TIMEOUT, transport_events.recv()).await {
           Ok(Ok(TransportEvent::DataReceived(data))) => {
               frame_decoder.feed(&data);
               if let Some(frame) = frame_decoder.decode_frame() {
                   break frame;
               }
           }
           // Handle errors...
       }
   };
   ```

5. **Send ClientFinish:**
   ```rust theme={null}
   let client_finish_bytes = handshake_state
       .read_server_hello_and_build_client_finish(&resp_frame)?;

   let framed = wacore::framing::encode_frame(&client_finish_bytes, None)?;
   transport.send(framed).await?;
   ```

6. **Complete handshake and persist the cert chain:**
   ```rust theme={null}
   let outcome = handshake_state.finish()?;
   info!("Handshake complete (XX), switching to encrypted communication");

   // The verified server cert chain is persisted so the next
   // connect can attempt Noise IK.
   persistence_manager
       .process_command(DeviceCommand::SetServerCertChain(
           outcome.server_cert_chain.into(),
       ))
       .await;

   Ok(Arc::new(NoiseSocket::new(
       runtime, transport, outcome.write_cipher, outcome.read_cipher,
   )))
   ```

### IK handshake (resumed)

When `select_pattern` returns `HandshakePattern::Ik(server_static_pub)`, the client uses the cached server static and skips the second round trip:

```rust theme={null}
// src/handshake.rs:run_ik_handshake
let mut ik = IkHandshakeState::new(
    device.noise_key.clone(),
    server_static_pub,
    client_payload,
    &WA_CONN_HEADER,
)?;

// IK ClientHello carries: ephemeral, encrypted client static, encrypted 0-RTT payload.
let client_hello_bytes = ik.build_client_hello()?;
send_first_handshake_message(&transport, device, &client_hello_bytes).await?;

let resp_frame = recv_frame(runtime, transport_events, &mut frame_decoder).await?;

match ik.read_server_hello(&resp_frame)? {
    IkServerHelloOutcome::Continue(out) => {
        // Single round trip — handshake is already complete.
        info!("Handshake complete (IK), switching to encrypted communication");
        // No client cert chain refresh — the cached entry stays authoritative.
    }
    IkServerHelloOutcome::Fallback(inputs) => {
        // Server rejected the cached static. Pivot to XXfallback below.
    }
}
```

On `Continue`, no `SetServerCertChain` command is issued — the on-disk cache remains authoritative.

### XXfallback (server-driven recovery)

If `read_server_hello` returns `IkServerHelloOutcome::Fallback(inputs)`, the server has signalled that the cached static is stale. The client constructs an `XxFallbackHandshakeState` from the IK transcript and finishes the handshake as if it had been XX from the start, **without reconnecting**:

```rust theme={null}
// src/handshake.rs
let mut fb = XxFallbackHandshakeState::from_ik_failure(*inputs, &WA_CONN_HEADER)?;
let client_finish_bytes = fb.build_client_finish()?;

let framed = wacore::framing::encode_frame(&client_finish_bytes, None)?;
transport.send(bytes::Bytes::from(framed)).await?;

let outcome = fb.finish()?;
info!("Handshake complete (XXfallback), switching to encrypted communication");

// XXfallback re-derives the server cert chain, so persist the fresh value.
persistence_manager
    .process_command(DeviceCommand::SetServerCertChain(
        outcome.server_cert_chain.into(),
    ))
    .await;
```

A `fallback_taken` flag is flipped before any operation that could fail. Failures **after** this pivot are not treated as crypto-fatal for IK cache invalidation purposes — by that point the server has already accepted the IK ClientHello and the cache is no longer the implicated party.

### IK failure handling

Errors are partitioned into two buckets in `HandshakeError`:

```rust theme={null}
impl HandshakeError {
    /// Transient — never invalidate the cache.
    pub fn is_transient(&self) -> bool { /* Transport, Timeout, Disconnected, StreamClosed */ }

    /// Crypto-fatal — the cached server static or cert chain is no longer trustworthy.
    pub fn is_crypto_fatal(&self) -> bool { /* AEAD/cert/parse failures */ }
}
```

If `do_handshake` selected IK, the pivot to XXfallback was not taken, and the error returns `is_crypto_fatal() == true`, the client:

1. Increments a process-local `ik_handshake_failures: AtomicU32`.
2. Issues `DeviceCommand::ClearServerCertChain` to drop the cached chain.
3. Forces XX on the next connect via `select_pattern`.

Programmer-side variants (`Proto`, generic `Crypto(String)`, `HkdfExpandFailed`, `CounterExhausted`) are explicitly **not** crypto-fatal — they indicate a code defect, and clearing the cache would mask it.

Location: `src/handshake.rs`

### Edge routing pre-intro

For optimized reconnection, the client can include edge routing info in the initial frame:

```rust theme={null}
pub fn build_handshake_header(
    edge_routing_info: Option<&[u8]>
) -> (Vec<u8>, bool) {
    let mut header = WA_CONN_HEADER.to_vec();  // [0x57, 0x41]
    
    if let Some(info) = edge_routing_info {
        if info.len() <= 8192 {  // Max size
            header.extend_from_slice(info);
            return (header, true);
        }
    }
    
    (header, false)
}
```

Location: `wacore/noise/src/edge_routing.rs`

Both XX and IK send their first handshake message through `send_first_handshake_message`, so the prologue (and any edge-routing pre-intro) is identical for the two patterns. This is required for the wire-side server to re-derive `h0` for transcript MAC checks regardless of which pattern is in use.

### Handshake Errors

```rust theme={null}
pub enum HandshakeError {
    Transport(#[from] anyhow::Error),
    Core(#[from] CoreHandshakeError),
    Timeout,
    /// Producer side of `transport_events` was dropped — distinct from a
    /// timeout because nothing more will ever arrive on the channel.
    StreamClosed,
    Disconnected,
    UnexpectedEvent(String),
}
```

Location: `src/handshake.rs`

## NoiseSocket

The `NoiseSocket` provides encrypted send/receive operations after handshake.

### Architecture

```rust theme={null}
pub struct NoiseSocket {
    read_key: Arc<NoiseCipher>,
    read_counter: Arc<AtomicU32>,
    send_job_tx: mpsc::Sender<SendJob>,
    sender_task_handle: JoinHandle<()>,
}

struct SendJob {
    plaintext_buf: Vec<u8>,
    out_buf: Vec<u8>,
    response_tx: oneshot::Sender<SendResult>,
}
```

Location: `src/socket/noise_socket.rs:21-32`

### Design Patterns

#### Dedicated sender task

The socket uses a dedicated task for sending to ensure frame ordering:

```rust theme={null}
impl NoiseSocket {
    pub fn new(
        transport: Arc<dyn Transport>,
        write_key: NoiseCipher,
        read_key: NoiseCipher,
    ) -> Self {
        let (send_job_tx, send_job_rx) = mpsc::channel::<SendJob>(32);
        
        // Spawn dedicated sender task
        let sender_task_handle = tokio::spawn(
            Self::sender_task(transport, write_key, send_job_rx)
        );
        
        Self {
            read_key: Arc::new(read_key),
            read_counter: Arc::new(AtomicU32::new(0)),
            send_job_tx,
            sender_task_handle,
        }
    }
}
```

**Why a dedicated task?**

1. **Ordering guarantee**: Frames must be sent with sequential counters
2. **Non-blocking**: Callers don't block on encryption or network I/O
3. **Backpressure**: Channel capacity (32) prevents unbounded queuing

Location: `src/socket/noise_socket.rs:34-62`

#### Sender task implementation

```rust theme={null}
async fn sender_task(
    transport: Arc<dyn Transport>,
    write_key: Arc<NoiseCipher>,
    mut send_job_rx: mpsc::Receiver<SendJob>,
) {
    let mut write_counter: u32 = 0;
    
    while let Some(job) = send_job_rx.recv().await {
        let result = Self::process_send_job(
            &transport,
            &write_key,
            &mut write_counter,
            job.plaintext_buf,
            job.out_buf,
        ).await;
        
        // Return buffers to caller for reuse
        let _ = job.response_tx.send(result);
    }
}
```

Location: `src/socket/noise_socket.rs:64-89`

### Encryption

#### Small Messages (≤16KB)

Encrypted inline to avoid thread pool overhead:

```rust theme={null}
if plaintext_buf.len() <= INLINE_ENCRYPT_THRESHOLD {
    // Copy to output buffer and encrypt in-place
    out_buf.clear();
    out_buf.extend_from_slice(&plaintext_buf);
    plaintext_buf.clear();
    
    write_key.encrypt_in_place_with_counter(counter, &mut out_buf)?;
    
    // Frame the ciphertext
    wacore::framing::encode_frame_into(&ciphertext, None, &mut out_buf)?;
}
```

Location: `src/socket/noise_socket.rs:103-130`

#### Large Messages (>16KB)

Offloaded to blocking thread pool:

```rust theme={null}
else {
    let plaintext_arc = Arc::new(plaintext_buf);
    let plaintext_arc_for_task = plaintext_arc.clone();
    
    // Offload to blocking thread
    let spawn_result = tokio::task::spawn_blocking(move || {
        write_key.encrypt_with_counter(
            counter, 
            &plaintext_arc_for_task[..]
        )
    }).await;
    
    // Recover original buffer
    plaintext_buf = Arc::try_unwrap(plaintext_arc)
        .unwrap_or_else(|arc| (*arc).clone());
    
    let ciphertext = spawn_result??;
    
    // Frame and send
    wacore::framing::encode_frame_into(&ciphertext, None, &mut out_buf)?;
    transport.send(out_buf).await?;
}
```

Location: `src/socket/noise_socket.rs:131-164`

<Note>
  The 16KB threshold is chosen based on benchmarking. Smaller messages benefit from inline encryption (no thread spawning overhead), while larger messages benefit from parallel execution.
</Note>

### Send API

```rust theme={null}
pub async fn encrypt_and_send(
    &self, 
    plaintext_buf: Vec<u8>, 
    out_buf: Vec<u8>
) -> SendResult {
    let (response_tx, response_rx) = oneshot::channel();
    
    let job = SendJob {
        plaintext_buf,
        out_buf,
        response_tx,
    };
    
    // Send to dedicated task
    if let Err(send_err) = self.send_job_tx.send(job).await {
        let job = send_err.0;
        return Err(EncryptSendError::channel_closed(
            job.plaintext_buf,
            job.out_buf,
        ));
    }
    
    // Wait for result
    match response_rx.await {
        Ok(result) => result,
        Err(_) => Err(EncryptSendError::channel_closed(
            Vec::new(), Vec::new()
        )),
    }
}
```

**Buffer Management:**

The API returns both buffers for reuse:

```rust theme={null}
type SendResult = Result<(Vec<u8>, Vec<u8>), EncryptSendError>;
//                        ^^^^^^^^^^^^^^^^^^^^^                  
//                        (plaintext_buf, out_buf) for reuse
```

This enables the client to reuse buffers across multiple sends:

```rust theme={null}
let mut plaintext_buf = Vec::with_capacity(1024);
let mut out_buf = Vec::with_capacity(1056);  // plaintext + 32 overhead

loop {
    plaintext_buf.clear();
    marshal_to_vec(&message, &mut plaintext_buf)?;
    
    (plaintext_buf, out_buf) = 
        noise_socket.encrypt_and_send(plaintext_buf, out_buf).await?;
}
```

Location: `src/socket/noise_socket.rs:173-200`

### Decryption

```rust theme={null}
pub fn decrypt_frame(&self, ciphertext: &[u8]) -> Result<Vec<u8>> {
    let counter = self.read_counter.fetch_add(1, Ordering::SeqCst);
    self.read_key
        .decrypt_with_counter(counter, ciphertext)
        .map_err(SocketError::Cipher)
}
```

**Decryption is synchronous** because:

1. Frames arrive sequentially in the transport receiver
2. Decryption is fast (AES-GCM hardware acceleration)
3. No ordering concerns (unlike send)

Location: `src/socket/noise_socket.rs:202-207`

### Cleanup

```rust theme={null}
impl Drop for NoiseSocket {
    fn drop(&mut self) {
        // Abort sender task to prevent resource leaks
        self.sender_task_handle.abort();
    }
}
```

Location: `src/socket/noise_socket.rs:210-217`

## Frame Protocol

Messages are framed before encryption:

```rust theme={null}
pub fn encode_frame(
    payload: &[u8], 
    header: Option<&[u8]>
) -> Result<Vec<u8>> {
    let mut output = Vec::new();
    
    if let Some(header) = header {
        output.extend_from_slice(header);
    }
    
    // 3-byte big-endian length
    let len = payload.len() as u32;
    output.push(((len >> 16) & 0xFF) as u8);
    output.push(((len >> 8) & 0xFF) as u8);
    output.push((len & 0xFF) as u8);
    
    output.extend_from_slice(payload);
    Ok(output)
}
```

Location: `wacore/src/framing/mod.rs:10-30`

### Frame Format

```
┌────────────────┬─────────────────┬────────────────────┐
│ Optional Header│  Length (3 bytes)│     Payload       │
│  (handshake)   │   (big-endian)  │   (encrypted)     │
└────────────────┴─────────────────┴────────────────────┘
```

**Frame length limits:**

* Maximum frame size: 16MB (enforced by framing layer)
* Typical message frame: \< 1KB
* Media messages: 100KB - 2MB (encrypted metadata)

### Frame Decoder

`FrameDecoder` (in `wacore/noise/src/framing.rs`) accumulates inbound bytes and splits out complete frames. It has two feed methods:

* **`feed(&[u8])`** — borrow path; always copies into the internal `BytesMut` staging buffer. Used during the handshake and in tests.
* **`feed_bytes(Bytes)`** — owned path; adopts the payload's allocation without copying in steady state (buffer empty + sole reference). Falls back to copying when the buffer has a partial frame or the payload is shared (refcount > 1).

```rust theme={null}
use bytes::{Bytes, BytesMut};

pub struct FrameDecoder {
    buffer: BytesMut,
}

impl FrameDecoder {
    /// Borrow path — used during the handshake and by existing tests.
    pub fn feed(&mut self, data: &[u8]) {
        self.buffer.extend_from_slice(data);
    }

    /// Owned path — adopts the payload's allocation zero-copy in steady state
    /// (buffer empty + sole reference). Falls back to `feed` otherwise.
    pub fn feed_bytes(&mut self, data: Bytes) {
        if self.buffer.is_empty() {
            match data.try_into_mut() {
                Ok(owned) => self.buffer = owned,
                Err(shared) => self.buffer.extend_from_slice(&shared),
            }
        } else {
            self.buffer.extend_from_slice(&data);
        }
    }

    pub fn decode_frame(&mut self) -> Option<BytesMut> {
        if self.buffer.len() < FRAME_LENGTH_SIZE {
            return None;
        }

        let len = (u32::from(self.buffer[0]) << 16
            | u32::from(self.buffer[1]) << 8
            | u32::from(self.buffer[2])) as usize;

        if self.buffer.len() < FRAME_LENGTH_SIZE + len {
            return None;
        }

        let _ = self.buffer.split_to(FRAME_LENGTH_SIZE);
        Some(self.buffer.split_to(len))
    }
}
```

Location: `wacore/noise/src/framing.rs`

**Steady-state zero-copy guarantee:** because the WebSocket transport delivers each message as its own `Bytes`, in steady state the staging buffer is empty when `feed_bytes` is called and the payload is the sole reference. `try_into_mut` succeeds, the bytes are adopted wholesale, and the decoded frame's slice points directly into the original WebSocket allocation at offset `FRAME_LENGTH_SIZE` — verified by pointer identity in `test_feed_bytes_adopts_unique_payload_without_copy`.

## Transport Abstraction

The transport layer abstracts WebSocket implementation:

```rust theme={null}
use bytes::Bytes;

#[async_trait]
pub trait Transport: Send + Sync {
    async fn send(&self, data: Vec<u8>) -> Result<(), anyhow::Error>;
    async fn disconnect(&self);
}

pub enum TransportEvent {
    Connected,
    Disconnected,
    /// Inbound payload. Carrying `Bytes` (rather than `Vec<u8>`) lets the
    /// receive path adopt the allocation zero-copy via `feed_bytes`.
    DataReceived(Bytes),
}
```

Location: `src/transport/mod.rs`

### WebSocket implementation

The transport is generic over any `AsyncRead + AsyncWrite` stream. The `from_websocket` function wraps an already-upgraded `WebSocketStream` into a `Transport` + event channel:

```rust theme={null}
pub fn from_websocket<S>(
    ws: WebSocketStream<S>,
) -> (Arc<dyn Transport>, async_channel::Receiver<TransportEvent>)
where
    S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
```

Internally, the WebSocket is split into a write half (guarded by `Arc<Mutex>`) and a read half (moved to a spawned `read_pump` task). A `watch` channel coordinates graceful shutdown between the transport and the read pump.

`TokioWebSocketTransportFactory` handles the default DNS/TCP/TLS connection and delegates to `from_websocket`. For custom connection strategies (IPv4 preference, TCP keepalive, proxies), call `from_websocket` directly.

Location: `transports/tokio-transport/src/lib.rs`

## Connection Lifecycle

### Connect timeout

Both the transport connection and the version fetch are wrapped in a 20-second timeout (`TRANSPORT_CONNECT_TIMEOUT`), matching WhatsApp Web's MQTT `CONNECT_TIMEOUT` and DGW `connectTimeoutMs` defaults. Without this, a dead network would block on the OS TCP SYN timeout (\~60-75s).

```rust theme={null}
const TRANSPORT_CONNECT_TIMEOUT: Duration = Duration::from_secs(20);
```

The client runs the version fetch and transport connection **in parallel** using `tokio::join!`, both under this timeout:

```rust theme={null}
let version_future = tokio::time::timeout(
    TRANSPORT_CONNECT_TIMEOUT,
    resolve_and_update_version(&persistence_manager, &http_client, override_version),
);
let transport_future = tokio::time::timeout(
    TRANSPORT_CONNECT_TIMEOUT,
    transport_factory.create_transport(),
);

let (version_result, transport_result) = tokio::join!(version_future, transport_future);
```

If either times out, the connection attempt fails with a descriptive error (e.g., `"Transport connect timed out after 20s"`).

Location: `src/client.rs:108-877`

### 1. Connect

```rust theme={null}
impl Client {
    pub async fn connect(&self) -> Result<()> {
        // Version fetch + transport connection run in parallel, both under 20s timeout
        let (transport, mut events) = tokio::time::timeout(
            TRANSPORT_CONNECT_TIMEOUT,
            self.transport_factory.create_transport(),
        ).await??;
        
        // Perform Noise handshake
        let device = self.persistence_manager.get_device_snapshot();
        let noise_socket = do_handshake(
            &device,
            transport,
            &mut events
        ).await?;
        
        // Store socket and start receivers
        self.noise_socket.store(Some(Arc::clone(&noise_socket)));
        self.start_frame_receiver(events, noise_socket).await;
        
        Ok(())
    }
}
```

### 2. Message loop (read loop)

The `read_messages_loop` runs on the `run()` caller's task — not spawned — so the keepalive loop (which runs in a separate spawned task) is never blocked by frame processing. This eliminates a class of bugs where a long-running batch of frames (e.g., offline sync) could starve the keepalive timer.

```rust theme={null}
async fn read_messages_loop(self: &Arc<Self>) -> Result<(), anyhow::Error> {
    let transport_events = self.transport_events.lock().await.take()
        .ok_or_else(|| anyhow!("Cannot start message loop: not connected"))?;
    // Noise socket is stable for the lifetime of this loop; resolve once
    // instead of locking the mutex on every inbound frame.
    let noise_socket = self.get_noise_socket().await
        .map_err(|_| anyhow!("Cannot start message loop: no noise socket"))?;
    let mut frame_decoder = FrameDecoder::new();

    loop {
        futures::select_biased! {
            _ = self.shutdown_notifier.listen().fuse() => {
                return Ok(());
            },
            event_result = transport_events.recv().fuse() => {
                match event_result {
                    Ok(TransportEvent::DataReceived(data)) => {
                        // Update dead-socket timer on arrival
                        self.last_data_received_ms.store(now_millis(), Ordering::Relaxed);
                        // Adopt the payload zero-copy when it is the sole reference
                        // and no partial frame is pending (steady state).
                        frame_decoder.feed_bytes(data);

                        let mut frames_in_batch: u32 = 0;
                        while let Some(encrypted_frame) = frame_decoder.decode_frame() {
                            if let Some(node) = self.decrypt_frame(&noise_socket, encrypted_frame) {
                                // Inline vs spawned processing (see below)
                                if is_critical(&node) {
                                    self.process_decrypted_node(node).await;
                                } else {
                                    self.runtime.spawn(/* ... */).detach();
                                }
                            }

                            // Cooperative yield every N frames
                            frames_in_batch += 1;
                            if frames_in_batch.is_multiple_of(self.runtime.yield_frequency()) {
                                if let Some(yield_fut) = self.runtime.yield_now() {
                                    yield_fut.await;
                                }
                            }
                        }

                        // Refresh timestamp after batch so keepalive sees
                        // batch completion time, not just arrival time
                        if frames_in_batch > 1 {
                            self.last_data_received_ms.store(now_millis(), Ordering::Relaxed);
                        }
                    },
                    Ok(TransportEvent::Disconnected) | Err(_) => { /* handle disconnect */ }
                    _ => {}
                }
            }
        }
    }
}
```

**Key design decisions:**

* **`select_biased!`** — the shutdown listener has priority over transport events, ensuring the loop exits promptly when `shutdown_notifier` fires (e.g., on stream error or disconnect)
* **Batch timestamp refresh** — after processing multiple frames, `last_data_received_ms` is updated again so the keepalive loop sees the batch completion time rather than the arrival time. This prevents false-positive dead-socket triggers during large offline sync batches that take seconds to drain
* **Cooperative yielding** — the loop yields to the runtime every `yield_frequency()` frames, preventing a large burst of frames from monopolizing the executor

#### Inline vs concurrent node processing

Frame decryption is always sequential (noise protocol counter ordering), but node processing uses a hybrid strategy:

| Node tag                             | Processing           | Reason                                                                                                                |
| ------------------------------------ | -------------------- | --------------------------------------------------------------------------------------------------------------------- |
| `success`, `failure`, `stream:error` | Inline               | Critical for connection state transitions                                                                             |
| `message`                            | Inline               | Preserves arrival order for per-chat queues (MessageHandler just enqueues + ACKs; heavy crypto runs in queue workers) |
| `ib`                                 | Inline               | Ensures offline sync tracking (expected count) is set up before offline messages are processed                        |
| Everything else                      | Spawned concurrently | Maximizes parallelism for non-ordering-sensitive stanzas                                                              |

````

### 3. Send message

```rust
impl Client {
    pub async fn send_node(&self, node: &Node) -> Result<()> {
        let noise_socket = self.noise_socket.load()
            .ok_or_else(|| anyhow!("not connected"))?;
        
        // Marshal node to binary with auto-sized buffer
        let plaintext_buf = marshal_auto(node)?;
        
        // Encrypt and send
        self.send_raw_bytes(plaintext_buf).await
    }
}
````

The `marshal_auto` function automatically selects an appropriate buffer capacity based on the node's characteristics. For nodes exceeding certain thresholds (24+ attributes, 64+ children, or 8KB+ scalar content), it pre-estimates the capacity to avoid reallocations. For typical small nodes, it uses the default 1024-byte capacity. This replaces the previous manual `Vec::with_capacity(1024)` + `marshal_to` pattern.

````

### 4. Disconnect

On disconnect, `cleanup_connection_state()` runs exactly once — in the `run()` method after the message loop exits — resetting all connection-scoped state:

```rust
impl Client {
    async fn cleanup_connection_state(&self) {
        self.shutdown_notifier.notify(usize::MAX);
        *self.transport.lock().await = None;
        *self.noise_socket.lock().await = None;
        self.is_connected.store(false, Ordering::Release);

        // Drop per-chat lane senders so workers exit via channel close.
        // Without this, stale workers from the old connection survive reconnects
        // holding outdated signal/crypto state.
        self.chat_lanes.invalidate_all();

        // Clear signal cache, pending retries, IQ waiters, offline sync state...
    }
}
````

Key cleanup actions include invalidating chat lanes (so stale message processing workers don't survive with outdated crypto state), clearing the signal cache, draining IQ response waiters, and resetting offline sync state. See [disconnect cleanup](/concepts/architecture#disconnect-cleanup) for the full list.

### Connection state tracking

The client tracks whether the noise socket is established using a dedicated `AtomicBool` (`is_connected`) rather than probing the noise socket mutex. This design prevents a TOCTOU race where `try_lock()` on the mutex fails due to contention (e.g., during frame encryption), not because the socket is absent — which previously caused `is_connected()` to return `false` on live connections, silently dropping receipt acks.

**State transitions:**

| Event                        | `is_connected` value | Ordering                           |
| ---------------------------- | -------------------- | ---------------------------------- |
| `connect()` start            | `false`              | `Relaxed` (reset)                  |
| Noise socket stored          | `true`               | `Release` (after socket is `Some`) |
| `cleanup_connection_state()` | `false`              | `Release` (after socket is `None`) |

The `Release`/`Acquire` ordering ensures that any task reading `is_connected() == true` is guaranteed to see the noise socket as `Some`, and any task reading `false` after cleanup sees the socket as `None`.

```rust theme={null}
// Lock-free connection check — never affected by mutex contention
pub fn is_connected(&self) -> bool {
    self.is_connected.load(Ordering::Acquire)
}
```

This is critical for the keepalive loop and stanza acknowledgment, both of which call `is_connected()` to decide whether to send data. Under the old `try_lock()` approach, concurrent `send_node()` calls holding the mutex would cause false negatives, leading to skipped keepalive pings or dropped ack stanzas.

## Error Handling

### Socket Errors

```rust theme={null}
pub enum SocketError {
    SocketClosed,
    Io(#[from] std::io::Error),
    Cipher(#[from] NoiseError),
    Marshal(#[source] BinaryError),
}

pub enum EncryptSendError {
    Crypto { error: anyhow::Error, ... },
    Transport { error: anyhow::Error, ... },
    Framing { error: FramingError, ... },
    Join { error: JoinError, ... },
    ChannelClosed { ... },
}
```

Each variant preserves the underlying typed error as a `source()`. `Cipher` wraps a `NoiseError` (from `wacore::handshake`), which itself carries a typed `CryptoProviderError` source. Walking the chain with `std::error::Error::source()` lets callers downcast to the original AES-GCM, libsignal, or binary-protocol error without parsing strings.

All variants return buffers for reuse:

```rust theme={null}
impl EncryptSendError {
    pub fn into_buffers(self) -> (Vec<u8>, Vec<u8>) {
        match self {
            Self::Crypto { plaintext_buf, out_buf, .. } => 
                (plaintext_buf, out_buf),
            Self::Transport { plaintext_buf, out_buf, .. } => 
                (plaintext_buf, out_buf),
            // ...
        }
    }
}
```

Location: `src/socket/error.rs`

### Stream error handling

When the server sends a `<stream:error>` stanza, it is processed inline (not spawned concurrently) because stream errors are critical for connection state. The `StanzaRouter` dispatches the node to a `StreamErrorHandler`, which calls `Client::handle_stream_error()`.

Each stream error sets `is_logged_in = false` and fires the `shutdown_notifier` to exit the keepalive loop and other background tasks.

**Error code behavior:**

| Code                                                                      | Action                                        | Event            | Reconnects?                            |
| ------------------------------------------------------------------------- | --------------------------------------------- | ---------------- | -------------------------------------- |
| **401**                                                                   | Disables auto-reconnect                       | `LoggedOut`      | No — session invalid, must re-pair     |
| **409**                                                                   | Disables auto-reconnect                       | `StreamReplaced` | No — prevents displacement loop        |
| **429**                                                                   | Adds 5 to backoff counter                     | None             | Yes — extended Fibonacci backoff       |
| **503**                                                                   | Normal handling                               | None             | Yes — standard backoff                 |
| **515**                                                                   | Marks as expected disconnect                  | None             | Yes — immediate, no backoff            |
| **516**                                                                   | Disables auto-reconnect                       | `LoggedOut`      | No — device removed                    |
| `<stream:error>` with an `<xml-not-well-formed/>` child (no numeric code) | Force-closes the socket                       | `StreamError`    | Yes — standard backoff                 |
| **500** / unknown / missing code                                          | Logs as warning, stays connected (since v0.6) | `StreamError`    | N/A — no disconnect; session preserved |

<Note>
  Before v0.6 the client treated every unknown stream-error code as fatal: it set `is_logged_in = false`, fired `shutdown_notifier`, and ended up in a "zombie" state where the connection survived but background tasks (keepalive, prekey upload) refused to run. v0.6 keeps the fatal set listed above and downgrades everything else (including code `500` and code-less `<stream:error><ack/>` routing wrappers) to a warning. `is_logged_in` stays `true`, the transport stays open, and any reconnect is driven by the server's `<xmlstreamend/>` — not by the stream-error handler itself.
</Note>

<Note>
  `<xml-not-well-formed/>` is checked before the ack-wrapper case above and is an exception to that downgrade: WA Web (`Handle/StreamError.js`) treats it as "bad xml, closing socket" (`CLOSE_SOCKET`). A malformed frame desyncs the stream, so the client proactively recycles the socket (`is_logged_in = false`, `should_disconnect = true`) instead of waiting for the server to end it. It counts toward the reconnect backoff like a normal disconnect — it is not an expected disconnect (515), so no immediate reconnect. Like every other case in this code-less/unknown-code branch, an `Event::StreamError` is still dispatched before the socket is closed — consumers see the event even though the connection is being force-recycled rather than gracefully downgraded.
</Note>

**Processing pipeline:**

```
Server sends <stream:error code="..."/>
  → read_messages_loop (inline, not spawned)
  → StanzaRouter → StreamErrorHandler
  → Client::handle_stream_error()
  → is_logged_in = false
  → Code-specific handling (see table above)
  → shutdown_notifier fires → keepalive exits
  → run() loop checks enable_auto_reconnect
```

### Stanza acknowledgment

The client automatically sends `<ack/>` nodes in response to incoming stanzas (messages, receipts, notifications, calls). The ack construction follows WhatsApp Web and whatsmeow behavior:

**Class gating for newsletter and status:** Newsletter and status\@broadcast inbound messages produce a `<ack class="message"/>` instead of a `<receipt>`, matching WA Web's `WAWebSendMsgAckOrReceiptJob`. Regular DMs and groups continue to skip the message-class ack because they ride on the regular receipt path.

**Ack attributes:**

| Attribute     | Value                                                                                                                                                                                                                            |
| ------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `class`       | Original stanza tag (e.g., `"message"`, `"receipt"`, `"notification"`)                                                                                                                                                           |
| `id`          | Copied from the incoming stanza                                                                                                                                                                                                  |
| `to`          | Flipped from the incoming `from` attribute                                                                                                                                                                                       |
| `participant` | Copied from the incoming stanza (when present)                                                                                                                                                                                   |
| `recipient`   | Echoed from the incoming stanza (when present) — required for LID-routed and hosted-companion messages so the server can route the ack back correctly. Stripping it caused `<stream:error>` disconnects for peer-routed traffic. |
| `from`        | Own device phone number JID — only included for message acks                                                                                                                                                                     |

**`type` attribute rules:**

The `type` attribute is handled differently depending on the stanza:

| Stanza                                                 | `type` in ack      | Reason                                                                                                           |
| ------------------------------------------------------ | ------------------ | ---------------------------------------------------------------------------------------------------------------- |
| `message`                                              | **Never included** | Matches whatsmeow: `node.Tag != "message"` guard skips type for messages                                         |
| `receipt` (with type, e.g. `"read"`)                   | **Echoed**         | WA Web echoes `type` when explicitly present                                                                     |
| `receipt` (without type, i.e. delivery)                | **Omitted**        | Delivery receipts have no `type`; including one (e.g., `type="delivery"`) causes `<stream:error>` disconnections |
| `notification`                                         | **Echoed**         | Type is echoed for most notifications                                                                            |
| `notification type="encrypt"` with `<identity/>` child | **Omitted**        | WA Web specifically drops `type` for identity-change notifications                                               |

<Warning>
  Sending incorrect `type` attributes in ack stanzas can cause the server to issue `<stream:error>` disconnections. The library handles this automatically — you don't need to build ack nodes manually.
</Warning>

Location: `src/client.rs` (`build_ack_node`, `is_encrypt_identity_notification`)

### Fibonacci backoff

The reconnection backoff follows the Fibonacci sequence, matching WhatsApp Web's behavior:

```
Sequence: 1s, 1s, 2s, 3s, 5s, 8s, 13s, 21s, 34s, 55s, 89s, 144s, ...
Maximum:  900s (15 minutes)
Jitter:   ±10%
```

For rate-limited errors (429), the backoff counter is incremented by 5 before the normal increment, causing the delay to jump significantly on the next reconnection attempt.

**Stability-gated reset.** The backoff counter does not reset to its base immediately on a successful `<success>` authentication. Instead, `connected_at_ms` records the auth time, and the counter only resets when the *next* disconnect finds the connection was stable for at least `STABLE_CONNECTION_RESET_MS` (30s) — matching WA Web's `resetDelay`. A connection that authenticates and then immediately drops keeps escalating the backoff instead of resetting to 1s and retrying in a tight loop.

An explicit penalty applied during the connection — a 429 rate-limit or a manual `Client::reconnect()` — sets `backoff_reset_suppressed`, which survives even a stable (≥30s) connection and prevents the next disconnect from erasing that deliberate backoff step (matching WA Web's `cancelReset()`). The suppression flag is cleared on the next successful `<success>`, so it does not carry over indefinitely. An expected disconnect (e.g., 515) resets `connected_at_ms` to 0 so a later failed connect cannot read the prior cycle's stale timestamp as "stable."

### Retry strategy

The `run()` method handles reconnection automatically:

```rust theme={null}
// Simplified reconnection logic in run()
loop {
    self.connect().await;
    self.read_messages_loop().await;

    if !self.enable_auto_reconnect.load(Ordering::Relaxed) {
        break; // 401, 409, 516 — stop permanently
    }

    if self.expected_disconnect.load(Ordering::Relaxed) {
        continue; // 515 — reconnect immediately
    }

    let errors = self.auto_reconnect_errors.fetch_add(1, Ordering::Relaxed);
    let delay = fibonacci_backoff(errors + 1);
    sleep(delay).await;
}
```

## Keepalive and dead socket detection

The keepalive loop monitors connection health, matching WhatsApp Web's behavior precisely.

### Constants

| Constant                       | Value | Description                                                      |
| ------------------------------ | ----- | ---------------------------------------------------------------- |
| `KEEP_ALIVE_INTERVAL_MIN`      | 15s   | Minimum interval between pings                                   |
| `KEEP_ALIVE_INTERVAL_MAX`      | 30s   | Maximum interval between pings                                   |
| `KEEP_ALIVE_RESPONSE_DEADLINE` | 20s   | Timeout waiting for pong response                                |
| `DEAD_SOCKET_TIME`             | 20s   | Max silence after the watchdog arms before declaring socket dead |

### Timestamp safety

All timestamp conversions from `now_millis()` (which returns `i64`) to `u64` are guarded with `.max(0)` before casting. This prevents silent wrap-around on negative clock values (e.g., from NTP corrections or virtualized environments) that would otherwise produce incorrect timestamps.

### Keepalive loop behavior

The loop runs every 15-30 seconds (randomized, matching WA Web's `15 * (1 + random())` formula) and performs these checks in order:

1. **Skip if recently active** — if data was received within `KEEP_ALIVE_INTERVAL_MIN` (15s), the connection is proven alive; skip the ping and reset the error counter
2. **Send keepalive ping** — sends the ping *before* the dead-socket check so that a successful pong updates `last_data_received_ms` and prevents false-positive dead-socket detection on idle-but-healthy connections
3. **RTT-adjusted clock skew** — on pong, calculates server time offset using the midpoint formula: `(startTime + rtt/2) / 1000 - serverTime`, matching WA Web's `onClockSkewUpdate`
4. **Skip ping when IQ pending** — if there are already pending IQ responses, the connection is implicitly being tested; skip the explicit ping

### Dead socket detection

Dead socket detection mirrors WA Web's `deadSocketTimer.onOrBefore` pattern, which keeps the **earliest** armed deadline rather than the most recent one:

* **Not armed** if nothing has been sent since the last receive (the anchor is zero)
* **Cancelled** if data was received after the anchor was armed
* **Fires** if `DEAD_SOCKET_TIME` (20s) has elapsed since the anchor with no receive since

The watchdog is anchored to `SessionStats::first_send_since_recv_ms` — the **first** send since the last receive, not the most recent send. `record_frame_sent` only stores a new anchor when the current one is unset or stale (`<=` the last-received timestamp); once armed, further sends leave it in place. Every receive resets the anchor to zero, and the next send re-arms it. Anchoring on the most recent send instead (the pre-fix behavior) let continued outgoing traffic — messages, receipts, presence — keep pushing the deadline forward, hiding a half-open socket (a peer that silently disappeared while writes still buffer and reads hang) for as long as the app kept emitting frames.

The dead-socket check runs on **every** keepalive tick — not just after a failed ping. This catches scenarios where pending IQs caused the ping to be skipped, or where the ping "succeeded" but the connection died immediately after. When a dead socket is detected, the client calls `reconnect_immediately()` and exits the keepalive loop.

WA Web's `deadSocketTimer.onOrBefore` (`WA/Shift/Timer.js`) arms on the first `callStanza` after a receive and is cancelled by `parseAndHandleStanza`; subsequent sends never push the deadline back out. The keepalive loop approximates this by checking `is_dead_socket_at(first_send_since_recv, last_recv, now)` unconditionally each iteration, where `first_send_since_recv` is the armed-anchor value described above. The tick reads the clock once into `now` and evaluates both the dead-socket check and the elapsed-time log message against that single instant, rather than re-reading the clock for each. There is no `last_data_sent_ms` field — nothing reads a "most recent send" timestamp, only the armed anchor.

### Error classification

Keepalive errors are classified exhaustively (compile-time enforced for new error variants):

| Error type                                                        | Classification | Behavior                                 |
| ----------------------------------------------------------------- | -------------- | ---------------------------------------- |
| `Socket`, `Disconnected`, `NotConnected`, `InternalChannelClosed` | Fatal          | Exit keepalive loop immediately          |
| `Timeout`, `ServerError`, `ParseError`                            | Transient      | Increment error count, check dead socket |

### Periodic maintenance

Approximately every 12 keepalive ticks (\~5 minutes), the keepalive loop runs background cleanup of expired sent messages from the database, based on `CacheConfig::sent_message_ttl_secs`.

## Performance Considerations

### Buffer Sizing

Optimal buffer capacity based on payload characteristics:

```rust theme={null}
// Encrypted size = plaintext + 16 (AES-GCM tag) + 3 (frame header)
let buffer_capacity = plaintext.len() + 32;  // Extra headroom
let out_buf = Vec::with_capacity(buffer_capacity);
```

Location: `src/socket/noise_socket.rs:373-407` (tests verify this formula)

### SIMD Encryption

The Noise cipher uses hardware AES acceleration when available:

```rust theme={null}
pub struct NoiseCipher {
    cipher: Aes256Gcm,  // Uses AES-NI on x86_64
}
```

### Zero-Copy Patterns

**Send path** — reuse the marshal and output buffers across sends:

```rust theme={null}
// Bad: Allocates new buffer
let data = node.to_bytes();
socket.send(data).await?;

// Good: Reuses buffer
let mut buf = Vec::with_capacity(1024);
marshal_to_vec(&node, &mut buf)?;
socket.send(buf).await?;
```

**Receive path** — `feed_bytes` adopts the WebSocket payload's allocation in steady state so inbound bytes cross the framing layer without a copy:

```rust theme={null}
// The transport delivers an owned Bytes; feed_bytes adopts it when
// the staging buffer is empty and the payload has no other references.
frame_decoder.feed_bytes(data);

// Each decoded frame then slices directly into the original
// WebSocket allocation — no intermediate staging copy.
while let Some(frame) = frame_decoder.decode_frame() { /* ... */ }
```

This eliminates one full `memcpy` of every inbound byte in the common case. Shared payloads and partial-frame continuations fall back to copying automatically.

## Testing

### Mock Transport

```rust theme={null}
pub struct MockTransport;

#[async_trait]
impl Transport for MockTransport {
    async fn send(&self, data: Vec<u8>) -> Result<()> {
        // Record for assertions
        Ok(())
    }
    async fn disconnect(&self) {}
}
```

Location: `src/transport/mock.rs`

### Test Cases

Key test scenarios:

```rust theme={null}
#[tokio::test]
async fn test_encrypt_and_send_returns_both_buffers()

#[tokio::test]
async fn test_concurrent_sends_maintain_order()

#[tokio::test]
async fn test_encrypted_buffer_sizing_is_sufficient()

#[tokio::test]
async fn test_handshake_with_edge_routing()
```

Location: `src/socket/noise_socket.rs:219-459`

## Related Components

* [Signal Protocol](/advanced/signal-protocol) - Message-level encryption
* [Binary Protocol](/advanced/binary-protocol) - Payload serialization
* [State Management](/advanced/state-management) - Connection state persistence

## References

* Handshake: `src/handshake.rs`
* NoiseSocket: `src/socket/noise_socket.rs`
* Framing: `wacore/noise/src/framing.rs`
* Transport: `src/transport/`
* [Noise Protocol Framework](https://noiseprotocol.org/noise.html)
* [Noise XX Pattern](https://noiseprotocol.org/noise.html#interactive-patterns)