fix(shard): drive partition repair and the commit walk from the tick

core/integration/tests/cluster/parked_frame_redispatch.rs

@@ -23,8 +23,9 @@
 //! that parked. This test pins the park path positively, and on the replica
 //! where getting it wrong creates a replica gap: a client request that never
 //! reaches the plane is answered with a retriable status and the SDK replays it,
-//! while a replicated PREPARE has no client behind it and must wait for a later
-//! commit heartbeat to arm repair if this path drops it.
+//! while a replicated PREPARE has no client behind it: nothing re-sends it once
+//! its op has quorum, so the backup gap-stops and waits out `tick_partitions`'
+//! repair debounce before anything refetches it.
 //!
 //! What makes the window wide on a backup is the commit broadcast. A backup
 //! learns a metadata commit from the `commit` field of the next prepare on that
@@ -50,7 +51,7 @@
 //! - Every acked message is readable in dense offset order, each producer's own
 //!   sends stay in the order it made them, and all three replicas hold
 //!   byte-identical segments. A prepare lost to the gap check leaves a backup
-//!   permanently short, since the gap never closes on its own.
+//!   short until the repair driver's next pass closes the gap.
 //!
 //! The harness removes an ambient `RUST_LOG` when this test supplies its explicit
 //! logging level, and the log oracle falls back from captured stdout to the
@@ -98,9 +99,14 @@ const DEGRADED_MARKERS: [&str; 3] = [
 ];
 
 /// `IggyPartition::on_replicate`'s backup gap check. A re-dispatch that appends
-/// behind an op already queued on the inbox surfaces here, and the dropped op
-/// forces repair that correct redispatch ordering should never need.
-const GAP_MARKER: &str = "dropping out-of-order prepare (gap)";
+/// behind an op already queued on the inbox surfaces here. The dropped op is
+/// refetched by `tick_partitions`' level-triggered repair driver, but only after
+/// its debounce interval, so a re-dispatch that trips this has already stalled
+/// the replica for ~1s and the marker still means the ordering broke.
+///
+/// Names its plane: the metadata plane logs its own gap drop at `warn`, which
+/// passes this test's `info` filter, and the counting below matches substrings.
+const GAP_MARKER: &str = "dropping out-of-order partition prepare (gap)";
 const PARTITION_PLANE_FIELD: &str = "plane=\"partitions\"";
 
 fn topic_name(index: u32) -> String {
@@ -347,7 +353,8 @@ fn assert_no_degraded_park_paths(harness: &TestHarness) {
         assert_eq!(
             partition_gaps, 0,
             "node {node} logged {GAP_MARKER:?}: a re-dispatched prepare lost its arrival \
-             position and forced avoidable partition repair"
+             position, and the op it displaced is recoverable only by waiting out the \
+             repair driver's debounce"
         );
     }
 }

core/partitions/src/iggy_partition.rs

@@ -140,6 +140,16 @@ where
     /// set when the recovery handshake finds this replica behind the group's
     /// commit frontier, cleared when `RepairDone` completes the walk.
     pub repair: Option<RepairSession>,
+    /// Consecutive shard-sweep ticks this partition has been seen gap-stopped
+    /// (committed ops it cannot walk to, because the op at its commit frontier
+    /// plus one is missing). Debounces the sweep's level-triggered repair arm.
+    /// Owned by the sweep, like [`Self::transfer_rearm`].
+    pub gap_ticks: u32,
+    /// Prepares the backup gap check destroyed since the shard last drained
+    /// the count. Replicated traffic has no client to answer and retransmit
+    /// skips ops that already reached quorum, so nothing else records that the
+    /// frame existed.
+    prepare_gap_drops: u64,
     /// Highest message offset recovered from segments at boot (`None` when
     /// the partition booted empty). Repaired batches at or below this line
     /// are already persisted and counted; the flush and commit paths skip
@@ -489,6 +499,8 @@ where
             consumer_offset_enforce_fsync: false,
             runtime_options: TopicRuntimeOptions::default(),
             repair: None,
+            gap_ticks: 0,
+            prepare_gap_drops: 0,
             recovered_durable_offset: None,
             installed_frontier: None,
             fatal: None,
@@ -1030,6 +1042,22 @@ where
         self.write_superblock(superblock.as_ref(), frontier).await
     }
 
+    /// Record one prepare destroyed by the backup gap check. Drained by the
+    /// shard sweep into `partition_prepare_gap_drops_total`; the counter is the
+    /// only production signal that the level-triggered repair driver has work
+    /// to do, since the drop itself answers nobody.
+    const fn note_prepare_gap_drop(&mut self) {
+        self.prepare_gap_drops = self.prepare_gap_drops.saturating_add(1);
+    }
+
+    /// Take and clear the gap-drop count.
+    #[must_use = "dropping the count loses the only record those prepares existed"]
+    pub const fn take_prepare_gap_drops(&mut self) -> u64 {
+        let drops = self.prepare_gap_drops;
+        self.prepare_gap_drops = 0;
+        drops
+    }
+
     /// Burn one transfer stall round; `true` once the budget is exhausted.
     /// Lives on the partition, not the session, so a re-minted session
     /// cannot reset it (see [`Self::transfer_attempts`]).
