# Phase 21: At-Least-Once Delivery Guarantees

## Goal

Formalize, implement, and document the delivery guarantee semantics of fafb-q so that developers can make informed decisions about using it in production systems. The system already has many of the building blocks (ack tracking, inflight timeouts, durable storage, consumer group offsets), but the guarantees are not documented, edge cases are not hardened, and there is no clear contract that developers can rely on. This plan closes those gaps and positions fafb-q's delivery semantics as a first-class feature rather than an emergent behavior.

## Guiding Principles

- **At-least-once is the default and only guarantee.** Exactly-once delivery is a distributed systems impossibility without application-level idempotency. fafb-q provides at-least-once: every published message is delivered to each subscriber/consumer group at least once, provided the subscriber eventually connects and processes it. Duplicates are possible and documented.
- **Document when duplicates happen.** Developers need to know exactly when duplicates occur so they can build idempotency where it matters. The three duplicate scenarios are: (1) inflight timeout before client ack, (2) client reconnect with offset rewind, (3) consumer group rebalance during processing.
- **Durability before acknowledgment.** A `PublishAck` means the message is durable — written to the storage layer (segment file) and will survive a server restart. This is the core contract. If fsync is set to `os_managed`, the durability window is documented (OS crash may lose recent messages).
- **Offset continuity is guaranteed.** Offsets are monotonically increasing, gap-free within a topic. No phantom offsets, no offset reuse after compaction. Consumers can rely on offset arithmetic for progress tracking.
- **Consumer group semantics are Kafka-like.** Within a consumer group, each message is delivered to exactly one consumer. Redelivery on failure goes to any available consumer in the group (not necessarily the original). Group offset advances only after ack.
- **Fan-out subscribers get independent delivery.** Each non-group subscriber gets every message independently. One subscriber's failure does not affect others.

## Current State & Gaps

| Guarantee | Current Implementation | Gap |
|---|---|---|
| Publish flow | In-memory buffer first → `storage.append()` (page cache write) → PublishAck sent → background flush does fsync later | **PublishAck is sent after page cache write but BEFORE fsync.** The code comment in `engine.rs:443-447` explicitly notes that consumers may read a message that won't survive restart if the OS crashes before fsync. This is NOT the same as "sent after segment write" — it's "sent after page cache write." |
| Inflight tracking | `InflightEntry { offset, message_id, delivered_to, delivered_at: Instant, delivery_count: u32 }` — per-consumer-group and per-individual-subscriber maps | `delivery_count` already exists. Max redelivery already exists (`max_redelivery_attempts`, default 5). Missing: per-entry backoff timing, configurable timeout per namespace. |
| Inflight timeout | Global `DEFAULT_INFLIGHT_TIMEOUT = 300s`. `AtomicU64` for lock-free reads. Adjustable via `set_inflight_timeout()` (used during drain). | Fixed timeout, no exponential backoff or per-namespace config. `evict_expired_inflight()` removes ALL expired entries in one sweep. |
| Offset durability | Offsets are in-memory, rebuilt from storage on restart via `engine.init()` replay | Consumer group offsets are lost on restart — consumers must track their own via `SubscribeAt` |
| Redelivery | In group delivery task: `evict_expired_inflight(timeout)` removes expired entries, then messages at those offsets are re-attempted on next delivery loop iteration | No exponential backoff, no jitter. All expired entries are evicted at once. |
| Ack semantics | `manager.ack()` removes from inflight map, advances group offset. Ack is fire-and-forget (no response frame). | Single-message ack only — no batch ack |
| Delivery ordering | Per-topic, per-consumer ordering within a single connection | Ordering guarantee not documented, broken by redelivery (redelivered message arrives out of order) |
| Exactly-once | Not supported | Not documented — developers may assume it works |
| Duplicate ack handling | `manager.ack()` returns error if message_id not found in inflight (already acked or expired). Session logs warning. | No metric for duplicate acks |

## Design

### Delivery Guarantee Specification

#### For Publishers

