# Phase 8: TCP Server

## Status: COMPLETED

## Goal
Accept TCP connections, run the protocol codec, and bridge to the namespace manager.

## Files
- `src/server.rs`
- `src/server/tcp.rs`
- `src/server/tcp/connection.rs`
- `src/server/tcp/session.rs`
- `src/config.rs` (extended with `idle_timeout_secs`, `max_connections`)
- `tests/common/mod.rs` (TestServer, TestClient infrastructure)
- `tests/tcp_integration.rs` (17 integration tests)

## Completed Work Items

### server/tcp/session.rs
- [x] Type-safe `Session` enum with `PreAuth` / `Authenticated` variants — no `Option::unwrap()` on auth state
- [x] `PreAuthState` handles Connect, Ping, Pong; rejects all other frames with NotAuthenticated
- [x] `AuthenticatedState` handles Subscribe, Unsubscribe, Publish, Ack, Error, Ping, Pong
- [x] Server-only frame rejection (ConnectAck, SubscribeAck, UnsubscribeAck, PublishAck, Deliver from client)
- [x] Version check (only version 1 supported)
- [x] Duplicate Connect rejection
- [x] Permission checks (Subscribe, Publish) via `PermissionSet::check()`
- [x] Message ID generation from shared `AtomicU64` counter
- [x] `build_deliver_frame()` — tracks `MessageId → TopicName` in `inflight_topics` for Ack resolution
- [x] Inflight tracking cap at 65,536 entries — prevents unbounded memory growth from slow-acking clients
- [x] Forwarder task spawning — one per subscription, bridges namespace manager's `mpsc::Receiver` into per-connection aggregation channel
- [x] Periodic dead forwarder reaping every 64 frames
- [x] `abort_forwarders()` for connection cleanup
- [x] 13 unit tests (server-only frame rejection, pre-auth gate, ping before auth, connect success/bad key/wrong version/already authenticated, subscribe permission denied, publish assigns message ID, ack unknown message ID, build_deliver_frame tracking, inflight cap)

### server/tcp/connection.rs
- [x] `ConnectionConfig` struct bundles per-connection parameters
- [x] `run_connection()` — `tokio::select!` loop multiplexing:
  - Inbound frames from `Framed<TcpStream, MessageCodec>`
  - Outbound deliveries from aggregation channel
  - Idle timeout (configurable, `Option`-based — no timer when disabled)
  - Shutdown signal (sends goodbye Error frame)
- [x] Codec error handling — sends Error frame, closes on fatal codes
- [x] Idle timer reset on any inbound frame or outbound delivery
- [x] Cleanup on exit: abort forwarders + `manager.disconnect()`

### server/tcp.rs
- [x] `TcpServer<M: NamespaceManager>` struct with config, manager, auth, counters, shutdown
- [x] `run()` — binds to configured address/port
- [x] `run_with_listener()` — accepts pre-bound `TcpListener` (eliminates test race)
- [x] Accept loop with `tokio::select!` on accept + shutdown signal
- [x] Max connections via `Arc<Semaphore>` with `try_acquire_owned`
- [x] Over-limit connections receive `Frame::Error { "server at capacity" }` before close
- [x] Shutdown signal propagated to each spawned connection via `watch::Receiver`
- [x] 1 unit test (server_is_send)

### config.rs (extended)
- [x] `TcpConfig.idle_timeout_secs` — default 300s, 0 = disabled
- [x] `TcpConfig.max_connections` — default 0 (unlimited)
- [x] `ConfigError::IdleTimeoutTooSmall` — rejects non-zero values < 10s
- [x] Environment variable overrides: `FAFB_TCP_IDLE_TIMEOUT`, `FAFB_TCP_MAX_CONNECTIONS`