@@ -2425,7 +2453,7 @@ where
             return;
         }
 
-        let journal_holds_op = self.log.journal().inner.header_by_op(header.op).is_some();
+        let journal_holds_op = self.log.journal().inner.holds_op(header.op);
         if journal_holds_op {
             // Retransmit after downstream flap: durable here but commit
             // hasn't caught up. Re-forward + re-ACK so primary's view of
@@ -2521,11 +2549,12 @@ where
                     tracing::Level::WARN,
                     &PartitionDiagEvent::new(
                         self.diag_ctx(),
-                        "dropping out-of-order prepare (gap)",
+                        "dropping out-of-order partition prepare (gap)",
                     )
                     .with_operation(header.operation)
                     .with_op(header.op),
                 );
+                self.note_prepare_gap_drop();
                 return;
             }
         } else {

core/partitions/src/journal.rs

@@ -762,6 +762,20 @@ where
         headers.iter().find(|header| header.op == op).copied()
     }
 
+    /// Whether `op` is resident, without reading its header.
+    ///
+    /// Equivalent to `header_by_op(op).is_some()` and answers in O(log n)
+    /// instead of scanning: `append` writes `headers` and
+    /// `op_to_storage_offset` together and every clear site
+    /// ([`Self::commit`], [`Self::evict_prefix`], the restore path) clears
+    /// both, so the two populations cannot diverge. Callers that only need
+    /// presence must use this - a miss is the common case on the residency
+    /// checks, and a miss is exactly when the scan walks the whole vec.
+    pub fn holds_op(&self, op: u64) -> bool {
+        let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() };
+        op_to_storage_offset.contains_key(&op)
+    }
+
     /// Presence and message-carrying shape of the repair window `(floor, to_op]`
     /// in ONE pass over the header vec.
     ///

core/server/config.toml

@@ -656,6 +656,12 @@ view_probe_attempts_max = 5
 # fire-and-forget over the lossy bus, so a session with no retry wedges forever
 # on a single dropped frame. Paces both the metadata and partition repair loops;
 # must be nonzero.
+#
+# Also how long a gap-stopped partition backup waits before OPENING a repair
+# session, so shortening this makes the shard react to a replication hole
+# sooner as well. That second use carries its own floor
+# (PARTITION_GAP_DEBOUNCE_TICKS_MIN), because a value near the tick interval
+# would otherwise arm repair against ordinary pipelining.
 repair_retry_interval = "1s"
 
 # Prepares a peer serves per repair round before the requester walks to the next

core/server/src/partition_reconciler.rs

@@ -109,9 +109,21 @@
 //! `park_dropped` when it parked and then lost its namespace. A prepare has
 //! nobody to answer, so the counter is the only record it existed.
 //!
+//! A shed or discarded *prepare* is not recovered by retransmit once its op has
+//! reached quorum (`consensus::retransmit_targets` skips entries with
+//! `ok_quorum_received`), so the backup gap-stops. `tick_partitions` opens a
+//! repair session for it: its level-triggered detector arms once a partition has
+//! been gap-stopped for the repair retry interval, independently of the
+//! edge-triggered arming sites (`StartView` adoption, the commit heartbeat, the
+//! post-transfer tail), whose edges a produce stream can starve. A repair range
+//! the primary has already evicted escalates to partition state transfer. The park policy
+//! above still shrinks the exposure to a genuinely exhausted byte budget and a
+//! namespace this shard cannot serve; the driver bounds how long either costs,
+//! and `partition_prepare_gap_drops_total` counts what reached the gap check.
+//!
 //! # Known gaps
 //!
-//! Recorded here because both were previously carried as a TODO on the
+//! Recorded here because they were previously carried as a TODO on the
 //! materialization barrier this module used to promise, and the barrier is gone
 //! (see above) while these are not:
 //!
@@ -3532,9 +3544,11 @@ mod tests {
         drop(inbox);
     }
 
-    /// The replicated-prepare shape, which no other test covers and where both
-    /// park critical are worst: a prepare has no client, so discarding it forces
-    /// the backup to wait for a later commit heartbeat and journal repair.
+    /// The replicated-prepare shape, which no other test covers and where the
+    /// park path's stakes are highest: a prepare has no client, so
+    /// `deny_parked_client_request` no-ops on it and anything that discards it
+    /// loses committed data silently, recoverable only once `tick_partitions`'
+    /// repair driver notices the gap it left.
     ///
     /// A backup receives the prepare before its own metadata commits (so the frame
     /// parks unstamped), then applies the commit and materialises. The prepare must
@@ -3574,7 +3588,7 @@ mod tests {
             shard.redispatched_frame_count(),
             1,
             "the parked prepare must be staged for re-dispatch; discarding it is \
-             an avoidable gap, since a prepare has no client to retry it"
+             silent committed-data loss, since a prepare has no client to answer"
         );
         let (served, answered) = drain_inbox(&inbox);
         assert_eq!(