1. **Durability contract:** When a client receives `PublishAck`, the message has been:
   - **Written to the in-memory buffer** (available for immediate delivery to subscribers)
   - **Appended to the segment file** via `storage.append()` (written to the OS page cache)
   - **NOT necessarily fsynced to disk** — fsync happens in the background flush task (every `batch_flush_interval_ms`, default 100ms) or on early flush when `batch_flush_size_bytes` (default 4MB) is exceeded.

   **Implications by fsync mode:**
   - `fsync = "every_batch"` (default): Data is in the page cache at PublishAck time. fsync follows within ~100ms. A **fafb-q process crash** does NOT lose data (the OS page cache survives process death). An **OS crash or power loss** within the flush interval may lose recent messages.
   - `fsync = "every_write"`: Same as `every_batch` but with more frequent fsync calls. Still not synchronous with PublishAck — the flush is background.
   - `fsync = "os_managed"`: Equivalent to `every_batch` from the client's perspective, but fsync is never explicitly called. Relies entirely on the OS writeback policy. Highest throughput, lowest durability guarantee.

   > **Key nuance:** PublishAck does NOT mean "fsynced to disk." It means "written to page cache and will survive a fafb-q process crash." This is the same guarantee as Kafka with `acks=1` (single broker, no replication). Document this clearly — developers coming from databases expect "ack = durable" to mean "on disk."

2. **Offset assignment:** The `PublishAck` includes the assigned `offset`. Offsets are sequential, gap-free, and unique within a `(namespace, topic)`. Once assigned, an offset is never reused.

3. **Ordering:** Messages published on a single TCP connection to a single topic are stored in send order. Messages from different connections or to different topics have no ordering guarantee relative to each other.

#### For Push Subscribers (Subscribe/Deliver/Ack)

1. **At-least-once delivery:** Every message published to a subscribed topic will be delivered to the subscriber at least once, provided the subscriber maintains or re-establishes a connection before the message is removed by retention.

2. **Duplicate scenarios:**
   - *Inflight timeout:* Client receives `Deliver`, starts processing, but does not `Ack` before the inflight timeout. Server redelivers. If client then acks both, the second ack is a no-op.
   - *Reconnect overlap:* Client disconnects mid-processing, reconnects with `SubscribeAt` at last-acked offset + 1. If the server already sent the message but ack was lost, the message is redelivered.
   - *Consumer group rebalance:* In a consumer group, if a consumer disconnects, its inflight messages are redelivered to other group members.

3. **Ordering:** Messages are delivered in offset order within a topic on a given connection. However, redelivered messages may arrive out of order relative to newer messages. Specifically: if message at offset 5 is redelivered while offset 10 has already been delivered, offset 5 arrives after offset 10. Applications that require strict ordering must process messages sequentially and ack before requesting the next.

4. **Fan-out independence:** Each non-group subscriber has its own delivery state. One subscriber being slow or disconnected does not affect delivery to others.

#### For Pull Consumers (Fetch)

1. **Stateless reads:** Fetch is a point-in-time read. The server maintains no state for fetch consumers. The client manages its own offset.

2. **Consistency:** A fetch at offset N returns the message at offset N if it exists and has not been removed by retention. The response includes `head_offset` (latest) and `next_offset` for pagination.

3. **No delivery guarantee:** Fetch provides no delivery tracking. If the client crashes between fetch and processing, it must re-fetch. This is by design — fetch consumers manage their own reliability.

### Implementation Hardening

#### 1. Redelivery with Backoff

Currently, redelivery uses a single global timeout (`DEFAULT_INFLIGHT_TIMEOUT = 300s`). All expired entries are evicted at once by `evict_expired_inflight()`. Add per-entry exponential backoff:

```rust
pub struct InflightEntry {
    // existing fields (already present):
    pub offset: TopicOffset,
    pub message_id: MessageId,
    pub delivered_to: ClientId,
    pub delivered_at: Instant,
    pub delivery_count: u32,
    // NEW:
    pub next_redeliver_at: Instant,
}

impl InflightEntry {
    /// Calculate next redelivery time with exponential backoff + jitter.
    /// Base: inflight_timeout. Backoff: min(base * 2^(attempts-1), max_backoff).
    /// Jitter: ±25% random.
    fn next_redelivery_delay(&self, base_timeout: Duration, max_backoff: Duration) -> Duration {
        let backoff = base_timeout * 2u32.pow((self.delivery_count - 1).min(10));
        let capped = backoff.min(max_backoff);
        let jitter = rand::thread_rng().gen_range(0.75..1.25);
        capped.mul_f64(jitter)
    }
}
```

**Change to `evict_expired_inflight()`:** Instead of checking `delivered_at.elapsed() > timeout`, check `Instant::now() >= next_redeliver_at`. This makes each entry independently timed.

