feat: transparent retry deduplication

docs/clients/c2_producer.md

@@ -117,15 +117,24 @@ this is; the codec tag says *how to read* it.
 
 ---
 
-## Idempotency hook (future)
+## Delivery contract
 
-Today produce is **at-least-once**: a produce that times out after the leader
-committed but before the client saw the ack will be retried and stored twice. Exactly-
-once needs a producer session id + per-record sequence numbers carried on the produce,
-deduplicated at the segment leader — the server-side half is its own backlog item. C2
-should leave a clean seam for it (a place to stamp session/sequence on each produce)
-even before the server enforces it, so enabling exactly-once later doesn't reshape the
-producer API.
+The SDK producer uses transparent retry deduplication. If an acknowledgment is lost after
+commit, retrying the same immutable batch returns its original position without storing it
+again. Producer identity, incarnation, sequence, and payload digest participate in the
+broker decision and are durable with the append.
+
+The protocol is specified in
+[D10: Transparent Retry Deduplication](../data-plane/d10_transparent_retry_deduplication.md). It assigns a
+sequence to each immutable per-range broker batch and orders each producer-range stream,
+uses metadata-backed incarnations for fencing, and retains range-scoped deduplication
+frontiers across rolls, failover, and lineage changes. A bounded recent-result window
+returns exact positions for normal retries without retaining one position forever per
+request. This is intentionally described as idempotent production rather than end-to-
+end exactly-once processing.
+
+Until D10 is implemented end to end, callers must not interpret construction with a
+stable producer UUID as a delivery guarantee.
 
 ---
 
@@ -142,8 +151,9 @@ producer API.
 5. **Compression** — optional client-side codec over the (batched) records, with a
    cleartext codec tag prefixing the compressed block so the consumer can decompress; the
    broker stays opaque and never decompresses.
-6. **Idempotency seam** — a no-op-today hook to stamp producer session + sequence,
-   ready for the server-side dedup work.
+6. **Idempotency seam** — a no-op-today identity allocation point. D10 will replace the
+   per-record counter with immutable batch identity and carry it through the broker
+   protocol.
 7. **Tests** — against the simulated cluster: produce to the right leader from a warm
    cache (one hop), correction after a roll/failover (`NotWriteLeader` follow),
    correction from a cold/stale cache (`ShardNotLocal` follow), `TopicNotFound`
@@ -156,4 +166,6 @@ producer API.
 - `d6_produce_consume_api.md` — the server's produce routing and redirect contract.
 - `d1_storage_engine.md` — the broker-opaque entry payload and end-to-end compression
   this stamps a codec into.
+- `../data-plane/d10_transparent_retry_deduplication.md` — session fencing, ordered producer-range streams,
+  durable range ledgers, topology handoff, and bounded retry guarantees.
 - `client_roadmap.md` — the idempotency / batching / compression backlog context.

docs/clients/client_roadmap.md

@@ -121,8 +121,11 @@ C2 (Producer)  C3 (Consumer)
 ---
 
 ## Backlog (beyond a working client)
-- **Idempotent / exactly-once produce** — producer session + sequence numbers, dedup
-  at the segment leader (the server-side half is its own backlog item).
+- **Idempotent produce** — the complete client/server protocol is designed in
+  [data-plane D10](../data-plane/d10_idempotent_production.md): immutable batch
+  identities, independently ordered producer-range streams, metadata-backed fencing,
+  recovery journal, and bounded retry lifetime. Cross-range transactions and end-to-end
+  exactly-once processing remain separate future work.
 - **Client-side batching & compression** — amortize round-trips and bytes.
 - **Adaptive replica selection** — pick the fastest/nearest replica per range from
   observed latency, not just `replica_set` order.

docs/data-plane/d8_consumer_offset_management.md

@@ -168,12 +168,56 @@ offset commit fsync
 
 Snapshot creation is driven by reclamation pressure. Many commits therefore collapse into a
 single asynchronous snapshot. WAL deletion uses the minimum safe watermark across segment
-checkpoints and consumer-offset checkpoints; neither state family can delete recovery data
-still needed by the other.
+checkpoints and the auxiliary-state checkpoint, which contains only consumer offsets today;
+neither state family can delete recovery data still needed by the other.
 
 On restart, recovery loads the latest snapshot and replays later offset records from the
 shared WAL. Recovery then writes the consolidated snapshot before removing the replayed WAL.
 
