From 0d9cc016a1b457908330316b6fb19a3ab20784a4 Mon Sep 17 00:00:00 2001 From: Tamir Suliman Date: Tue, 11 Aug 2026 12:12:22 +0000 Subject: [PATCH 1/3] =?UTF-8?q?protocol,broker,node,cli,chaos:=20a=20leade?= =?UTF-8?q?r=20seals=20its=20tail=20on=20demand=20=E2=80=94=20repair=20rea?= =?UTF-8?q?ches=20its=20whole=20position=20(#306)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Only sealed segments transfer, and that stays true: the fix is not to transfer a moving tail but to stop it moving, deliberately. SealTail (wire kinds 81/82 — nothing below 81 is free; 70 is the live legacy cursor alias) asks the leader to roll now: the tail seals, a successor opens at its end, and the freshly sealed segment is transferable the moment the call returns. The arithmetic the issue states — a repaired replica up to 8 GiB behind a leader whose retransmission buffer holds 8 MiB — closes because the prefix now reaches the leader's position at the seal, not wherever the range last happened to roll. The issue's four questions, answered: WHO ASKS — the replica-plane RPC, driven by vtopctl node repair --seal-tail, gated by the same transfer allowlist (sealing exists FOR the transfer; a peer that may not pull the bytes has no business reshaping the leader's segments). FENCING — checked before the seal with the transfer plane's own one-snapshot check; a deposed leader refuses rather than sealing a tail the cluster moved past, and the broker-level seal runs under the append path's own lock because rolling belongs to the append critical section. REFUSALS — an empty tail over a sealed prefix is an idempotent no-op reported as zero records sealed, distinguishable from progress by a retrying repair; a never-written range is refused with the reason. COST — one shorter segment per repair, taken only on demand, and retention runs after the seal exactly as it runs after an ordinary roll. LocalBroker::seal_tail is the first production caller of SegmentSet::roll. The handler default refuses, in the fence camp: a peer that silently succeeded without sealing would report a tail transferable that is still moving. Scenario 12's inverted pin fires positive: the replacement must now reach the FULL acknowledged floor — the bound moves from the sealed- prefix end to , the repair must exit 0, and a run that stops at the old boundary fails the scenario instead of being tolerated with a caveat. Proven live: committed through 1500 >= 1500, eight sealed segments, across the leader transition. The byte-for-byte check now matches segments BY NAME — with the sealed tail in the transfer, 'the last .segment' in the two directories are different segments. --- crates/vtop-broker/src/lib.rs | 46 ++++++ crates/vtop-broker/src/replication/network.rs | 25 ++++ .../vtop-broker/src/replication/transfer.rs | 88 +++++++++++ .../tests/sealed_segment_transfer.rs | 139 ++++++++++++++++++ crates/vtop-cli/src/node_tools.rs | 62 +++++++- crates/vtop-log/src/segment_set.rs | 9 ++ crates/vtop-node/src/data_node.rs | 14 ++ crates/vtop-protocol/src/lib.rs | 60 ++++++++ .../scenarios/12-data-replica-replacement.sh | 63 ++++---- 9 files changed, 470 insertions(+), 36 deletions(-) diff --git a/crates/vtop-broker/src/lib.rs b/crates/vtop-broker/src/lib.rs index 21187c1..48de8ec 100644 --- a/crates/vtop-broker/src/lib.rs +++ b/crates/vtop-broker/src/lib.rs @@ -1129,6 +1129,52 @@ impl LocalBroker { state.segment.commit().map_err(BrokerError::from) } + /// Seal the active tail so the sealed prefix reaches this leader's + /// actual position (#306). + /// + /// Only sealed segments transfer, so before this a repair stopped at + /// wherever the range last happened to roll — up to a whole segment + /// bound behind the leader. Sealing on demand is an ordinary roll taken + /// deliberately: the tail seals, a successor opens at its end, and the + /// freshly sealed segment is transferable the moment this returns. + /// Under the SAME lock as the append path, because rolling belongs to + /// the append critical section — a produce landing mid-seal would + /// otherwise race the successor's creation. + /// + /// Returns `(sealed_end, records_sealed)`. An EMPTY tail over a sealed + /// prefix is an idempotent no-op — the prefix already reaches the + /// leader's position, records_sealed is zero, and a retrying repair can + /// tell that from progress. An empty tail with NO sealed prefix is + /// refused: a never-written range has nothing a transfer could carry, + /// and minting a degenerate sealed segment to say so would cost a file + /// and a lie. + /// + /// The caller is responsible for fencing: this is a WRITE, and the + /// transfer plane's per-request epoch check runs before it. + pub fn seal_tail(&self) -> BrokerResult<(u64, u64)> { + let mut state = self + .state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let tail_base = state.segment.active().base_offset(); + let tail_next = state.segment.next_offset(); + if tail_next == tail_base { + if state.segment.sealed().is_empty() { + return Err(BrokerError::InvalidConfig( + "this range has never held a record; there is nothing to seal and nothing a transfer could carry — produce to the range before repairing from it" + .to_owned(), + )); + } + return Ok((tail_next, 0)); + } + let records_sealed = tail_next - tail_base; + state.segment.roll_minting().map_err(BrokerError::from)?; + // A seal adds a sealed segment, and retention reasons in sealed + // units — the same follow-through the produce path's roll performs. + self.run_retention(&mut state.segment); + Ok((tail_next, records_sealed)) + } + /// Non-blocking `(local_committed_offset, next_offset)`, for observation /// only (#224). /// diff --git a/crates/vtop-broker/src/replication/network.rs b/crates/vtop-broker/src/replication/network.rs index 585c149..416a655 100644 --- a/crates/vtop-broker/src/replication/network.rs +++ b/crates/vtop-broker/src/replication/network.rs @@ -165,6 +165,25 @@ pub trait ReplicaPeerHandler: Send + Sync { "this peer does not serve sealed-segment transfer".to_owned(), )) } + + /// Seal the active tail so the sealed prefix reaches the leader's + /// position (#306). + /// + /// Defaults to REFUSING, in the [`Self::fence`] camp: sealing is a + /// WRITE, and a peer that silently "succeeded" without sealing would + /// report a tail transferable that is still moving — the answer that + /// can be mistaken for success is the one this default must not give. + fn seal_tail( + &self, + _peer: Uuid, + _range: &vtop_protocol::RangeIdentity, + _fencing_epoch: u64, + ) -> Result { + Err(( + ErrorCode::InvalidRequest, + "this peer does not serve sealed-segment transfer".to_owned(), + )) + } } impl ReplicaPeerHandler for InProcessFollower { @@ -408,6 +427,12 @@ fn dispatch_replica_frame( Err((code, message)) => error_message(code, message), } } + Message::SealTailRequest(request) => { + match handler.seal_tail(peer, &request.range, request.fencing_epoch) { + Ok(response) => Message::SealTailResponse(response), + Err((code, message)) => error_message(code, message), + } + } Message::ReplicaFenceRequest(request) => { let leader_epoch_starts: Vec = request .leader_epoch_starts diff --git a/crates/vtop-broker/src/replication/transfer.rs b/crates/vtop-broker/src/replication/transfer.rs index 2ba1d66..9f84085 100644 --- a/crates/vtop-broker/src/replication/transfer.rs +++ b/crates/vtop-broker/src/replication/transfer.rs @@ -232,6 +232,30 @@ impl ReplicaPeerHandler for LeaderSegmentTransferHandler { }) } + fn seal_tail( + &self, + // Same division of labour as the transfer RPCs: authorization + // belongs to the node that installs this handler. + _peer: Uuid, + range: &RangeIdentity, + fencing_epoch: u64, + ) -> Result { + self.check_range(range)?; + // BEFORE the seal, deliberately: sealing is a write, and the fencing + // check is what stops a deposed leader sealing a tail the cluster + // has moved past. The same one-snapshot-one-lock discipline as every + // other fenced request applies inside check_fencing. + self.check_fencing(fencing_epoch)?; + let (sealed_end, records_sealed) = self + .broker + .seal_tail() + .map_err(|problem| (ErrorCode::InvalidRequest, problem.to_string()))?; + Ok(vtop_protocol::SealTailResponse { + sealed_end, + records_sealed, + }) + } + fn list_sealed_segments( &self, // This handler serves whoever reaches it; the authorization decision @@ -362,6 +386,70 @@ impl SegmentTransferClient { /// primary renames last, so its presence proves the bundle verified), and /// a segment that was mid-receive left only ignorable staging debris the /// receiver swept at open. + /// Ask the leader to seal its active tail, so the sealed prefix reaches + /// its actual position (#306). + /// + /// Fenced like every mutating request: a deposed leader refuses rather + /// than sealing a tail the cluster has moved past. Returns + /// `(sealed_end, records_sealed)`; zero records sealed means the tail + /// was already empty and the sealed prefix already reached the leader's + /// position — an idempotent no-op, not a failure. + pub async fn seal_tail( + &self, + addr: SocketAddr, + server_name: &str, + expected_node: Uuid, + range: &RangeIdentity, + fencing_epoch: u64, + ) -> BrokerResult<(u64, u64)> { + let name = rustls::pki_types::ServerName::try_from(server_name.to_owned()) + .map_err(|error| { + crate::BrokerError::InvalidConfig(format!("server name {server_name:?}: {error}")) + })? + .to_owned(); + let mut stream = timeout(self.request_timeout, async { + let tcp = TcpStream::connect(addr) + .await + .map_err(|source| crate::BrokerError::Io { + path: PathBuf::from("seal-tail"), + source, + })?; + self.connector + .connect(name, tcp) + .await + .map_err(|source| crate::BrokerError::Io { + path: PathBuf::from("seal-tail-tls"), + source, + }) + }) + .await + .map_err(|_| crate::BrokerError::Timeout("seal tail connect"))??; + assert_peer_uuid(peer_certs_client(&stream), expected_node)?; + let mut request_id = 1_u64; + match self + .round_trip( + &mut stream, + &mut request_id, + Message::SealTailRequest(vtop_protocol::SealTailRequest { + range: range.clone(), + fencing_epoch, + }), + ) + .await? + { + Message::SealTailResponse(response) => { + Ok((response.sealed_end, response.records_sealed)) + } + Message::Error(error) => Err(crate::BrokerError::InvalidConfig(format!( + "leader refused to seal its tail: {:?} {}", + error.code, error.message + ))), + other => Err(crate::BrokerError::InvalidConfig(format!( + "unexpected reply to a seal-tail request: {other:?}" + ))), + } + } + /// The source's sealed-segment listing, served UNDER THE FENCING CHECK. /// /// Exists for two callers' reasons at once (#315): the listing's highest diff --git a/crates/vtop-broker/tests/sealed_segment_transfer.rs b/crates/vtop-broker/tests/sealed_segment_transfer.rs index 20517f8..9cd3e53 100644 --- a/crates/vtop-broker/tests/sealed_segment_transfer.rs +++ b/crates/vtop-broker/tests/sealed_segment_transfer.rs @@ -380,6 +380,20 @@ fn the_leader_refuses_by_name_what_the_transfer_plane_does_not_serve() { .unwrap_err(); assert_eq!(code, ErrorCode::Fenced); + // The strongest of the three: a SealTail served under a stale epoch + // MUTATES the leader's range (#306), so a deposed leader's client must + // be refused before anything rolls. + let sealed_before = h.leader.sealed_segment_handles().len(); + let (code, _) = handler + .seal_tail(TEST_PEER, &h.range, FENCING_EPOCH - 1) + .unwrap_err(); + assert_eq!(code, ErrorCode::Fenced); + assert_eq!( + h.leader.sealed_segment_handles().len(), + sealed_before, + "a fenced seal request must not have rolled anything" + ); + // A frontier the listing advertised as absent cannot be fetched: the // first sealed segment inherited nothing. let first = sealed @@ -915,3 +929,128 @@ fn a_condemned_range_refuses_to_open_and_to_be_adopted() { "a condemned range must not be adoptable either" ); } + +/// The #306 arithmetic, closed: sealing on demand moves the sealed prefix to +/// the leader's actual position, and the very next transfer carries what was +/// the tail. The leader keeps serving appends afterwards — the seal is an +/// ordinary roll taken deliberately, not a shutdown. +#[test] +fn sealing_the_tail_makes_the_leaders_position_transferable() { + let h = harness(); + let (_, tail_next_before) = h.leader.local_offsets(); + let sealed_end_before = h + .leader + .sealed_segment_handles() + .iter() + .map(|handle| handle.next_offset) + .max() + .unwrap(); + assert!( + sealed_end_before < tail_next_before, + "the fixture must hold tail records, or this test proves nothing about sealing them" + ); + + let (sealed_end, records_sealed) = h + .runtime + .block_on( + h.client + .seal_tail(h.addr, "localhost", LEADER, &h.range, FENCING_EPOCH), + ) + .expect("a fenced seal from an authorized repairer must succeed"); + assert_eq!( + sealed_end, tail_next_before, + "the sealed prefix must now reach the leader's position at the seal" + ); + assert_eq!(records_sealed, tail_next_before - sealed_end_before); + + // Idempotent: the tail is now empty, so a retrying repair seals nothing + // and learns that from the count rather than from an error. + let (again_end, again_records) = h + .runtime + .block_on( + h.client + .seal_tail(h.addr, "localhost", LEADER, &h.range, FENCING_EPOCH), + ) + .expect("sealing an already-empty tail is a no-op, not a failure"); + assert_eq!(again_end, sealed_end); + assert_eq!(again_records, 0, "nothing was left to seal"); + + // The transfer now carries what was the tail: the received prefix ends + // at the leader's sealed position, and adoption serves every record. + let receiver = SegmentReceiver::open(&Env::real(), h.destination.path()).unwrap(); + transfer(&h, &receiver).expect("the transfer after the seal must succeed"); + let adopted = + SegmentSet::adopt_in(&Env::real(), h.destination.path(), Uuid::from_u128(0x5EA1)).unwrap(); + assert_eq!( + adopted.next_offset(), + tail_next_before, + "the repaired replica must begin exactly at the leader's sealed position — the gap \ + the tail used to hold is what #306 exists to close" + ); + + // And the leader is still a leader: the successor tail serves appends. + let response = h.leader.handle( + Role::Producer, + WireFrame { + request_id: 999, + stream_id: 1, + message: Message::ProduceRequest(ProduceRequest { + range: h.range.clone(), + fencing_epoch: FENCING_EPOCH, + producer_id: PRODUCER, + producer_epoch: 1, + first_sequence: 48, + durability: WireDurability::LocalFsync, + records: vec![ProduceRecord { + timestamp_millis: 99_000, + key: b"k".to_vec(), + value: vec![b'y'; 64], + }], + }), + }, + ); + match response.message { + Message::ProduceResponse(_) => {} + other => panic!("the leader must keep serving appends after a seal: {other:?}"), + } +} + +/// A range that has never held a record refuses the seal with a reason: a +/// degenerate sealed segment would cost a file to say nothing, and adoption +/// would still refuse the empty prefix it decorated. +#[test] +fn sealing_a_never_written_range_is_refused_with_a_reason() { + let dir = tempfile::tempdir().unwrap(); + let range = range_identity(); + let descriptor = SegmentDescriptor { + segment_id: Uuid::from_u128(0xE1), + topic: range.topic.clone(), + topic_epoch: range.topic_epoch, + lineage: RangeLineage { + range_id: range.range_id, + generation: range.range_generation, + key_range: KeyRange::full(), + parents: Vec::new(), + }, + base_offset: 0, + }; + let segment = SegmentSet::create_in( + &Env::real(), + dir.path(), + descriptor, + SegmentConfig::default(), + ) + .unwrap(); + let epochs = ProducerEpochJournal::open(dir.path().join("epochs")).unwrap(); + let broker = Arc::new(LocalBroker::new(segment, epochs, range.clone(), FENCING_EPOCH).unwrap()); + let handler = LeaderSegmentTransferHandler::new(Arc::clone(&broker)); + + let (code, message) = handler + .seal_tail(TEST_PEER, &range, FENCING_EPOCH) + .unwrap_err(); + assert_eq!(code, ErrorCode::InvalidRequest); + assert!( + message.contains("never held a record"), + "the refusal must say why a never-written range cannot be repaired from: {message}" + ); +} diff --git a/crates/vtop-cli/src/node_tools.rs b/crates/vtop-cli/src/node_tools.rs index 48e8293..6e7a366 100644 --- a/crates/vtop-cli/src/node_tools.rs +++ b/crates/vtop-cli/src/node_tools.rs @@ -72,6 +72,14 @@ pub enum NodeCommand { /// the serving side and must match what metadata granted. #[arg(long)] fencing_epoch: u64, + /// Ask the leader to SEAL its active tail first (#306), so the + /// transferred prefix reaches the leader's position at the seal + /// rather than wherever the range last happened to roll. Without + /// this, up to a whole segment bound of records stays behind in a + /// tail that only the append path can deliver. Fenced like the + /// transfer itself; sealing early costs one shorter segment. + #[arg(long)] + seal_tail: bool, #[arg(long, default_value_t = 30, value_parser = clap::value_parser!(u64).range(1..=3600))] timeout_seconds: u64, }, @@ -449,6 +457,7 @@ async fn run_inner(command: NodeCommand, json: bool) -> Result { from, into, fencing_epoch, + seal_tail, timeout_seconds, } => { let config = load_config(&common.config)?; @@ -457,6 +466,7 @@ async fn run_inner(command: NodeCommand, json: bool) -> Result { from, &into, fencing_epoch, + seal_tail, Duration::from_secs(timeout_seconds), json, ) @@ -799,6 +809,7 @@ async fn repair( from: Uuid, into: &Path, fencing_epoch: u64, + seal_tail: bool, timeout: Duration, json: bool, ) -> Result { @@ -923,7 +934,9 @@ async fn repair( // This matters because re-running repair is the natural reflex when a gap // is reported, and it is the wrong move: a gap lives in the source's ACTIVE // segment, which never transfers, so no second repair can shrink it. Only - // the replica catching up from the leader can. + // the replica catching up from the leader can — or a fresh repair into an + // empty directory with --seal-tail, which moves the sealed prefix to the + // leader's position first (#306). if marker.exists() { let adopted = std::fs::read_dir(into) .map_err(|error| format!("read {}: {error}", into.display()))? @@ -938,12 +951,12 @@ async fn repair( return Err(format!( "{} was repaired already — it holds an adopted range with an active segment at \ {}, so this repair finished and running it again would re-seed a live range. If \ - a gap was reported, another repair cannot close it: those records are in the \ - source's active segment, which never transfers, and only the replica catching \ - up from the leader will bring them over. Start the replica against this \ - directory. If it is refused for a base-offset mismatch then the gap was too \ - large to catch up, and the remedy is a fresh repair into an EMPTY directory \ - rather than a second one over this.", + a gap was reported, another repair over this directory cannot close it: those \ + records are in the source's active segment, which never transfers. Start the \ + replica against this directory and let it catch up. If it is refused for a \ + base-offset mismatch then the gap was too large to catch up, and the remedy is \ + a fresh repair into an EMPTY directory with --seal-tail, which seals the \ + leader's tail so the transferred prefix reaches its position (#306).", into.display(), entry.path().display() )); @@ -981,6 +994,32 @@ async fn repair( ) })?; let addr = resolve_endpoint(&source.addr).map_err(|error| error.to_string())?; + // BEFORE the transfer, so the listing that drives it already includes + // the freshly sealed tail (#306). Ordering is the point: sealing after + // the transfer would close the gap for the NEXT repair, not this one. + if seal_tail { + let (sealed_end, records_sealed) = client + .seal_tail( + addr, + &source.server_name, + source.node_uuid, + &config.range.identity(), + fencing_epoch, + ) + .await + .map_err(|error| format!("seal the tail on {from}: {error}"))?; + if !json { + if records_sealed > 0 { + println!( + "the leader sealed {records_sealed} tail record(s); the sealed prefix now reaches offset {sealed_end}" + ); + } else { + println!( + "the leader's tail was already empty; the sealed prefix already reaches offset {sealed_end}" + ); + } + } + } let installed = client .transfer_sealed_prefix( addr, @@ -1236,7 +1275,14 @@ async fn repair( so its bytes can be superseded by a truncation mid-copy. They must be replayed \ from the leader's retransmission buffer when the replica starts, and that buffer \ is bounded (8 MiB by default). If the gap exceeds it the replica will be refused \ - for a base-offset mismatch and this repair will not have been enough." + for a base-offset mismatch and this repair will not have been enough.{}", + if seal_tail { + "\nThe tail WAS sealed for this repair, so these records arrived after the \ + seal — a retry narrows the gap to whatever lands during the next transfer." + } else { + "\nRe-running with --seal-tail into an empty directory would seal the tail \ + first, so the transferred prefix reaches the leader's position (#306)." + } ); } } diff --git a/crates/vtop-log/src/segment_set.rs b/crates/vtop-log/src/segment_set.rs index 65b4f44..40d7fc4 100644 --- a/crates/vtop-log/src/segment_set.rs +++ b/crates/vtop-log/src/segment_set.rs @@ -942,6 +942,15 @@ impl SegmentSet { Ok(()) } + /// [`Self::roll`], minting the successor's identity from this set's own + /// environment — for a caller with no meaningful id to choose, like a + /// leader sealing its tail on demand (#306). Same minting as + /// `append_group_minting` and cross-segment truncation. + pub fn roll_minting(&mut self) -> VtopLogResult<()> { + let successor_id = Uuid::from_u128(self.env.rng.next_u128()); + self.roll(successor_id) + } + /// Change the range's roll thresholds, from the tail forward (#314). /// /// A segment's limits live in its header and nowhere else, and that diff --git a/crates/vtop-node/src/data_node.rs b/crates/vtop-node/src/data_node.rs index 6e9b0f0..3c62f98 100644 --- a/crates/vtop-node/src/data_node.rs +++ b/crates/vtop-node/src/data_node.rs @@ -330,6 +330,20 @@ impl ReplicaPeerHandler for LeaderStatusReplica { self.authorize_transfer(peer)?; self.transfer.fetch_segment_chunk(peer, request) } + + fn seal_tail( + &self, + peer: Uuid, + range: &vtop_protocol::RangeIdentity, + fencing_epoch: u64, + ) -> Result { + // The transfer allowlist gates the seal too: sealing exists FOR the + // transfer, and a peer that may not pull the bytes has no business + // reshaping the leader's segments to prepare for a pull it will be + // refused. + self.authorize_transfer(peer)?; + self.transfer.seal_tail(peer, range, fencing_epoch) + } } impl LeaderStatusReplica { diff --git a/crates/vtop-protocol/src/lib.rs b/crates/vtop-protocol/src/lib.rs index e803f53..dd4f319 100644 --- a/crates/vtop-protocol/src/lib.rs +++ b/crates/vtop-protocol/src/lib.rs @@ -496,6 +496,34 @@ pub struct ListSealedSegmentsRequest { pub fencing_epoch: u64, } +/// Ask a leader to SEAL its active tail, so the sealed prefix reaches its +/// actual position (#306). +/// +/// Only sealed segments transfer, and a leader may be gigabytes past the +/// sealed prefix inside a tail catch-up cannot replay — so a repair without +/// this stops short by exactly that tail. Sealing is a WRITE: it carries the +/// same fencing discipline as the append plane, or a deposed leader could +/// seal a tail the cluster has moved past. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SealTailRequest { + pub range: RangeIdentity, + /// Range-leader fencing epoch, checked exactly as the append plane checks + /// it. + pub fencing_epoch: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SealTailResponse { + /// The end of the sealed prefix AFTER the seal — the offset a transfer + /// taken now can reach. + pub sealed_end: u64, + /// How many records the seal moved out of the tail. Zero means the tail + /// was already empty and the call was an idempotent no-op — the sealed + /// prefix already reached the leader's position — which a retrying + /// repair must be able to distinguish from progress. + pub records_sealed: u64, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct ListSealedSegmentsResponse { /// A CONTIGUOUS ascending run: each entry begins exactly where the @@ -612,6 +640,8 @@ pub enum Message { ListSealedSegmentsResponse(ListSealedSegmentsResponse), FetchSegmentChunkRequest(FetchSegmentChunkRequest), FetchSegmentChunkResponse(FetchSegmentChunkResponse), + SealTailRequest(SealTailRequest), + SealTailResponse(SealTailResponse), } impl Message { @@ -655,6 +685,8 @@ impl Message { Self::ListSealedSegmentsResponse(_) => 78, Self::FetchSegmentChunkRequest(_) => 79, Self::FetchSegmentChunkResponse(_) => 80, + Self::SealTailRequest(_) => 81, + Self::SealTailResponse(_) => 82, } } } @@ -891,6 +923,8 @@ fn encoded_payload_size(message: &Message, limits: ProtocolLimits) -> Result range_size(&value.range)? + 8, + Message::SealTailResponse(_) => 8 + 8, }; if total > limits.max_payload_bytes() { return Err(ProtocolError::Limit(format!( @@ -1021,6 +1055,8 @@ fn decode_header(header: &[u8], limits: ProtocolLimits) -> Result) -> Result<(), ProtocolEr put_u64(out, value.total_bytes); put_bytes(out, &value.bytes)?; } + Message::SealTailRequest(value) => { + put_range(out, &value.range)?; + put_u64(out, value.fencing_epoch); + } + Message::SealTailResponse(value) => { + put_u64(out, value.sealed_end); + put_u64(out, value.records_sealed); + } } Ok(()) } @@ -1646,6 +1690,14 @@ fn decode_message( let bytes = decoder.bounded_bytes(MAX_SEGMENT_CHUNK_BYTES as usize, "segment chunk")?; Message::FetchSegmentChunkResponse(FetchSegmentChunkResponse { total_bytes, bytes }) } + 81 => Message::SealTailRequest(SealTailRequest { + range: decoder.range()?, + fencing_epoch: decoder.u64()?, + }), + 82 => Message::SealTailResponse(SealTailResponse { + sealed_end: decoder.u64()?, + records_sealed: decoder.u64()?, + }), other => return Err(ProtocolError::UnknownKind(other)), }) } @@ -2279,6 +2331,14 @@ mod tests { total_bytes: 96, bytes: vec![7; 32], }), + Message::SealTailRequest(SealTailRequest { + range: range(), + fencing_epoch: 12, + }), + Message::SealTailResponse(SealTailResponse { + sealed_end: 4096, + records_sealed: 640, + }), ]; for (request_id, message) in messages.into_iter().enumerate() { let frame = WireFrame { diff --git a/scripts/live-chaos/scenarios/12-data-replica-replacement.sh b/scripts/live-chaos/scenarios/12-data-replica-replacement.sh index eba98c9..31bba98 100755 --- a/scripts/live-chaos/scenarios/12-data-replica-replacement.sh +++ b/scripts/live-chaos/scenarios/12-data-replica-replacement.sh @@ -351,19 +351,25 @@ REPAIR_CONFIG="$(emit_repair_config 0)" REPAIR_EXIT=0 "$VTOPCTL" node repair \ --config "$REPAIR_CONFIG" --from "$LEADER_UUID" --into "$SPARE_DIR" \ - --fencing-epoch "$EPOCH" > "$WORKDIR/logs/repair.log" 2>&1 || REPAIR_EXIT=$? -# 0 current, 1 behind by a measured gap, 2 adopted but unmeasured. A gap is -# EXPECTED here: the records written after the seal are in the leader's active -# segment, which never transfers, and the replica closes that by catching up. -[[ "$REPAIR_EXIT" -le 1 ]] \ - || fail "repair failed (exit $REPAIR_EXIT): $(tail -5 "$WORKDIR/logs/repair.log")" + --fencing-epoch "$EPOCH" --seal-tail > "$WORKDIR/logs/repair.log" 2>&1 || REPAIR_EXIT=$? +# EXACTLY 0, the inverted pin #306 finally fires positive on: --seal-tail +# seals the leader's tail before the transfer, so the prefix reaches the +# leader's position and no gap remains. Nothing produces during the repair in +# this scenario, so any nonzero exit here means the seal did not do its job — +# the gap this suite spent two releases stating out loud is supposed to be +# closed now. +[[ "$REPAIR_EXIT" -eq 0 ]] \ + || fail "repair with --seal-tail must close the gap entirely (exit $REPAIR_EXIT): \ +$(tail -5 "$WORKDIR/logs/repair.log")" log "repair finished with exit $REPAIR_EXIT: $(grep -c . "$WORKDIR/logs/repair.log") lines of report" -SEALED_COPY="" -for candidate in "$SPARE_DIR"/*.segment; do - [[ -f "$candidate" ]] && SEALED_COPY="$candidate" -done -[[ -n "$SEALED_COPY" ]] || fail "repair left no sealed segment in $SPARE_DIR" +# MATCHED BY NAME, not by position: with --seal-tail the transfer carries the +# sealed former tail too, so "the last .segment" in the two directories are +# different segments — the source was captured before the seal existed. The +# byte-for-byte claim is per segment, so compare the file with the same name. +SEALED_COPY="$SPARE_DIR/$(basename "$SEALED_SOURCE")" +[[ -f "$SEALED_COPY" ]] \ + || fail "repair left no copy of $(basename "$SEALED_SOURCE") in $SPARE_DIR" # BYTE-EXACT, not merely valid. The transfer's whole claim is that the bytes # are the same bytes; a copy that verified on its own but differed would be a @@ -571,31 +577,32 @@ done [[ "$SPARE_SEALED" -gt 0 ]] \ || fail "the replacement lost its sealed segments to the leader transition; the repaired \ range was truncated, which is #315 reopened" -# The committed offset must HOLD THE SEALED-PREFIX END, not merely leave files -# on disk: a truncation would drag the boundary back toward the base even if a -# directory listing still showed segments. Deadline-polled, because the -# restarted leader's HWM stream has to reach the newcomer before its boundary -# is republished. +# The committed offset must HOLD THE ACKNOWLEDGED FLOOR, not merely leave +# files on disk: a truncation would drag the boundary back toward the base +# even if a directory listing still showed segments. Deadline-polled, because +# the restarted leader's HWM stream has to reach the newcomer before its +# boundary is republished. # -# The bound is $SEG_NEXT — the end of the sealed prefix the repair installed — -# and NOT the acknowledged floor of $ACKED. The records in ($SEG_NEXT, $ACKED] -# live in the leader's active tail, which only the append path can deliver, -# and a restarted leader's retransmission buffer is empty: that remaining gap -# is #306's boundary, stated below rather than silently waited on. +# The bound is $ACKED — the FULL acknowledged floor — which for two releases +# this scenario could only state as out of reach: the tail records lived in +# the leader's active segment, which never transfers. --seal-tail (#306) +# sealed that tail before the transfer, so the repaired prefix carries every +# acknowledged record and this is the inverted pin fired positive: a +# replacement that stops at the old sealed-prefix end now FAILS the scenario +# instead of being tolerated with a caveat. SPARE_DEADLINE=$((SECONDS + PROGRESS_TIMEOUT_SECONDS)) SPARE_OFFSET="$(follower_committed_offset 3)" -while [[ "$SPARE_OFFSET" -lt "$SEG_NEXT" ]]; do +while [[ "$SPARE_OFFSET" -lt "$ACKED" ]]; do [[ $SECONDS -lt $SPARE_DEADLINE ]] \ || fail "the replacement kept $SPARE_SEALED sealed segment(s) but its committed offset \ -reads $SPARE_OFFSET, below the sealed-prefix end of $SEG_NEXT; the repaired range was \ -truncated or never re-served, which is #315 reopened" +reads $SPARE_OFFSET, below the acknowledged floor of $ACKED; either the repaired range was \ +truncated (#315 reopened) or the sealed tail did not reach the replica (#306 reopened)" sleep 1 SPARE_OFFSET="$(follower_committed_offset 3)" done -log "the replacement survived the leader transition with $SPARE_SEALED sealed segment(s), \ -committed through $SPARE_OFFSET >= $SEG_NEXT — the repair outlives the failover that used to \ -erase it. The $((ACKED - SEG_NEXT)) tail records above the sealed prefix remain out of its \ -reach until #306 gives the gap a road back" +log "the replacement survived the leader transition committed through $SPARE_OFFSET >= $ACKED \ +with $SPARE_SEALED sealed segment(s) — the repair carries the leader's WHOLE position, tail \ +included, and outlives the failover. The gap #306 tracked is closed" # Named directly rather than through `${!name}`: indirect expansion is not a # USE as far as shellcheck is concerned, so the variables stayed flagged and From 462730d3e1d1689f121ec4b8ca819c452b32c1b1 Mon Sep 17 00:00:00 2001 From: Tamir Suliman Date: Tue, 11 Aug 2026 12:30:02 +0000 Subject: [PATCH 2/3] broker,cli: the seal is fenced under the append path's own locks, and retention waits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round one, both P1s verified real and fixed. The seal's fencing check ran through the transfer handler's one-shot check_fencing, then took the broker state lock separately — leaving a window where a grant or release lands between the check and the roll, and a deposed leader still mutates its layout. The produce path already documents the discipline that closes it: metadata lease view first, then broker state, BOTH held through the write. seal_tail now takes the range and epoch and validates them inside, under both locks, with the same check_range the produce path uses — the handler's job reduces to authorization and transport. And retention no longer runs in the seal: a bytes-bound pass could reclaim the very segment the call just sealed — the committed floor covers it by construction — and the RPC would then report a sealed_end the transfer listing cannot reach. The seal exists to make the tail transferable; housekeeping that could unmake it waits for the next ordinary roll. Also from review: the operator hint after a post-seal gap now says a SECOND repair over the adopted directory is refused and names the fresh-directory remedy; the client method no longer splits transfer_sealed_prefix's doc block; and the never-written refusal's string uses proper continuations instead of embedding an indent run. --- crates/vtop-broker/src/lib.rs | 47 ++++++++++++++----- .../vtop-broker/src/replication/transfer.rs | 30 +++++------- crates/vtop-cli/src/node_tools.rs | 5 +- 3 files changed, 51 insertions(+), 31 deletions(-) diff --git a/crates/vtop-broker/src/lib.rs b/crates/vtop-broker/src/lib.rs index 48de8ec..1bd4c18 100644 --- a/crates/vtop-broker/src/lib.rs +++ b/crates/vtop-broker/src/lib.rs @@ -1137,9 +1137,20 @@ impl LocalBroker { /// bound behind the leader. Sealing on demand is an ordinary roll taken /// deliberately: the tail seals, a successor opens at its end, and the /// freshly sealed segment is transferable the moment this returns. - /// Under the SAME lock as the append path, because rolling belongs to - /// the append critical section — a produce landing mid-seal would - /// otherwise race the successor's creation. + /// + /// FENCED IN HERE, under the produce path's own lock discipline — + /// metadata lease view first, then broker state, both held through the + /// roll — so a grant or release cannot land between the fencing check + /// and the seal. A check performed outside these locks would leave + /// exactly that window, and sealing is a write a deposed leader must + /// not perform (review: the one-snapshot check alone was not enough). + /// + /// RETENTION DOES NOT RUN HERE, deliberately. A bytes-bound retention + /// pass could reclaim the very segment this call just sealed — the + /// committed floor covers it by construction — and the RPC would then + /// report a `sealed_end` the transfer listing cannot reach. The seal + /// exists to make the tail transferable; housekeeping that could + /// unmake it waits for the next ordinary roll. /// /// Returns `(sealed_end, records_sealed)`. An EMPTY tail over a sealed /// prefix is an idempotent no-op — the prefix already reaches the @@ -1148,30 +1159,40 @@ impl LocalBroker { /// refused: a never-written range has nothing a transfer could carry, /// and minting a degenerate sealed segment to say so would cost a file /// and a lie. - /// - /// The caller is responsible for fencing: this is a WRITE, and the - /// transfer plane's per-request epoch check runs before it. - pub fn seal_tail(&self) -> BrokerResult<(u64, u64)> { + pub fn seal_tail( + &self, + range: &RangeIdentity, + fencing_epoch: u64, + ) -> Result<(u64, u64), (ErrorCode, String)> { + // Lock order: metadata lease view, then broker state — identical to + // the produce path, which documents why: held together through the + // write, a concurrent grant/release cannot revoke between the + // fencing check and the mutation. + let meta = self.meta_fencing_epoch.lock(); let mut state = self .state .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); + self.check_range(&meta, range, fencing_epoch) + .map_err(|(code, message)| (code, message.to_owned()))?; let tail_base = state.segment.active().base_offset(); let tail_next = state.segment.next_offset(); if tail_next == tail_base { if state.segment.sealed().is_empty() { - return Err(BrokerError::InvalidConfig( - "this range has never held a record; there is nothing to seal and nothing a transfer could carry — produce to the range before repairing from it" + return Err(( + ErrorCode::InvalidRequest, + "this range has never held a record; there is nothing to seal and nothing \ + a transfer could carry — produce to the range before repairing from it" .to_owned(), )); } return Ok((tail_next, 0)); } let records_sealed = tail_next - tail_base; - state.segment.roll_minting().map_err(BrokerError::from)?; - // A seal adds a sealed segment, and retention reasons in sealed - // units — the same follow-through the produce path's roll performs. - self.run_retention(&mut state.segment); + state + .segment + .roll_minting() + .map_err(|problem| (ErrorCode::Storage, problem.to_string()))?; Ok((tail_next, records_sealed)) } diff --git a/crates/vtop-broker/src/replication/transfer.rs b/crates/vtop-broker/src/replication/transfer.rs index 9f84085..45025c1 100644 --- a/crates/vtop-broker/src/replication/transfer.rs +++ b/crates/vtop-broker/src/replication/transfer.rs @@ -240,16 +240,12 @@ impl ReplicaPeerHandler for LeaderSegmentTransferHandler { range: &RangeIdentity, fencing_epoch: u64, ) -> Result { - self.check_range(range)?; - // BEFORE the seal, deliberately: sealing is a write, and the fencing - // check is what stops a deposed leader sealing a tail the cluster - // has moved past. The same one-snapshot-one-lock discipline as every - // other fenced request applies inside check_fencing. - self.check_fencing(fencing_epoch)?; - let (sealed_end, records_sealed) = self - .broker - .seal_tail() - .map_err(|problem| (ErrorCode::InvalidRequest, problem.to_string()))?; + // Fencing is checked INSIDE the broker, under the produce path's own + // lock discipline, held through the roll. A check out here — this + // handler's one-shot check_fencing — leaves a window between the + // check and the seal where a grant or release can land, and sealing + // is a write a deposed leader must not perform (review P1). + let (sealed_end, records_sealed) = self.broker.seal_tail(range, fencing_epoch)?; Ok(vtop_protocol::SealTailResponse { sealed_end, records_sealed, @@ -379,13 +375,6 @@ impl SegmentTransferClient { self } - /// Pull every sealed segment `receiver` does not already hold. - /// - /// Returns the installed primary paths, oldest first. Idempotent across - /// interruptions: a segment whose primary already landed is skipped (the - /// primary renames last, so its presence proves the bundle verified), and - /// a segment that was mid-receive left only ignorable staging debris the - /// receiver swept at open. /// Ask the leader to seal its active tail, so the sealed prefix reaches /// its actual position (#306). /// @@ -450,6 +439,13 @@ impl SegmentTransferClient { } } + /// Pull every sealed segment `receiver` does not already hold. + /// + /// Returns the installed primary paths, oldest first. Idempotent across + /// interruptions: a segment whose primary already landed is skipped (the + /// primary renames last, so its presence proves the bundle verified), and + /// a segment that was mid-receive left only ignorable staging debris the + /// receiver swept at open. /// The source's sealed-segment listing, served UNDER THE FENCING CHECK. /// /// Exists for two callers' reasons at once (#315): the listing's highest diff --git a/crates/vtop-cli/src/node_tools.rs b/crates/vtop-cli/src/node_tools.rs index 6e7a366..a5f0630 100644 --- a/crates/vtop-cli/src/node_tools.rs +++ b/crates/vtop-cli/src/node_tools.rs @@ -1278,7 +1278,10 @@ async fn repair( for a base-offset mismatch and this repair will not have been enough.{}", if seal_tail { "\nThe tail WAS sealed for this repair, so these records arrived after the \ - seal — a retry narrows the gap to whatever lands during the next transfer." + seal. Starting the replica lets it replay them from the leader; if the gap \ + exceeds the retransmission buffer, run a FRESH repair into an empty \ + directory with --seal-tail — this directory is adopted now, and a second \ + repair over it is refused." } else { "\nRe-running with --seal-tail into an empty directory would seal the tail \ first, so the transferred prefix reaches the leader's position (#306)." From 0a84b1b645bb10a6f7d03e53d0e96e94de521a17 Mon Sep 17 00:00:00 2001 From: Tamir Suliman Date: Tue, 11 Aug 2026 12:40:20 +0000 Subject: [PATCH 3/3] broker,cli: the seal's persistence window is stated, detected, and named MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round two. Codex is right that suppressing retention inside seal_tail closes only the in-call window: retention runs after every successful append, so a produce landing between the seal RPC and the repair's listing can still reclaim the freshly sealed segment under a bytes bound smaller than it. What does NOT follow is a lie — a shorter listing becomes a measured, reported gap (exit 1), and a segment reclaimed mid-fetch is a clean resumable refusal, the same window every listed segment already lives in until its chunks land. A cross-RPC pin was considered and rejected, with the reason in the doc: the leader cannot know when a repairer is done, and a pin that outlives a crashed repairer is a retention bound that silently stopped being one. What the operator gets instead is the CAUSE, named: the repair compares the seal's promised end with what the fenced listing actually held, and when retention won the race it says so — 'the leader's retention reclaimed sealed segments between the two' — instead of presenting a mysteriously shorter prefix. cubic's doc precision applied: the comment no longer claims the segment persists until a later roll. Two more embedded-whitespace string literals fixed in the same pass. --- crates/vtop-broker/src/lib.rs | 15 ++++++++++++--- crates/vtop-cli/src/node_tools.rs | 24 ++++++++++++++++++++++-- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/crates/vtop-broker/src/lib.rs b/crates/vtop-broker/src/lib.rs index 1bd4c18..2307504 100644 --- a/crates/vtop-broker/src/lib.rs +++ b/crates/vtop-broker/src/lib.rs @@ -1148,9 +1148,18 @@ impl LocalBroker { /// RETENTION DOES NOT RUN HERE, deliberately. A bytes-bound retention /// pass could reclaim the very segment this call just sealed — the /// committed floor covers it by construction — and the RPC would then - /// report a `sealed_end` the transfer listing cannot reach. The seal - /// exists to make the tail transferable; housekeeping that could - /// unmake it waits for the next ordinary roll. + /// report a `sealed_end` the transfer listing cannot reach from within + /// this very call. What this does NOT promise is persistence: retention + /// runs after every successful append, so a produce landing between + /// this seal and the repair's listing can still reclaim the segment + /// under a bound smaller than the sealed tail. That window is the same + /// one every listed segment already lives in until its chunks are + /// fetched, and both ends of it fail HONESTLY — a shorter listing is a + /// measured, reported gap (exit 1), and a segment reclaimed mid-fetch + /// is a clean, resumable refusal. A cross-RPC pin was considered and + /// rejected: the leader cannot know when a repairer is done, and a pin + /// that outlives a crashed repairer is a retention bound that silently + /// stopped being one. /// /// Returns `(sealed_end, records_sealed)`. An EMPTY tail over a sealed /// prefix is an idempotent no-op — the prefix already reaches the diff --git a/crates/vtop-cli/src/node_tools.rs b/crates/vtop-cli/src/node_tools.rs index a5f0630..5afd6c4 100644 --- a/crates/vtop-cli/src/node_tools.rs +++ b/crates/vtop-cli/src/node_tools.rs @@ -997,6 +997,7 @@ async fn repair( // BEFORE the transfer, so the listing that drives it already includes // the freshly sealed tail (#306). Ordering is the point: sealing after // the transfer would close the gap for the NEXT repair, not this one. + let mut sealed_tail_end: Option = None; if seal_tail { let (sealed_end, records_sealed) = client .seal_tail( @@ -1008,14 +1009,17 @@ async fn repair( ) .await .map_err(|error| format!("seal the tail on {from}: {error}"))?; + sealed_tail_end = Some(sealed_end); if !json { if records_sealed > 0 { println!( - "the leader sealed {records_sealed} tail record(s); the sealed prefix now reaches offset {sealed_end}" + "the leader sealed {records_sealed} tail record(s); the sealed prefix now \ + reaches offset {sealed_end}" ); } else { println!( - "the leader's tail was already empty; the sealed prefix already reaches offset {sealed_end}" + "the leader's tail was already empty; the sealed prefix already reaches \ + offset {sealed_end}" ); } } @@ -1095,6 +1099,22 @@ async fn repair( ) })?; let sealed_end = listing.iter().map(|entry| entry.next_offset).max(); + // The seal's promise is checked against what the listing actually held: + // retention runs after every append on the leader, so a produce landing + // between the seal and this listing can reclaim the freshly sealed + // segment under a bytes bound smaller than it. Nothing lies — the gap + // is measured and reported below — but the operator deserves the CAUSE + // named rather than a mysteriously shorter prefix (#306 review). + if let (Some(promised), Some(listed)) = (sealed_tail_end, sealed_end) { + if listed < promised && !json { + println!( + "note: the leader sealed through offset {promised}, but the transfer listing \ + reaches only {listed} — the leader's retention reclaimed sealed segments \ + between the two (its bound is smaller than what was sealed). The repair \ + carries what remained; the reported gap below includes the reclaimed records" + ); + } + } if let Some(newest) = history.last() { if newest.epoch > fencing_epoch { return Err(format!(