Configuration (in `[queue]` section, matching `QueueConfig`):
```toml
[queue]
inflight_timeout_secs = 300        # Base timeout (new config — currently hardcoded as DEFAULT_INFLIGHT_TIMEOUT)
max_redelivery_backoff_secs = 3600 # Max backoff cap (new, default 1 hour)
```

> **Note:** The existing `ManagerConfig` stores the inflight timeout as `DEFAULT_INFLIGHT_TIMEOUT` (hardcoded 300s) in an `AtomicU64`. This should become configurable via `QueueConfig.inflight_timeout_secs`. The `AtomicU64` stays for drain-time adjustment via `set_inflight_timeout()`.

#### 2. Batch Ack

Allow clients to ack multiple messages in one frame to reduce round-trips:

```
Frame::BatchAck (0x18) — client → server
  [u32 count]
  for each:
    [u64 message_id]

Frame::BatchAckResponse (0x19) — server → client
  [u32 acked_count]
  [u32 unknown_count]    — message_ids not found in inflight (already acked or expired)
```

Requires `CAP_BATCH_ACK` capability (`1 << 5`, bit 5). Note: bit 4 is already used by `CAP_WILDCARD` (Plan 20).

Semantics: each message_id is acked independently. Unknown IDs are counted but not treated as errors (idempotent ack). The response tells the client how many were actually outstanding.

#### 3. Ack Offset Reporting

Add an optional field to `Deliver` that tells the client the consumer group's committed offset:

```
Frame::Deliver (updated, when CAP_BATCH_ACK negotiated):
  ... existing fields ...
  [u64 committed_offset]    — the group's last contiguously-acked offset
```

This lets clients track their progress without server round-trips and detect gaps.

#### 4. Publish Confirm Modes

Add a "fire-and-forget" publish mode for high-throughput scenarios where the publisher does not need durability confirmation:

```toml
# Per-namespace config
[namespace.publish]
require_ack = true    # default: publisher gets PublishAck
                      # false: server does not send PublishAck (higher throughput)
```

When `require_ack = false`, the server still stores the message durably but skips the `PublishAck` response. The client is responsible for flow control. This is opt-in and documented as reducing the durability contract to "best effort from the client's perspective."

### Metrics (New)

- `fafb_redelivery_total{namespace, topic}` — messages redelivered (inflight timeout expired)
- `fafb_redelivery_delay_seconds` (histogram) — actual delay before redelivery (shows backoff distribution)
- `fafb_ack_latency_seconds{namespace, topic}` (histogram) — time between Deliver and Ack (shows consumer processing time)
- `fafb_batch_ack_size` (histogram) — messages per BatchAck frame
- `fafb_duplicate_ack_total{namespace, topic}` — acks for already-acked messages (duplicate indicator)
- `fafb_consumer_lag{namespace, topic, group}` (gauge) — head_offset - committed_offset per group

### Documentation Deliverable

Create `docs/delivery_guarantees.md`:

1. **Guarantee summary table** — at-a-glance comparison with Kafka, NATS, RabbitMQ
2. **Publisher guarantees** — durability contract, fsync modes, ordering
3. **Subscriber guarantees** — at-least-once, duplicate scenarios, ordering caveats
4. **Consumer group semantics** — offset tracking, rebalance behavior, redelivery
5. **Fetch semantics** — stateless, no guarantee, client-managed
6. **Building idempotent consumers** — patterns and recommendations
7. **Failure mode matrix** — what happens on: server crash, client crash, network partition, disk full, OOM

## Implementation Phases

### Phase 21.1 — Delivery Guarantees Documentation

**Goal:** Write the definitive document on fafb-q delivery semantics. No code changes — this phase ensures we understand and accurately document what exists before hardening it.

- [x] Create `docs/delivery_guarantees.md` with the following sections:
  - [x] **Guarantee summary table** — at-a-glance comparison with Kafka (`acks=1`), NATS, RabbitMQ
  - [x] **Publisher guarantees** — durability contract per fsync mode, offset assignment, ordering rules
  - [x] **Subscriber guarantees (push)** — at-least-once semantics, the three duplicate scenarios (inflight timeout, reconnect overlap, group rebalance), ordering caveats with redelivery
  - [x] **Consumer group semantics** — per-group offset tracking, round-robin delivery, redelivery to any group member, offset only advances after ack
  - [x] **Fetch semantics (pull)** — stateless, no delivery tracking, client manages own reliability
  - [x] **Building idempotent consumers** — patterns: dedup by message_id, idempotent writes, exactly-once via external store
  - [x] **Failure mode matrix** — what happens on: fafb-q process crash, OS crash/power loss, client crash mid-processing, network partition, disk full, OOM