+### Consolidation with producer frontier state
+
+The implemented snapshot currently contains consumer-offset state only. D10 adds another
+range-scoped, end-state ledger for producer retry frontiers. It must extend this checkpoint
+into one **range-auxiliary-state snapshot**, rather than introducing a second independently
+fsynced snapshot:
+
+```text
+shared WAL
+  ├── consumer offset records
+  └── produced entries carrying producer sequence identity
+              │ reclamation reaches uncovered state
+              ▼
+one auxiliary-state snapshot at one LSN boundary
+  ├── current consumer offsets + placement readiness
+  └── current producer frontiers + bounded recent results
+              │ durable rename completes
+              ▼
+advance one auxiliary checkpoint watermark
+```
+
+This follows the metadata log-to-snapshot model: replayable log records are authoritative
+until one complete current-state image safely replaces their covered prefix. The snapshot
+is asynchronous and reclamation-driven, so its single fsync is not part of offset-commit or
+produce latency. Segment payload checkpointing remains separate because segment files retain
+application data, whereas both auxiliary ledgers retain only their latest state.
+
+### Migration from the consumer-only checkpoint
+
+The D10 implementation should replace, rather than wrap, the current consumer-only snapshot
+pipeline:
+
+1. Replace the consumer-offset-only snapshot image with the typed auxiliary-state image.
+2. Replace its dedicated checkpoint job, completion signal, in-flight flag, and watermark
+   with one auxiliary-state checkpoint lifecycle.
+3. Keep offset commits and graduation imports in the shared WAL; their WAL fsync is still the
+   acknowledgement durability boundary.
+4. Reclaim covered WAL only after the consolidated snapshot's durable rename and directory
+   sync complete.
+
+This cleanup removes the independent consumer-snapshot fsync once D10 adds producer state. It
+does not remove the shared-WAL fsync required before acknowledging an offset commit, nor the
+single asynchronous fsync that makes a replacement snapshot safe for WAL truncation.
+
 ---
 
 ## Range Lifecycle and Placement

docs/data-plane/d9_offset_placement_graduation.md

@@ -179,6 +179,38 @@ positions while ordinary commits may still arrive. Sharing the reconciliation sh
 a second recovery lifecycle without pretending the two kinds of state have the same merge
 rules.
 
+### Generalization for range auxiliary state
+
+D10 extends this implemented offset-graduation path into a shared **range auxiliary-state
+graduation** framework. A placement announcement, successor-segment token, ready-source
+selection, snapshot transport, shared-WAL completion marker, joining/ready transition,
+acknowledgement drain, and reconciliation retry are placement mechanics; they should not be
+implemented again for each range-scoped ledger.
+
+The transferred snapshot is an envelope with typed sections:
+
+| Section | Merge authority | Meaning |
+|---|---|---|
+| Consumer offsets | Group generation and committed position | Where each group may safely resume |
+| Producer frontiers | Session incarnation and highest committed producer-range sequence | Which producer retries must not append again |
+
+The envelope and graduation state machine are shared, while each section owns its validation,
+monotonic merge, memory index, and snapshot encoding. A replica becomes authoritative for a
+placement only after every required section and the placement completion marker are durable.
+This avoids duplicate lifecycle machinery without hiding materially different ledger rules
+inside an untyped generic map.
+
+The same envelope becomes the node's consolidated range-auxiliary-state checkpoint. Offset
+updates and producer identities first become durable through typed records in the shared WAL.
+When reclamation needs to replace those records, one checkpoint job snapshots every auxiliary
+section at one LSN boundary, performs one asynchronous snapshot fsync, and advances one
+auxiliary watermark. Producer frontiers therefore do not add an independent snapshot file,
+fsync, or reclamation gate.
+
+The current implementation graduates consumer offsets only. Producer-frontier participation is
+part of D10 and must preserve the existing offset behavior while extending the snapshot envelope
+and readiness check.
+
 ---
 
 ## Leadership Changes

docs/data-plane/data_plane_roadmap.md

@@ -243,6 +243,7 @@ Port layout:
 | [D6: Produce/Consume API](d6_produce_consume_api.md) | Server-side produce/consume routing via redirects | D4, D5 | Server routing (client SDK: see clients/) |
 | [D7: Retention GC](d7_retention_gc.md) | Optional per-topic **age** retention; expire sealed segments oldest-first, reclaim files | D3, D5 | Opt-in, logical (time); keep-forever is the default. Disk capacity is a separate node-level concern |
 | [D8: Consumer Offset Management](d8_consumer_offset_management.md) | Generation-fenced offset tracking and consumer-group work distribution | D2, D4, D6 | Raft-backed assignment + shared-WAL offset replication |
