feat: transparent retry deduplication#201
Merged
Merged
Conversation
We have completed the refactoring of the wire protocol to flatten responses directly into ClientResponse::Ok(ClientSuccess) and ClientResponse::Err(ServerError) across the codebase: 1. Wire Protocol Structure: - Removed ControlPlaneResponse, DataPlaneResponse, and AdminResponse sub-protocol wrapper enums. - Updated ClientSuccess in mod.rs to directly contain all 16 success variants (TopicCreated, TopicDeleted, TopicList, TopicDetail, ConsumerGroupAssignment, ConsumerGroupLeft, ProducerSessionOpened, Produced, Fetched, RangeOffset, ConsumerOffsetCommitted, ConsumerOffset, ClusterInfo, TopicStats, ShardInfo, ShardLeader). - Standardized ClientResponse to consist of: - ClientResponse::Ok(ClientSuccess) - ClientResponse::Err(ServerError) - ClientResponse::Stop 2. Server Controller Integration: - Refactored handler methods in controller.rs (handle_create_topic, handle_delete_topic, handle_produce, handle_fetch, etc.) to return Result<ClientSuccess, ServerError>. 3. Client SDK & Integration Tests
refactored FetchResult and ListOffsetsResult to leverage standard Result<..., ServerError> types.
Introduce ProducerTracker to manage producer session verification, sequence tracking, and append deduplication: - Track session incarnation, expiration, and sequence frontiers in ProducerSession - Enforce strict sequence continuity and detect sequence gaps or identity conflicts - Maintain a sliding deduplication window (ProducerDeduplicationEntry) to handle idempotent retries - Track in-flight appends to prevent concurrent duplicate processing
…ryStateManager - Integrate ProducerTracker into AuxiliaryStateManager alongside consumer offsets - Refactor DataPlane::handle_produce to delegate segment validation and staging to SegmentStore - Rename SegmentTracker::stage_producer_entry to stage_entry - Standardize producer_identity across ReplicationState and commit callbacks
… and domain range resolution - Change ClientController handlers to return Result<ClientSuccess, ServerError> directly - Map MetadataError into ServerError to unify control-plane error handling - Encapsulate range routing (resolve_writable_range) and segment lookup (active_write_segment) into TopicMeta/RangeMeta - Flatten ClientResponse into Ok(ClientSuccess) and Err(ServerError)
…ce tracking - Add ClientProducerSessionManager and RangeSequences in session.rs for per-range sequence generation and topology tracking. - Retain in-flight sequence counters during range retirement and handle session lease renewals safely. - Consolidate transparent retry logic in producer flush for SessionExpired and Request InFlight errors. - Add end-to-end SDK tests for producer deduplication.
Migorithm
added
bug
During a FlappingPartition fault, a data write leader (node-1) experiences follower
replication timeouts and requests a segment roll (RequestSegmentRoll). When the metadata
state machine commits the roll, DataPlane::handle_segment_roll_committed (state.rs) replays
uncommitted producer entries into the new successor segment and sets self.needs_flush =
true.
However, handle_segment_roll_committed was not scheduling a BatchFlushTimer with the ticker
system. Without a scheduled timer (or a subsequent incoming client request), flush_batch()
was never triggered for the replayed records. As a result, the replayed entries remained in
dirty_segments, and the client's produce request was left stranded on reply_rx.await until
timing out after 40 seconds.
#### Fixes Applied
1. Timer Scheduling on Segment Roll:
• In state.rs, scheduled BatchFlushTimer::deadline() whenever
handle_segment_roll_committed sets self.needs_flush = true. This ensures flush_batch()
is promptly executed to write the WAL, trigger replication, and complete pending
producer ACK channels for successor segments.
2. Scenario Recovery Validation:
• In scenario.rs, updated is_valid() so FlappingPartition and PartitionLeader scenarios
require at least 15s of post-heal recovery budget before simulation_secs ends, ensuring
full SWIM reconnect, Raft consensus catch-up, and produce ACK verification before
simulation termination.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
closes #182
1. Architecture & Protocol Layering (d10_transparent_retry_deduplication.md)
eviction) via metadata Raft state machine.
payload_digest)), enforce sequence continuity, deduplicate retries using a bounded result ring, and persist ProducerAppendIdentity in WAL entry headers.
2. Data Plane Deduplication & Auxiliary State
snapshots, and periodic checkpointing.
3. Control Plane Session Authority
4. Client SDK & Producer Session Manager