diff --git a/crates/vtop-broker/src/lib.rs b/crates/vtop-broker/src/lib.rs index 21187c1..2307504 100644 --- a/crates/vtop-broker/src/lib.rs +++ b/crates/vtop-broker/src/lib.rs @@ -1129,6 +1129,82 @@ 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. + /// + /// 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 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 + /// 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. + 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(( + 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(|problem| (ErrorCode::Storage, problem.to_string()))?; + 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..45025c1 100644 --- a/crates/vtop-broker/src/replication/transfer.rs +++ b/crates/vtop-broker/src/replication/transfer.rs @@ -232,6 +232,26 @@ 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 { + // 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, + }) + } + fn list_sealed_segments( &self, // This handler serves whoever reaches it; the authorization decision @@ -355,6 +375,70 @@ impl SegmentTransferClient { self } + /// 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:?}" + ))), + } + } + /// Pull every sealed segment `receiver` does not already hold. /// /// Returns the installed primary paths, oldest first. Idempotent across 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..5afd6c4 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,36 @@ 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. + let mut sealed_tail_end: Option = None; + 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}"))?; + 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}" + ); + } else { + println!( + "the leader's tail was already empty; the sealed prefix already reaches \ + offset {sealed_end}" + ); + } + } + } let installed = client .transfer_sealed_prefix( addr, @@ -1056,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!( @@ -1236,7 +1295,17 @@ 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. 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)." + } ); } } 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