From 0c99709990121caa102de7b13c9c49184cab2174 Mon Sep 17 00:00:00 2001 From: jx12n Date: Tue, 9 Jun 2026 18:06:58 -0600 Subject: [PATCH] fix(replication): redeliver_missed shipped to the wrong shard over gRPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Partition recovery (heal_region) and await_convergence never worked over real GrpcTransports: redeliver_missed sent to entry.source_shard (the LEADER's shard), but a follower's transport only knows its own shard as a peer (self-loop wiring in tidal-server::cluster), so every recovery ship failed the peer lookup — and the error was swallowed by 'let _ ='. ChannelTransport ignores the destination, which is why the in-process suite never caught it. Found live on k3s: POST /cluster/partition + writes + /cluster/heal left the region at lag=3 forever, no log line. Fix: ship to transport.local_shard() (matching the eager path's send_segment(ShardId(region.0), ...)); keep entry.source_shard in the payload (receiver advances applied_seqno per SOURCE shard). WARN on ship failure instead of dropping it. Adds a RecordingTransport regression test that locks dest == local_shard + payload source == leader — the exact case channels cannot catch. cargo test: 263 replication + 13 testing + tidal-server all pass. --- tidal/src/testing/cluster_transport.rs | 110 ++++++++++++++++++++++++- 1 file changed, 109 insertions(+), 1 deletion(-) diff --git a/tidal/src/testing/cluster_transport.rs b/tidal/src/testing/cluster_transport.rs index bdac090..15c01c6 100644 --- a/tidal/src/testing/cluster_transport.rs +++ b/tidal/src/testing/cluster_transport.rs @@ -140,7 +140,21 @@ pub(super) const fn single_event_payload( /// would flood the transport with redundant payloads; relying on it alone /// (without the receiver's monotonic advance) would double-apply on any race /// between the `applied_seqno` read here and the receiver thread's advance. +/// +/// 3. **Destination is the follower's own shard.** `send_segment`'s first +/// argument is the DESTINATION peer, and a follower's transport only knows +/// its own shard as a peer (the gRPC self-loop wiring in +/// `tidal-server::cluster`). The eager path ships with +/// `send_segment(ShardId(region.0), …)`; re-delivery must match it via +/// `transport.local_shard()` — the follower this transport belongs to. +/// Shipping to `entry.source_shard` (the LEADER's shard) fails the peer +/// lookup on every `GrpcTransport` and recovery silently never happens. +/// `ChannelTransport` ignores the destination, which is exactly why the +/// in-process suite cannot catch that regression — the unit test below +/// locks it instead. The payload itself still carries `entry.source_shard`: +/// the receiver advances `applied_seqno` PER SOURCE shard. pub(super) fn redeliver_missed(transport: &dyn Transport, db: &TidalDb, log: &[BatchEntry]) { + let dest = transport.local_shard(); for entry in log { // Read the follower's high-water-mark per source shard. The // `seqno > applied` gate enforces both invariants documented above: @@ -153,7 +167,18 @@ pub(super) fn redeliver_missed(transport: &dyn Transport, db: &TidalDb, log: &[B if entry.seqno > applied { let payload = single_event_payload(entry.source_shard, entry.seqno, entry.bytes.clone()); - let _ = transport.send_segment(entry.source_shard, payload); + // Mirror the eager path's observability contract: recovery is + // best-effort, but a swallowed ship error leaves an operator + // staring at a follower that never catches up. WARN, don't drop. + if let Err(e) = transport.send_segment(dest, payload) { + tracing::warn!( + dest_shard = dest.0, + source_shard = entry.source_shard.0, + seqno = entry.seqno, + error = %e, + "re-delivery ship failed; will retry on next heal/convergence pass" + ); + } } } } @@ -186,4 +211,87 @@ mod tests { assert_eq!(payload.id.seqno, seqno); assert_eq!(payload.bytes, bytes); } + + /// A transport that records every `(dest, payload-source)` pair. Unlike + /// [`ChannelTransport`] (which IGNORES the destination — the reason the + /// in-process suite cannot catch a wrong-destination regression), this + /// mock asserts on it. + struct RecordingTransport { + local: ShardId, + sent: std::sync::Mutex>, // (dest, payload source, seqno) + } + + impl Transport for RecordingTransport { + fn send_segment( + &self, + to: ShardId, + payload: WalSegmentPayload, + ) -> Result<(), TransportError> { + self.sent + .lock() + .unwrap() + .push((to, payload.id.shard_id, payload.id.seqno)); + Ok(()) + } + fn recv_segment(&self) -> Option { + None + } + fn local_shard(&self) -> ShardId { + self.local + } + } + + /// Re-delivery must ship to the FOLLOWER'S OWN shard (`local_shard`) — + /// never to `entry.source_shard` (the leader). A follower's gRPC + /// transport only knows itself as a peer (self-loop wiring), so the wrong + /// destination fails the peer lookup and partition recovery silently + /// never happens (the m8p9 heal bug). The payload, by contrast, must keep + /// the LEADER's shard: the receiver advances `applied_seqno` per source. + #[test] + fn redeliver_ships_to_local_shard_with_source_payload() { + let leader_shard = ShardId(0); + let follower_shard = ShardId(2); + let transport = RecordingTransport { + local: follower_shard, + sent: std::sync::Mutex::new(Vec::new()), + }; + let db = crate::TidalDb::builder() + .ephemeral() + .open() + .expect("open ephemeral db"); + let log = vec![ + BatchEntry { + source_shard: leader_shard, + seqno: 1, + bytes: vec![1], + }, + BatchEntry { + source_shard: leader_shard, + seqno: 2, + bytes: vec![2], + }, + ]; + + redeliver_missed(&transport, &db, &log); + + let sent = transport.sent.lock().unwrap(); + assert_eq!(sent.len(), 2, "both unapplied entries must re-ship"); + for (dest, payload_source, _seqno) in sent.iter() { + assert_eq!( + *dest, follower_shard, + "re-delivery destination must be the follower's own shard \ + (transport.local_shard()), not the leader's" + ); + assert_eq!( + *payload_source, leader_shard, + "payload must carry the SOURCE shard so the receiver advances \ + applied_seqno for the right origin" + ); + } + assert_eq!( + (sent[0].2, sent[1].2), + (1, 2), + "per-source FIFO order must be preserved" + ); + } }