- [x] Review and document existing fsync behavior accurately
  - [x] Trace the publish path: in-memory buffer → `storage.append()` (page cache) → `PublishAck` → background flush (fsync)
  - [x] Document the key nuance: PublishAck = "in page cache" not "fsynced to disk"
  - [x] Document each fsync mode's implications (`every_batch`, `every_write`, `os_managed`)
- [x] Walk through each failure mode and verify documented behavior matches actual code behavior
- [x] Peer review the document for accuracy

**Verification:** `docs/delivery_guarantees.md` exists and covers all sections. No code changes in this phase.

---

### Phase 21.2 — Redelivery with Exponential Backoff

**Goal:** Replace the fixed inflight timeout with per-entry exponential backoff. Make the base timeout configurable instead of hardcoded. This gives transient failures more time to recover before DLQ routing.

- [x] Add `next_redeliver_at: Instant` field to `InflightEntry` in `src/namespace/state.rs`
  - [x] Set `next_redeliver_at = now + base_timeout` on first delivery
  - [x] Update `next_redeliver_at` on each redelivery using exponential backoff formula
- [x] Implement `redelivery_delay()` on `InflightEntry`
  - [x] Formula: `min(base_timeout * 2^(delivery_count - 1), max_backoff)`
  - [x] Add ±25% random jitter to prevent thundering herd
  - [x] Cap exponent at 10 to prevent overflow
- [x] Change `evict_expired_inflight()` to check `Instant::now() >= next_redeliver_at` instead of `delivered_at.elapsed() > timeout`
  - [x] Each entry is independently timed based on its own backoff schedule
- [x] Make inflight timeout configurable
  - [x] Add `inflight_timeout_secs: u64` to `QueueConfig` (default 300, replacing `DEFAULT_INFLIGHT_TIMEOUT`)
  - [x] Wire through to namespace manager — keep `AtomicU64` for drain-time adjustment via `set_inflight_timeout()`
- [x] Add `max_redelivery_backoff_secs: u64` to `QueueConfig` (default 3600 = 1 hour)
- [x] Add `fafb_redelivery_total{namespace, topic}` counter metric — incremented on each redelivery
- [x] Add `fafb_redelivery_delay_seconds` histogram metric — actual delay before each redelivery (shows backoff distribution)
- [x] Unit tests
  - [x] First delivery: `next_redeliver_at` = now + base_timeout
  - [x] Second delivery: delay ≈ 2× base (within jitter bounds)
  - [x] Backoff caps at `max_redelivery_backoff_secs`
  - [x] Jitter stays within ±25% bounds
  - [x] `evict_expired_inflight()` only evicts entries past their individual `next_redeliver_at`
- [x] Integration test: publish → deliver → no ack → verify redelivery delay increases on each attempt (covered by existing `individual_inflight_sweep_escalates_count_below_threshold` test which validates the backoff path)
- [x] Update `docs/delivery_guarantees.md` with backoff behavior and timing examples

**Additional hardening (completed alongside 21.2):**
- [x] Consumer group delivery task inflight sweep — added periodic sweep mirroring individual task (was missing: group inflight entries only checked on next delivery, never timed out independently)
- [x] Drain-time deadline reset — `set_inflight_timeout()` now walks all namespaces/topics and caps `next_redeliver_at` to `now + accelerated_timeout` so long-backoff entries don't outlive the drain window
- [x] Config hot-reload wiring — `refresh_config!` macro reads timeout from `SharedConfig` (hot-reloadable), sweep reads from atomics (drain-responsive). Clean separation: config reload → insertion path, drain override → sweep path
- [x] Consolidated timeout loads — removed per-call-site atomic loads in favor of cached vars refreshed once per consumed message via `refresh_config!`
- [x] `TopicDeliveryState::reset_inflight_deadlines(timeout)` added to `state.rs`

