[order-stream] Stale WebSocket cleanup can remove a replacement connection

[grandpig]

Summary

The order-stream WebSocket implementation supports reconnecting with the same address by replacing the existing entry in state.connections.

However, both WebSocket cleanup and broadcast cleanup remove connections by address only. This creates a race where cleanup belonging to an older connection can remove a newer replacement connection.

Race condition

1. Connection A is stored at connections[address].
2. Connection B reconnects with the same address and replaces A.
3. A later exits, or a broadcast task detects A's sender as closed.
4. Cleanup calls connections.remove(&address).
5. The entry for B is removed instead of A.

As a result, B's WebSocket may remain open while no longer being present in the server-side connection map. Subsequent orders will not be broadcast to B, and connection capacity statistics become inaccurate.

Additional observation

drop(old_connection.sender) does not necessarily terminate the old WebSocket task because websocket_connection still retains its own sender_channel clone.

Proposed fix

Assign every connection a unique ID or generation number:

struct ClientConnection {
    id: u64,
    sender: mpsc::Sender<String>,
}

Cleanup should remove the entry only when the stored connection ID matches the ID of the task performing cleanup. Broadcast snapshots should also retain and validate this ID.

The replacement path should additionally use an explicit cancellation mechanism, or avoid retaining an extra sender clone, so that the old WebSocket task is reliably terminated.

Tests

Add deterministic tests covering:

• stale cleanup of A after B replaces A;
• stale broadcast cleanup of A after B replaces A;
• confirmation that B remains in the connection map.

[linear]

BM-3095