# Phase 3: Protocol Codec

**Status: COMPLETED**

## Goal

Implement the TCP wire format encoder/decoder as specified in `docs/protocol.md`. The protocol spec is the authoritative source — this plan describes the Rust implementation of that spec.

## Prerequisites

Before starting codec implementation:

1. **`GroupName` type** — added to `src/types.rs` (Phase 1 amendment, DD-004). The codec `Frame` enum uses `GroupName` for Subscribe/Unsubscribe frames.
2. **`max_message_size` config field** — added to `TcpConfig` in `src/config.rs` (Phase 2 amendment, DD-007). The `MessageCodec` reads this value to enforce frame size limits.
3. **`docs/protocol.md`** — complete and reviewed. (Done.)

## Dependencies

- `bytes` (already in Cargo.toml) — `Bytes`, `BytesMut`, `BufMut`, `Buf`
- `tokio-util` with `codec` feature (already in Cargo.toml) — `Encoder`, `Decoder` traits
- `proptest` (added to dev-dependencies) — property-based testing

## Files

- `src/types.rs` — `GroupName` newtype (amendment, completed in prior phase)
- `src/config.rs` — `max_message_size` in `TcpConfig` (amendment, completed in prior phase)
- `src/codec.rs` — `MessageCodec` struct, re-exports
- `src/codec/frame.rs` — `Frame` enum, `MessageType` enum, `ErrorCode` enum
- `src/codec/encoder.rs` — `impl Encoder<Frame> for MessageCodec`
- `src/codec/decoder.rs` — `impl Decoder for MessageCodec`
- `benches/codec_bench.rs` — encode/decode benchmarks

## Work Items

### 1. src/types.rs — Add GroupName (Phase 1 amendment)

- [x] `GroupName` newtype wrapping `String`
- [x] Derives: `Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize`
- [x] `Display` impl
- [x] `FromStr` impl — rejects empty strings (same pattern as `TopicName`)
- [x] Unit tests: display/fromstr roundtrip, empty rejection

### 2. src/config.rs — Add max_message_size (Phase 2 amendment)

- [x] Add `max_message_size: u32` field to `TcpConfig`
- [x] Default: `1_048_576` (1 MB)
- [x] `#[serde(default)]` on the field so existing configs don't break
- [x] Add `FAFB_MAX_MESSAGE_SIZE` env var override in `apply_env_overrides`
- [x] Add validation: `max_message_size` must be `>= 1024` (frames smaller than 1KB are unusable)
- [x] Add `MaxMessageSizeTooSmall` variant to `ConfigError`
- [x] Update `config/config.example.toml` with the new field
- [x] Unit tests: default value applied, env override, validation of too-small value

### 3. src/codec/frame.rs — Frame and MessageType definitions

- [x] `MessageType` enum with `repr(u8)` byte assignments matching protocol.md (13 variants)
- [x] `impl TryFrom<u8> for MessageType` — returns `Error::Codec` for unknown bytes
- [x] `impl From<MessageType> for u8`
- [x] `ErrorCode` enum with `repr(u16)` matching protocol.md error code table (9 variants)
- [x] `impl TryFrom<u16> for ErrorCode` — returns `Error::Codec` for unknown codes
- [x] `impl From<ErrorCode> for u16`
- [x] `ErrorCode::is_fatal(&self) -> bool` — AuthFailed, InvalidFrame, FrameTooLarge, UnsupportedVersion, NotAuthenticated
- [x] `ErrorCode::from_error(&Error) -> ErrorCode` — maps each `Error` variant to a wire code
- [x] `Frame` enum with all 13 variants, derives `Debug, Clone, PartialEq`
- [x] `Frame::message_type(&self) -> MessageType` helper method
- [x] Payloads use `bytes::Bytes` for zero-copy fan-out (DD-010)
- [x] `group: Option<GroupName>` — `None` maps to empty string on wire, `Some(name)` maps to non-empty string
- [x] 8 unit tests (MessageType roundtrip, unknown byte, ErrorCode roundtrip, unknown value, is_fatal, from_error, Frame::message_type, PartialEq with Bytes)