**Verification:** Redelivery uses per-entry exponential backoff. Base timeout is configurable. Existing drain-time timeout adjustment still works. Consumer group inflight entries now properly time out and route to DLQ. Config hot-reload updates timeout values. `cargo test` passes (685 unit + all integration).

---

### Phase 21.3 — Batch Ack

**Goal:** Allow clients to acknowledge multiple messages in a single frame, reducing round-trips for high-throughput consumers.

- [x] Add frame types to codec (`src/codec/frame.rs`)
  - [x] `MessageType::BatchAck = 0x18`
  - [x] `MessageType::BatchAckResponse = 0x19`
  - [x] `Frame::BatchAck { message_ids: Vec<MessageId> }` — client → server
  - [x] `Frame::BatchAckResponse { acked_count: u32, unknown_count: u32 }` — server → client
- [x] Implement encode/decode for BatchAck and BatchAckResponse
  - [x] BatchAck: `[u32 count]` + `[u64 message_id]` per entry
  - [x] BatchAckResponse: `[u32 acked_count][u32 unknown_count]`
  - [x] Limit: max 10,000 message_ids per BatchAck frame
- [x] Add `CAP_BATCH_ACK` capability constant (`1 << 5`, bit 5) — bit 4 is taken by `CAP_WILDCARD`
  - [x] Add to `SERVER_CAPABILITIES` bitmask
  - [x] Reject BatchAck without capability negotiated
- [x] Handle BatchAck in TCP session
  - [x] Iterate message_ids, ack each via `manager.ack()`
  - [x] Count successes (acked) and unknowns (not in inflight — already acked or expired)
  - [x] Remove acked message_ids from `session.inflight_topics`
  - [x] Return BatchAckResponse with counts
- [x] Allow BatchAck during drain (same as single Ack — falls through Draining handler naturally)
- [x] Add `fafb_batch_ack_size` histogram metric — messages per BatchAck frame
- [x] Unit tests for codec: BatchAck/BatchAckResponse encode/decode round-trip (3 encoder + 4 decoder tests)
- [x] Unit tests for session: BatchAck with mix of valid and unknown message_ids (4 tests)
- [x] Integration test: publish 5 → deliver all → BatchAck all 5 → verify acked_count=5
- [x] Integration test: BatchAck with some already-acked IDs → verify `unknown_count` in response

**Additional hardening (completed alongside 21.3):**
- [x] Optimized `batch_ack()` method on `NamespaceManager` trait and `DefaultNamespaceManager` — groups message_ids by topic, acquires lock once per topic, removes all matching entries, evicts once, computes minimum once. O(M topics) lock acquisitions instead of O(N messages). `BatchAckResult { acked, unknown }` return type.
- [x] Session handler collects `(MessageId, TopicName)` pairs from `inflight_topics` and delegates to `manager.batch_ack()` in a single call
- [x] Boundary decode tests: count=10,000 accepted, count=10,001 rejected with descriptive error

**Verification:** BatchAck processes multiple acks in one frame with O(topics) lock overhead. Unknown IDs are counted but not errors (idempotent). `cargo test` passes (698 unit + all integration).

---

### Phase 21.4 — Delivery Observability Metrics

**Goal:** Add consumer lag, ack latency, and duplicate ack tracking metrics for production observability.

- [x] Add `fafb_consumer_lag{namespace, topic, group}` gauge
  - [x] Computed as `head_offset - next_offset + inflight.len()` per consumer group per topic (O(1))
  - [x] Update on publish (head advances, all groups updated) and on ack (inflight clears)
  - [x] Expose in Prometheus metrics endpoint
  - [x] Document cardinality concern: `namespaces × topics × groups` — documented in `docs/delivery_guarantees.md`
- [x] Add `fafb_ack_latency_seconds{namespace, topic}` histogram
  - [x] `delivered_at` is already tracked as `Instant` on `InflightEntry`
  - [x] On ack (single and batch), compute `delivered_at.elapsed()` and observe the duration
  - [x] Shows consumer processing time distribution
- [x] Add `fafb_duplicate_ack_total{namespace, topic}` counter
  - [x] Incremented when `manager.ack()` / `manager.batch_ack()` finds no inflight entry for a message_id (already acked or expired)
  - [x] Session layer also increments (namespace-only label) when `inflight_topics.remove()` returns None for single Ack frames
  - [x] Useful for detecting client-side double-ack bugs or aggressive retry
