protocol,broker,node,cli,chaos: a leader seals its tail on demand — repair reaches its whole position (#306)

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).
     ///

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<vtop_protocol::SealTailResponse, (ErrorCode, String)> {
+        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<crate::fencing_epochs::EpochStart> = request
                     .leader_epoch_starts

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<vtop_protocol::SealTailResponse, (ErrorCode, 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,
+        })
+    }
+
     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

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}"
+    );
+}