feat: graduate consumer offset replicas through durable range snapshots before it can serve

CHANGELOG.md

@@ -33,6 +33,7 @@ All notable changes to this project will be documented in this file.
 - **[PR #81]** Add data models for state machine
 
 ### Data Plane & Storage
+- Graduate consumer-offset replicas through durable range snapshots before they can serve reads, while allowing joining replicas to receive live commits without delaying client acknowledgement.
 - **[PR #136]** feat: 135
 - **[PR #132]** feat: d4 - cold path
 - **[PR #126]** feat: d3 - node membership reconcilation

src/client/routing.rs

@@ -24,7 +24,7 @@ struct RangeRoute {
     /// `replica_set[0]` of the active segment, the produce target. `None` when the
     /// range has no active segment (sealed/deleting) or its replica set is empty.
     write_leader: Option<NodeAddressInfo>,
-    replicas: Box<[NodeAddressInfo]>,
+    replica_addrs: Box<[NodeAddressInfo]>,
 }
 
 // Active segment only — correct for produce (the write head) and a tailing consumer.
@@ -48,7 +48,7 @@ impl From<&RangeDetail> for RangeRoute {
                 .active_segment
                 .as_ref()
                 .and_then(|_| replicas.first().cloned()),
-            replicas,
+            replica_addrs: replicas,
         }
     }
 }
@@ -77,14 +77,14 @@ impl TopicRouting {
         self.ranges
             .iter()
             .find(|r| r.range_id == range_id)
-            .map(|r| r.replicas.as_ref())
+            .map(|r| r.replica_addrs.as_ref())
     }
 
     pub(crate) fn write_leader_for_range(&self, range_id: RangeId) -> Option<NodeAddressInfo> {
         self.ranges
             .iter()
             .find(|range| range.range_id == range_id)
-            .and_then(|range| range.replicas.first().cloned())
+            .and_then(|range| range.replica_addrs.first().cloned())
     }
 }
 
@@ -174,7 +174,7 @@ impl RoutingCache {
                     range_id: r.range_id,
                     keyspace_start: r.keyspace_start.clone(),
                     write_leader: r.write_leader.as_ref().map(|_| wrong.clone()),
-                    replicas: r.replicas.clone(),
+                    replica_addrs: r.replica_addrs.clone(),
                 })
                 .collect();
 

src/connections/controller.rs

@@ -232,7 +232,7 @@ impl ClientController {
         let cmd = CreateTopic {
             storage_policy,
             name,
-            replica_set: group.members,
+            replica_set: group.replicas,
             created_at,
         };
         Ok(self
@@ -598,7 +598,7 @@ impl ClientController {
                 shard_group_id: group.id,
                 leader_node_id: leader.as_ref().map(|e| e.leader.node_id.to_string()),
                 leader_addr: leader.map(|e| e.leader.client_addr()),
-                member_node_ids: group.members.iter().map(|n| n.to_string()).collect(),
+                member_node_ids: group.replicas.iter().map(|n| n.to_string()).collect(),
             });
         Ok(ClientResponse::Admin(AdminResponse::ShardInfo { detail }))
     }