- [x] Unit test: consumer lag updates correctly on publish and ack (`consumer_lag_updates_on_publish_and_ack`)
- [x] Unit test: ack latency histogram records correct duration (`ack_latency_histogram_records_positive_duration`)
- [x] Unit test: batch ack records latency and duplicate metrics (`batch_ack_records_latency_and_duplicate_metrics`)
- [x] Unit test: duplicate ack increments counter (`duplicate_ack_increments_counter`)
- [x] Integration test: publish → deliver → ack → verify consumer lag converges (`consumer_lag_decreases_after_ack`)
- [x] Integration test: double-ack same message_id → verify duplicate ack path exercised (`duplicate_ack_is_idempotent`)

**Additional hardening (completed alongside 21.4):**
- [x] `TopicDeliveryState.cached_head_offset` — publish path caches the head offset; ack/batch_ack read it instead of calling `engine.head_offset()`, eliminating an async lock acquisition per ack
- [x] Publish-path lag gauge iteration moved outside topic lock — collects `(GroupName, lag)` tuples under lock, sets gauges after lock release to reduce contention
- [x] Session-layer `fafb_duplicate_ack_total` — single Ack frames that fail `inflight_topics.remove()` now increment the counter at the session layer (where the duplicate is actually detected for individual subscribers)
- [x] Lag gauge cleanup on group removal — `fafb_consumer_lag` zeroed in five group-removal paths: `cleanup_expired_groups()` (now takes ns/topic params), `unsubscribe()` (group empty), `disconnect()` exact-match and wildcard paths, delivery task slow-consumer ejection

**Verification:** All three metrics (lag, ack latency, duplicate acks) are correctly recorded and exposed via Prometheus. Lag gauge uses cached head_offset (zero async overhead on ack). Stale lag label combos cleaned up on group removal. `cargo test` passes (702 unit + all integration), zero clippy warnings.

## Testing Strategy

- **Unit tests:** Backoff calculation, batch ack processing, consumer lag computation
- **Integration tests:** Redelivery after timeout with backoff, batch ack flow, reconnect + redelivery (duplicate scenario), consumer group rebalance redelivery
- **Chaos tests:** Kill consumer mid-processing → verify redelivery to another group member, kill server → restart → verify messages not lost
- **Documentation review:** Walk through each failure mode in the matrix and verify documented behavior matches actual behavior

## Risks and Mitigations

| Risk | Mitigation |
|---|---|
| Documenting PublishAck ≠ durable-on-disk may reduce confidence | Be honest. Kafka has the same guarantee with `acks=1`. Most applications don't need synchronous fsync — page cache survives process crashes. Offer `fsync = "every_write"` for applications that need stronger guarantees, and document the throughput trade-off. |
| Exponential backoff delays poison message DLQ routing | Backoff and DLQ are complementary: backoff gives consumers more time on transient failures, DLQ catches permanent failures. After `max_redelivery_attempts` (default 5), the message goes to DLQ regardless of backoff timing. The total time before DLQ = sum of backoff delays, which with default settings is ~300s + 600s + 1200s + 2400s + 3600s ≈ 2.3 hours. Document this and make it configurable. |
| Batch Ack introduces a new capability gate — adoption friction | Batch Ack is optional. Existing single-Ack flow works unchanged. SDK can auto-batch transparently. |
| Fire-and-forget publish (`require_ack = false`) may cause silent data loss | Document clearly as "best effort from client's perspective." The server still stores durably — the client just doesn't get confirmation. Primarily useful for metrics/telemetry streams where occasional loss is acceptable. |
| Consumer lag metric cardinality explosion | Labels are `{namespace, topic, group}`. With 100 namespaces × 100 topics × 10 groups = 100K series. Consider making this opt-in or aggregating by namespace only. |

## Out of Scope

- **Exactly-once delivery** — Requires distributed transactions or application-level idempotency. Not a server feature.
- **Synchronous fsync on PublishAck** — Would require changing the publish flow to fsync before returning. Possible as a future `fsync = "synchronous"` mode but would dramatically reduce throughput.
- **Consumer group offset persistence** — Storing committed offsets on the server so they survive restart. Currently, consumers must track their own offsets and use `SubscribeAt` after reconnect. Server-side offset storage is a separate feature (similar to Kafka's `__consumer_offsets` topic).
- **Transactional publish** — Publish multiple messages across topics atomically. Requires a transaction log — separate feature.