+| [D10: Idempotent Production](d10_idempotent_production.md) | Session fencing, ordered producer-range streams, and crash-durable deduplication | D2, D3, D5, D6, D9, client C2 | Retry one logical batch without appending it twice |
 
 D1 defines the storage primitives (WAL, segment files, sparse index) and the threading model that drives them: DataPlaneActor on a dedicated OS thread (WAL + cache publish), lock-free per-segment `SegmentRingBuffer` (concurrent consumer reads without locking), and I/O thread pools (checkpoint writes + cold reads). D2–D5 extend the D1 foundation with replication, metadata integration, consumer tracking, and crash recovery. D6 adds the client-facing protocol layer (produce/consume wire format, connection management) — consumer tasks on tokio read directly from `SegmentRingBuffer`.
 
@@ -267,6 +268,11 @@ D6 (Produce API)   D7 (Retention GC)   ← D7 also depends on D3
  |
  v
 D8 (Consumer Groups)
+
+D2 + D3 + D5 + D6 + client C2
+              |
+              v
+      D10 (Idempotent Produce)
 ```
 
 D6 completes the **server-side** routing. The **client SDK** (producer, consumer,
@@ -290,8 +296,9 @@ admin) that consumes those redirects is its own track — see
 
 ## Backlog 
 
-### Exactly-Once Semantics
-Producer idempotency keys, deduplication at segment leader. Requires producer session tracking.
+### Transactions / End-to-End Exactly-Once
+Idempotent append is designed in D10. Atomic writes across ranges and atomic coupling of
+produced entries to consumer-offset commits remain future transaction work.
 
 
 

src/client/consumer/group.rs

@@ -7,11 +7,11 @@ use arc_swap::ArcSwap;
 use dashmap::DashMap;
 use uuid::Uuid;
 
-use crate::client::{Client, ClientError};
-use crate::connections::protocol::{ClientResponse, ConsumerGroupSyncAction, ControlPlaneResponse};
+use crate::client::{Client, ClientError, ClientSuccess};
+use crate::connections::protocol::{ClientResponse, ConsumerGroupSyncAction};
 use crate::control_plane::metadata::consumer_group::GenerationId;
 use crate::control_plane::metadata::{EntryId, RangeId, SyncConsumerGroupRequest, TopicId};
-use crate::data_plane::consumer_offset_management::ledger::{
+use crate::data_plane::auxiliary_states::consumer_offsets::state::{
     ConsumerOffsetKey, ConsumerOffsetPosition, ConsumerOffsetUpdate,
 };
 
@@ -201,9 +201,7 @@ impl ConsumerGroup {
             .await?
             .response
         {
-            ClientResponse::ControlPlane(ControlPlaneResponse::ConsumerGroupAssignment(
-                assignment,
-            )) => assignment,
+            ClientResponse::Ok(ClientSuccess::ConsumerGroupAssignment(assignment)) => assignment,
             _ => return Err(ClientError::UnexpectedResponse),
         };
 

src/client/consumer/mod.rs

@@ -10,11 +10,12 @@ use crate::client::consumer::topic_fetch_manager::{
 use crate::client::redirect::Served;
 use crate::client::{Client, ClientError};
 use crate::connections::protocol::{
-    ClientDataPlaneRequest, ClientResponse, DataPlaneResponse, FetchByIdRequest, RangeDetail,
-    RangeOffsetRequest, RangeProgressSignal, RangeTransition, SegmentDetail, TopicDetail,
+    ClientDataPlaneRequest, ClientResponse, FetchByIdRequest, RangeDetail, RangeOffsetRequest,
+    RangeProgressSignal, RangeTransition, SegmentDetail, TopicDetail,
 };
 use crate::control_plane::metadata::{EntryId, RangeId, RangeState, TopicId};
-use crate::data_plane::consumer_offset_management::ledger::ConsumerOffsetPosition;
+use crate::data_plane::auxiliary_states::consumer_offsets::state::ConsumerOffsetPosition;
+use crate::data_plane::messages::query::RangeOffsets;
 
 pub mod config;
 pub use config::*;
@@ -85,10 +86,10 @@ impl Consumer {
 
         if matches!(config.start_policy, StartPolicy::Latest) {
             for cursor in cursors.iter_mut() {
-                let (_, tail_entry_id) = client
+                let RangeOffsets { next_entry_id, .. } = client
                     .fetch_range_entry_ids(&topic, cursor.range_id)
                     .await?;
-                cursor.next_entry_id = tail_entry_id;
+                cursor.next_entry_id = next_entry_id;
             }
         }