---
applyTo: "src/network/**"
description: "Use when editing network code: connection lifecycle, pool management, priority channels, correlation ID tracking, reconnection, and lock discipline."
---

# Network Module Rules

## Connection Lifecycle

1. `connect()` → DNS → TCP → optional TLS upgrade → SASL handshake → fetch API versions → spawn event loop
2. Reader and writer are split; reader runs in a separate spawned task.
3. `close()` sends `ConnectionCommand::Close` via the **high-priority** channel.
4. On drop, pending requests are drained with "connection closed" errors.

Adding state to `BrokerConnection` → verify it is cleaned up in close and in the drop/drain path.

## Priority Channels

- **High-priority**: heartbeats, metadata, close commands — checked first via `try_recv()` + `biased` select.
- **Normal-priority**: produce, fetch — processed only when high-priority is empty.
- Never route user data through the high-priority channel.

## Correlation ID Tracking

- Generated by `CorrelationIdGenerator` (`AtomicI32::fetch_add`).
- Each request stored in `pending: HashMap<CorrelationId, PendingRequest>` with a `oneshot::Sender`.
- Response reader extracts correlation ID from bytes 0–3, removes entry, and resolves the oneshot.
- `send_fire_and_forget()` does **not** register a pending entry (acks=0 path).

## Pool & Reconnection

- Dual indexing: `connections` (by broker ID) and `connections_by_addr` (by address).
- `close_all()` must drain **both** maps to prevent bootstrap connection leaks.
- `reconnect_with_backoff()` drops all locks **before** network I/O — never hold RwLock across `.await`.
- Non-retriable errors (auth, protocol) fail immediately; only network errors retry.

## Request Timeout

- 1-second sweep task evicts timed-out pending requests — granularity is ±1 s.
- Per-request timeout stored as `Instant` in `PendingRequest.sent_at`.

## Wire Framing

- 4-byte big-endian length prefix + body.
- Response length validated: `> 0` and `≤ max_response_size` (default 100 MB).
- Reject oversized frames to guard against memory exhaustion.