### 4. src/codec/encoder.rs — Encoding

- [x] `impl Encoder<Frame> for MessageCodec`
- [x] Associated `type Error = crate::error::Error`
- [x] Backfill length prefix pattern: write placeholder u32, serialize message type + payload, then overwrite length
- [x] String encoding helper: `write_string` — u16 length prefix + UTF-8 bytes, `Error::Codec` if > 65,535 bytes
- [x] Payload encoding helper: `write_payload` — u32 length prefix + bytes
- [x] Topic+group helper: `write_topic_group` — shared by Subscribe/SubscribeAck/Unsubscribe/UnsubscribeAck
- [x] Per-message encoding matches protocol.md field layouts exactly
- [x] Ping/Pong: empty payload (frame length = 1)
- [x] 14 unit tests (encode_ping, encode_pong, encode_connect, encode_connect_ack_success/failure, encode_subscribe_with_group/without_group, encode_publish/publish_empty_payload, encode_publish_ack, encode_deliver, encode_ack, encode_error, encode_string_too_long)

### 5. src/codec/decoder.rs — Decoding

- [x] `impl Decoder for MessageCodec`
- [x] Associated `type Item = Frame`, `type Error = crate::error::Error`
- [x] Peek frame length via indexing (does not consume bytes if incomplete)
- [x] Validate frame length > 0 and <= max_frame_size
- [x] `split_to` to isolate frame body (parsing errors can't corrupt subsequent frames)
- [x] Strict trailing-bytes rejection
- [x] String decoding helper: `read_string` — u16 length, UTF-8 validation
- [x] Payload decoding helper: `read_payload` — u32 length, zero-copy via `split_to` + `freeze`
- [x] Group decoding helper: `read_group` — empty string → None, non-empty → Some(GroupName)
- [x] Per-message decoding (inverse of encoder)
- [x] 13 decode tests, 4 partial-buffer tests, 7 malformed-input tests, 18 roundtrip tests

### 6. src/codec.rs — Module root

- [x] `MessageCodec` struct with `max_frame_size: u32` field, derives `Debug`
- [x] `MessageCodec::new(max_frame_size: u32)` constructor
- [x] Re-export `Frame`, `MessageType`, `ErrorCode` from `frame`

### 7. benches/codec_bench.rs — Benchmarks

- [x] `codec_encode` group: Ping, Connect, Subscribe, Publish (64B/1KB/64KB), Deliver (1KB)
- [x] `codec_decode` group: Pong, Publish (64B/1KB/64KB), Deliver (1KB), partial_buffer_none
- [x] `codec_roundtrip` group: Connect, Subscribe, Publish 1KB, Deliver 1KB
- [x] Baseline captured

## Cross-Subsystem Integration

The `Frame` enum is the interface contract between the codec and the rest of the system. Here's how each subsystem uses it:

| Subsystem | Produces | Consumes |
|---|---|---|
| TCP Server (Plan 8) | ConnectAck, SubscribeAck, UnsubscribeAck, PublishAck, Deliver, Error, Pong | Connect, Subscribe, Unsubscribe, Publish, Ack, Ping |
| Auth (Plan 4) | (via TCP Server) ConnectAck | (via TCP Server) Connect |
| Namespace Manager (Plan 7) | (via TCP Server) SubscribeAck, PublishAck, Deliver | (via TCP Server) Subscribe, Unsubscribe, Publish, Ack |
| HTTP Server (Plan 9) | Does not use Frame — HTTP has its own JSON API | — |

## Resolved Decisions

These were open questions in the original plan. See `docs/design_decisions.md` for full rationale.

| Question | Decision | Reference |
|---|---|---|
| Handshake or per-message auth? | Handshake via Connect/ConnectAck | DD-001 |
| Reply-to on Publish? | No, not in v1 | DD-002 |
| Subscribe carries group name? | Yes, optional GroupName field | DD-003 |
| Payload type Vec\<u8\> or Bytes? | `bytes::Bytes` for zero-copy fan-out | DD-010 |
| Ping/Pong carry data? | No, empty frames | DD-008 |
| Version in every frame or Connect only? | Connect only, reject on mismatch | DD-009 |
| Where does max_frame_size come from? | `TcpConfig.max_message_size` | DD-007 |
| Does PublishAck exist? | Yes, returns server-assigned MessageId | DD-005 |

## Test Plan

### Roundtrip Tests (encode → decode identity)

- [x] Every `Frame` variant: construct frame, encode to `BytesMut`, decode back, assert equality
- [x] Frames with empty strings (ConnectAck success with empty message)
- [x] Frames with empty payloads (Publish with 0-byte payload)
- [x] Frames with max-length strings (65,535 bytes)
- [x] Subscribe with `group: None` and `group: Some(...)`

### Partial Buffer Tests

- [x] Buffer with only 0-3 bytes → decoder returns `Ok(None)`
- [x] Buffer with length prefix but incomplete payload → decoder returns `Ok(None)`
- [x] Buffer with complete frame + partial next frame → decoder returns first frame, leaves remainder
- [x] Multiple complete frames in buffer → decoder returns them one at a time

### Malformed Input Tests

- [x] Unknown message type byte (e.g., `0xFF`) → `Err(Error::Codec)`
- [x] Frame length of 0 → `Err(Error::Codec)`
- [x] Frame length exceeding `max_message_size` → `Err(Error::Codec)`
- [x] String field claims N bytes but buffer has fewer → `Err(Error::Codec)`
- [x] Non-UTF-8 bytes in string field → `Err(Error::Codec)`
- [x] Payload length field claims more bytes than frame contains → `Err(Error::Codec)`
- [x] Extra trailing bytes after a valid message → `Err(Error::Codec)` (strict parsing)

### Property-Based Tests

- [x] Add `proptest` to `[dev-dependencies]`
- [x] For any valid `Frame`, `decode(encode(frame)) == frame` (roundtrip property)
- [x] For any random byte sequence, decoder either returns a valid `Frame` or `Err` — never panics

### Integration Readiness

- [x] `Framed<TcpStream, MessageCodec>` compiles (verified by static assertion in test)

## Implementation Notes

- **`MessageType::Error` variant** caused ambiguity with `TryFrom::Error` associated type in Rust 2024 edition. Resolved by using fully-qualified syntax `<Self as TryFrom<u8>>::Error` in the `TryFrom` impl.
- **Backfill length prefix** avoids extra allocation per encode — write placeholder u32, serialize, then overwrite.
- **`split_to` frame isolation** ensures parsing errors within one frame can't corrupt the buffer for subsequent frames.
- **`Bytes` for payloads** — `copy_to_bytes` on `BytesMut` is zero-copy; `Bytes::clone` is O(1) refcount for fan-out.
- **Codec does NOT validate business logic** — empty namespace/topic strings are encoded faithfully; auth layer validates.

## Completion Summary

- `cargo test` — 109 tests pass (69 new codec tests added to 40 existing)
- `cargo clippy` — clean
- `cargo fmt -- --check` — clean
- `cargo bench` — baselines captured (Ping encode ~5ns, 1KB publish roundtrip ~120ns)

## Done When

- [x] `docs/protocol.md` fully specifies every message type (done)
- [x] `GroupName` type added to `src/types.rs` with tests
- [x] `max_message_size` added to `TcpConfig` with default, env override, validation, and tests
- [x] `Frame` enum, `MessageType` enum, and `ErrorCode` enum defined
- [x] `Encoder` and `Decoder` impls pass all roundtrip tests for every frame variant
- [x] Partial buffer test: decoder returns `None` when bytes are incomplete
- [x] All malformed input tests pass (no panics, clean error returns)
- [x] Property-based roundtrip test passes for 1000+ generated frames
- [x] Benchmark baseline captured in `benches/codec_bench.rs`
- [x] `cargo clippy` clean, `cargo fmt` clean
- [x] `cargo test` — all existing tests still pass, all new tests pass