@@ -664,7 +664,7 @@ mod tests {
     use crate::control_plane::metadata::TopicStats as MetadataTopicStats;
     use crate::control_plane::metadata::strategy::{PartitionStrategy, StoragePolicy};
     use crate::control_plane::metadata::{RangeId, TopicId, TopicMeta};
-    use crate::control_plane::{NodeAddress, NodeId, SwimNode, SwimNodeState};
+    use crate::control_plane::{NodeAddress, NodeId, Replicas, SwimNode, SwimNodeState};
     use crate::data_plane::actor::DataPlaneSender;
     use crate::data_plane::messages::DataPlaneMessage;
     use crate::data_plane::messages::command::{DataPlaneCommand, ProduceAck};
@@ -681,7 +681,7 @@ mod tests {
     fn test_shard_group() -> ShardGroup {
         ShardGroup {
             id: ShardGroupId(42),
-            members: vec![node_id("node-1")],
+            replicas: Replicas::new(vec![node_id("node-1")]),
         }
     }
 
@@ -745,7 +745,7 @@ mod tests {
         TopicMeta::new(
             "t1".into(),
             TopicId(1),
-            vec![node_id(leader)],
+            Replicas::new(vec![node_id(leader)]),
             0,
             StoragePolicy {
                 retention_ms: Some(3_600_000),
@@ -785,7 +785,7 @@ mod tests {
         let owner_addr = NodeAddress::test(addr(18003), addr(8083));
         let group = ShardGroup {
             id: ShardGroupId(42),
-            members: vec![owner.clone()],
+            replicas: Replicas::new(vec![owner.clone()]),
         };
         let swim = swim_sender_with(move |cmd| match cmd {
             SwimActorCommand::Query(QueryCommand::ResolveShardGroup { reply, .. }) => {
@@ -815,7 +815,7 @@ mod tests {
         let leader_addr = NodeAddress::test(addr(18004), addr(8084));
         let group = ShardGroup {
             id: ShardGroupId(1),
-            members: vec![me.clone(), leader.clone()],
+            replicas: Replicas::new(vec![me.clone(), leader.clone()]),
         };
         let swim = swim_sender_with(move |cmd| match cmd {
             SwimActorCommand::Query(QueryCommand::ResolveShardGroup { reply, .. }) => {
@@ -849,7 +849,7 @@ mod tests {
         let me = node_id("self");
         let group = ShardGroup {
             id: ShardGroupId(1),
-            members: vec![me.clone()],
+            replicas: Replicas::new(vec![me.clone()]),
         };
         let swim = swim_sender_with(move |cmd| {
             if let SwimActorCommand::Query(QueryCommand::ResolveShardGroup { reply, .. }) = cmd {
@@ -878,7 +878,7 @@ mod tests {
         let owner_addr = NodeAddress::test(addr(18005), addr(8085));
         let group = ShardGroup {
             id: ShardGroupId(9),
-            members: vec![owner.clone()],
+            replicas: Replicas::new(vec![owner.clone()]),
         };
         let swim = swim_sender_with(move |cmd| match cmd {
             SwimActorCommand::Query(QueryCommand::ResolveShardGroup { reply, .. }) => {
@@ -1030,7 +1030,7 @@ mod tests {
         let owner_addr = NodeAddress::test(addr(18002), addr(8082));
         let group = ShardGroup {
             id: ShardGroupId(42),
-            members: vec![owner.clone()],
+            replicas: Replicas::new(vec![owner.clone()]),
         };
         let swim = {
             let group = group.clone();
@@ -1071,7 +1071,7 @@ mod tests {
         let me = node_id("self");
         let group = ShardGroup {
             id: ShardGroupId(7),
-            members: vec![me.clone()],
+            replicas: Replicas::new(vec![me.clone()]),
         };
         let swim = swim_sender_with(move |cmd| {
             if let SwimActorCommand::Query(QueryCommand::ResolveShardGroup { reply, .. }) = cmd {

src/control_plane/consensus/messages/actor.rs

@@ -5,10 +5,12 @@ use crate::control_plane::NodeId;
 use crate::control_plane::consensus::raft::errors::ProposalError;
 use crate::control_plane::membership::ShardGroupId;
 use crate::control_plane::metadata::{ConsumerGroupAssignment, TopicMeta, TopicStats};
-use crate::data_plane::messages::command::{CatchUpAck, SealBoundaryReport, SegmentAssignmentAck};
+use crate::data_plane::messages::command::{
+    DurableSegmentEndReported, SegmentCaughtUp, SegmentPlaced,
+};
 
 use super::command::{
-    CoordinatorSealRequest, EnsureGroup, InboundRaftRpc, MetadataProposal, RaftProtocolMessage,
+    EnsureGroup, InboundRaftRpc, MetadataProposal, ProposeSegmentRoll, RaftProtocolMessage,
     RemoveGroup,
 };
 use super::timer::RaftTimeoutCallback;
@@ -51,18 +53,18 @@ pub enum MultiRaftActorCommand {
         reply: oneshot::Sender<Option<TopicMeta>>,
     },
     GetConsumerGroupAssignment(GetConsumerGroupAssignment),
-    /// Data plane SealRequest forwarded to coordinator for Raft proposal.
-    Coordinator(CoordinatorSealRequest),
-    /// Data-leader confirmation that it received a `SegmentAssignment`. Marks the
+    /// Data-plane request forwarded to the metadata coordinator for proposal.
+    ProposeSegmentRoll(ProposeSegmentRoll),
+    /// Data-leader confirmation that it received a `PlaceSegment`. Marks the
     /// segment confirmed so the leader's heartbeat sweep stops re-driving it.
-    AssignmentAck(SegmentAssignmentAck),
-    /// A survivor's reply to a leader-crash `SealBoundaryQuery` — its durable extent
+    AssignmentAck(SegmentPlaced),
+    /// A survivor's reply to a leader-crash `RequestDurableSegmentEnd` — its durable extent
     /// for the segment, gathered to recover the committed seal end.
-    SealBoundaryReport(SealBoundaryReport),
+    DurableSegmentEndReported(DurableSegmentEndReported),
     /// A replica's confirmation that it holds a reassigned sealed segment through
     /// `sealed_end`. Clears the member from the coordinator's catch-up re-drive so
     /// the heartbeat sweep stops re-announcing the assignment.
-    CatchUpAck(CatchUpAck),
+    SegmentCaughtUp(SegmentCaughtUp),
 }
 
 pub struct GetConsumerGroupAssignment {

src/control_plane/consensus/messages/command.rs

@@ -3,7 +3,7 @@ use crate::control_plane::consensus::messages::rpc::{OutboundRaftPacket, RaftRpc
 use crate::control_plane::consensus::messages::timer::RaftTimeoutCallback;
 use crate::control_plane::membership::{ShardGroup, ShardGroupId};
 use crate::control_plane::metadata::command::MetadataCommand;
-use crate::data_plane::messages::command::SealRequest;
+use crate::data_plane::messages::command::RequestSegmentRoll;
 use crate::impl_from_variant;
 
 pub struct InboundRaftRpc {
@@ -50,6 +50,6 @@ pub enum RaftTransportCommand {
 }
 
 #[derive(Debug)]
-pub struct CoordinatorSealRequest {
-    pub request: SealRequest,
+pub struct ProposeSegmentRoll {
+    pub request: RequestSegmentRoll,
 }

src/control_plane/consensus/messages/event.rs

@@ -48,11 +48,11 @@ pub enum RaftEvent {
     DisconnectPeer(NodeId),
     MetadataCommitted(MetadataCommitted),
     /// Idempotent re-delivery of assignment messages each heartbeat, so a lost
-    /// fire-and-forget send self-heals: active-segment `SegmentAssignment`s and
-    /// sealed-segment `CatchUpAssignment`s. The actor forwards them to the data
+    /// fire-and-forget send self-heals: active-segment `PlaceSegment`s and
+    /// sealed-segment `AssignSegmentCatchUp`s. The actor forwards them to the data
     /// transport.
     RedriveAssignments(Vec<DataTransportCommand>),
-    /// Leader-crash `SealBoundaryQuery` fan-out to a segment's survivors.
+    /// Leader-crash `RequestDurableSegmentEnd` fan-out to a segment's survivors.
     /// A coordinator-initiated data-plane send not tied to a committed entry.
     /// The actor just forwards these to the data transport.
     SealBoundaryQueries(Vec<DataTransportCommand>),