### Test infrastructure (tests/common/mod.rs)
- [x] `TestServer` struct with address and shutdown handle
- [x] `TestClient` with `connect()`, `send()`, `recv()`, `recv_timeout()`, `connect_and_auth()`
- [x] `start_test_tcp_server()` — full-permissions server on random port
- [x] `start_test_tcp_server_with_perms()` — custom permissions
- [x] `start_test_tcp_server_with_config()` — custom permissions + config overrides
- [x] Uses `run_with_listener()` with pre-bound listener — no sleep, no port race

### Integration tests (tests/tcp_integration.rs) — 17 tests
- [x] `connect_success` — Connect → ConnectAck { success: true }
- [x] `connect_bad_key` — wrong key → ConnectAck { success: false }, connection closes
- [x] `connect_wrong_version` — version=2 → Error { UnsupportedVersion }, closes
- [x] `frame_before_auth` — Subscribe before Connect → Error { NotAuthenticated }, closes
- [x] `client_sends_server_frame` — ConnectAck from client → Error { InvalidFrame }, closes
- [x] `subscribe_publish_deliver` — Client A subscribes, Client B publishes → Client A receives Deliver
- [x] `publish_ack_flow` — Publish → PublishAck with MessageId
- [x] `subscribe_ack_then_deliver_then_ack` — full lifecycle: subscribe → publish → deliver → ack
- [x] `unsubscribe_stops_delivery` — Subscribe, unsubscribe, publish → no Deliver
- [x] `multiple_subscriptions` — Subscribe to 2 topics → receives Delivers for both
- [x] `concurrent_connections` — 20 clients subscribe, 100 messages published → each receives all 100
- [x] `disconnect_cleanup` — Client drops → subsequent publish works, other clients unaffected
- [x] `publish_only_client_cannot_subscribe` — publish+query perms → Subscribe returns PermissionDenied, connection stays alive
- [x] `subscribe_only_client_cannot_publish` — subscribe+query perms → Publish returns PermissionDenied, connection stays alive
- [x] `idle_timeout_closes_connection` — 1s timeout, idle client receives Error frame + close
- [x] `idle_timeout_reset_by_activity` — pings at 1s intervals keep 2s-timeout connection alive for 4s+
- [x] `max_connections_rejects_with_error_frame` — limit=2, third client receives "server at capacity" Error

## Key Design Decisions

| Decision | Rationale |
|----------|-----------|
| Type-safe Session enum (PreAuth/Authenticated) | Eliminates unwrap() on auth state; impossible to call authenticated operations pre-auth |
| One forwarder task per subscription | Bridges namespace manager's per-subscription `mpsc::Receiver` into single aggregation channel; simpler than `FuturesUnordered` at v1 scale |
| Inflight cap (65K) | Prevents unbounded HashMap growth from slow-acking clients; namespace manager's inflight timeout reclaims untracked entries |
| `ConnectionConfig` struct | Bundles 6 per-connection parameters to avoid too-many-arguments lint |
| `run_with_listener()` | Eliminates bind-drop-rebind port race in tests; production uses `run()` which binds internally |
| Error frame before capacity rejection | Gives clients actionable information instead of silent TCP RST |
| Shutdown propagation via `watch::Receiver` | Each connection sends goodbye Error frame on server shutdown instead of being silently abandoned |
| `Option`-based idle timer | No timer allocated when idle timeout is disabled; uses `std::future::pending()` instead of `Duration::MAX` hack |
| Periodic forwarder reaping (every 64 frames) | Catches dead forwarders without scanning on every frame; covers cases where client never re-subscribes |

## Resolved Open Questions

| Question | Resolution |
|----------|-----------|
| Connection limits? | `TcpConfig.max_connections` with `Semaphore`, default unlimited |
| Keepalive? | Ping/Pong supported (always allowed, even pre-auth). Configurable idle timeout (default 300s) closes inactive connections |
| TLS? | Not implemented in v1. Design does not preclude it — `run_with_listener` can accept a TLS-wrapped listener |

## Verification

```
cargo test                    # 245 tests pass (216 unit + 29 integration)
cargo clippy                  # 0 warnings
cargo fmt -- --check          # 0 diffs
```
