From 9bf2b20d3f410bd107c94fd73ed6c594920c3c9f Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 31 Aug 2026 12:58:34 +0200 Subject: [PATCH 1/4] fix(shard): keep a parked prepare's stamp and op order on re-dispatch --- core/integration/tests/cluster/mod.rs | 1 + .../tests/cluster/parked_frame_redispatch.rs | 348 +++++++++++++++++ core/server/src/partition_reconciler.rs | 350 ++++++++++------- core/shard/src/lib.rs | 368 +++++++++--------- core/shard/src/metrics.rs | 9 +- core/shard/src/router.rs | 15 + 6 files changed, 755 insertions(+), 336 deletions(-) create mode 100644 core/integration/tests/cluster/parked_frame_redispatch.rs diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs index 940e1fcee8..b794264a0a 100644 --- a/core/integration/tests/cluster/mod.rs +++ b/core/integration/tests/cluster/mod.rs @@ -25,6 +25,7 @@ mod fast_primary_rejoin; mod metadata_checkpoint_restart; mod metadata_state_transfer; mod multi_shard_partition_convergence; +mod parked_frame_redispatch; mod partition_primary_routing; mod partition_state_transfer; mod register_forwarding; diff --git a/core/integration/tests/cluster/parked_frame_redispatch.rs b/core/integration/tests/cluster/parked_frame_redispatch.rs new file mode 100644 index 0000000000..f5939a92dc --- /dev/null +++ b/core/integration/tests/cluster/parked_frame_redispatch.rs @@ -0,0 +1,348 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! A BACKUP parking replicated partition prepares, and the frames it parked +//! reaching the plane in op order once its partition materialises. +//! +//! `multi_shard_partition_convergence` covers the same fence on one node and +//! says outright that it cannot tell a request served straight through from one +//! that parked. This test pins the park path positively, and on the replica +//! where getting it wrong is unrecoverable: 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 the partition plane has no +//! normal-status repair driver to refetch one it dropped. +//! +//! 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 +//! plane or from the primary's `CommitMessage` heartbeat, whose interval is +//! `cluster.commit_broadcast_interval` (500ms by default). Nothing in +//! `create_topic`'s reply path waits for that, so a produce issued the instant +//! `create_topic` returns reaches the backups as a partition prepare for a +//! namespace they have not yet heard of, let alone built. Four producers on +//! their own connections keep a burst in flight across that gap, so several ops +//! of one partition park together and the order they leave in is observable. +//! +//! Three things are asserted, and they fail separately: +//! +//! - The path was entered on a backup. `redispatch_parked_frames` logs at +//! `debug`, hence the `system.logging.level` override; the marker on a node +//! that is not the leader is proof, because a fresh partition group seeds its +//! view from the metadata plane, so every partition primary here is the +//! metadata leader and no client request lands anywhere else. +//! - No park path degraded into a shed, an aged-out answer, or an incarnation +//! rejection, and no replica dropped a prepare for arriving out of order. +//! That last marker is the direct symptom of re-dispatch losing a frame's +//! arrival position. +//! - 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. +//! +//! `RUST_LOG` in the test process environment overrides the config level and +//! would take the positive marker with it; the assertion says so when it fires. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::str::FromStr; +use std::time::Duration; + +use futures::future::join_all; +use iggy::prelude::*; +use integration::harness::{TestHarness, disk}; +use integration::iggy_harness; +use tokio::time::{Instant, sleep}; + +const STREAM: &str = "parked-redispatch-stream"; +/// Each topic is one shot at the race, and each costs about one commit +/// broadcast interval. +const TOPICS: u32 = 6; +const PARTITIONS: u32 = 4; +/// The reconciler builds a topic's namespaces in id order, so the last one has +/// the longest wait for its `InsertOwned`. +const TARGET_PARTITION: u32 = PARTITIONS - 1; +/// Separate connections, because one `IggyClient` serialises its requests and a +/// single in-flight prepare would never expose park ordering. +const PRODUCERS: usize = 4; +const PER_PRODUCER: usize = 8; +const TOTAL_MESSAGES: usize = PRODUCERS * PER_PRODUCER; + +/// Budget for eagerly flushed batches to reach every node's segment files. +const FLUSH_INSTALL_TIMEOUT: Duration = Duration::from_secs(20); +const POLL_INTERVAL: Duration = Duration::from_millis(250); + +/// `IggyShard::redispatch_parked_frames`, at `debug`. +const REDISPATCH_MARKER: &str = "re-dispatching parked partition frames after materialisation"; + +/// Modes in which the park path gives up instead of converging. All three cost +/// a frame: the first two shed or answer, the third refuses a namespace whose +/// incarnation moved under it. +const DEGRADED_MARKERS: [&str; 3] = [ + "park buffer at capacity", + "outlived their admission window", + "rejecting parked partition frame", +]; + +/// `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 +/// is gone: nothing refetches it while the replica's status is normal. +const GAP_MARKER: &str = "dropping out-of-order prepare (gap)"; + +fn topic_name(index: u32) -> String { + format!("parked-redispatch-topic-{index}") +} + +#[iggy_harness(cluster_nodes = 3, server(system.logging.level = "info,shard=debug"))] +async fn given_a_produce_burst_right_after_create_topic_when_backups_park_the_prepares_should_re_dispatch_them_in_order( + harness: &mut TestHarness, +) { + // Read once, before any topic exists: the leader is the primary of every + // partition group created below, so it is also the only node a produce can + // be admitted on. + let leader = disk::leader_node_index_via(harness, 0).await; + let setup = harness + .root_client_for_node(leader) + .await + .expect("root client on the metadata leader"); + setup.create_stream(STREAM).await.expect("create stream"); + let stream = Identifier::named(STREAM).expect("stream identifier"); + + // Connected and logged in before the first `create_topic`, so the burst + // costs one round trip rather than a handshake. + let mut producers = Vec::with_capacity(PRODUCERS); + for _ in 0..PRODUCERS { + producers.push( + harness + .root_client_for_node(leader) + .await + .expect("root client for a producer"), + ); + } + + let mut all_payloads = Vec::with_capacity(TOPICS as usize * TOTAL_MESSAGES); + for topic_index in 0..TOPICS { + let name = topic_name(topic_index); + create_topic(&setup, &stream, &name).await; + let topic = Identifier::named(&name).expect("topic identifier"); + + let sent = produce_burst(&producers, &stream, &topic, topic_index).await; + let polled = poll_payloads(&setup, &stream, &topic).await; + assert_eq!( + polled.len(), + TOTAL_MESSAGES, + "topic {topic_index} must serve every acked message, got {polled:?}" + ); + assert_producer_order(&polled, &sent, topic_index); + all_payloads.extend(polled); + } + + let data_paths: Vec = harness + .all_servers() + .iter() + .map(|server| server.data_path()) + .collect(); + wait_until_payloads_installed(harness, &all_payloads).await; + disk::wait_for_log_convergence(&data_paths).await; + // Also flushes each node's non-blocking log appender, so the markers below + // are read off a complete file. + harness + .stop() + .await + .expect("stop the cluster for the at-rest comparison"); + + assert_backup_re_dispatched(harness, leader); + assert_no_degraded_park_paths(harness); + disk::assert_replica_data_identical(&data_paths, false); +} + +/// `messages_required_to_save` + `enforce_fsync` persist every committed batch +/// on every replica, which is what makes the on-disk assertions mean anything +/// on a run this small; the default thresholds would ack from RAM alone. +async fn create_topic(client: &IggyClient, stream: &Identifier, name: &str) { + client + .create_topic( + stream, + name, + &TopicCreateOptions { + partitions_count: Some(PARTITIONS), + message_expiry: Some(IggyExpiry::NeverExpire), + messages_required_to_save: Some(1), + enforce_fsync: Some(true), + ..TopicCreateOptions::default() + }, + ) + .await + .unwrap_or_else(|error| panic!("create_topic {name}: {error}")); +} + +/// Fire every producer at once, returning each one's payloads in the order it +/// sent them. +async fn produce_burst( + producers: &[IggyClient], + stream: &Identifier, + topic: &Identifier, + topic_index: u32, +) -> Vec> { + let partitioning = Partitioning::partition_id(TARGET_PARTITION); + let sends = producers.iter().enumerate().map(|(producer, client)| { + let partitioning = &partitioning; + async move { + let mut sent = Vec::with_capacity(PER_PRODUCER); + for sequence in 0..PER_PRODUCER { + let payload = format!("t{topic_index}-p{producer}-{sequence}"); + let mut messages = vec![IggyMessage::from_str(&payload).expect("build message")]; + client + .send_messages(stream, topic, partitioning, &mut messages) + .await + .unwrap_or_else(|error| panic!("send_messages {payload}: {error}")); + sent.push(payload); + } + sent + } + }); + join_all(sends).await +} + +/// Payloads of the target partition in offset order, asserting the offsets are +/// dense on the way out: a hole would mean an acked op the leader itself cannot +/// serve. +async fn poll_payloads( + client: &IggyClient, + stream: &Identifier, + topic: &Identifier, +) -> Vec { + let polled = client + .poll_messages( + stream, + topic, + Some(TARGET_PARTITION), + &Consumer::default(), + &PollingStrategy::offset(0), + TOTAL_MESSAGES as u32, + false, + ) + .await + .unwrap_or_else(|error| panic!("poll_messages: {error}")); + for (expected, message) in polled.messages.iter().enumerate() { + assert_eq!( + message.header.offset, expected as u64, + "offsets must be dense from 0, got {} at position {expected}", + message.header.offset + ); + } + polled + .messages + .iter() + .map(|message| String::from_utf8_lossy(&message.payload).into_owned()) + .collect() +} + +/// Each producer's sends must appear in the order it made them. Nothing pins +/// the interleaving of four connections, but a partition that reordered one +/// producer's own ops reordered the log. +fn assert_producer_order(polled: &[String], sent: &[Vec], topic_index: u32) { + let positions: HashMap<&str, usize> = polled + .iter() + .enumerate() + .map(|(position, payload)| (payload.as_str(), position)) + .collect(); + for (producer, payloads) in sent.iter().enumerate() { + let mut previous: Option<(&str, usize)> = None; + for payload in payloads { + let position = *positions.get(payload.as_str()).unwrap_or_else(|| { + panic!("topic {topic_index}: {payload} was acked but never polled back") + }); + if let Some((earlier, earlier_position)) = previous { + assert!( + earlier_position < position, + "topic {topic_index}: producer {producer} sent {earlier} before {payload}, \ + but they polled back at {earlier_position} and {position}" + ); + } + previous = Some((payload.as_str(), position)); + } + } +} + +/// Poll until every node's segments hold every payload at a non-decreasing +/// position. A backup that lost a prepare to the gap check never gets it back, +/// so this is where the loss surfaces first, naming the node. +async fn wait_until_payloads_installed(harness: &TestHarness, payloads: &[String]) { + let deadline = Instant::now() + FLUSH_INSTALL_TIMEOUT; + loop { + let pending: Vec = (0..harness.cluster_size()) + .filter_map(|node| { + disk::installed_payloads_complete(&harness.node(node).data_path(), payloads) + .err() + .map(|error| format!("node {node}: {error}")) + }) + .collect(); + if pending.is_empty() { + return; + } + assert!( + Instant::now() < deadline, + "every acked payload must reach every replica's segments within \ + {FLUSH_INSTALL_TIMEOUT:?}: {pending:?}" + ); + sleep(POLL_INTERVAL).await; + } +} + +/// The point of the test. A non-leader node logging the re-dispatch is a backup +/// that parked REPLICATED prepares: client requests only ever reach the leader, +/// which is the primary of every partition group created here. +fn assert_backup_re_dispatched(harness: &TestHarness, leader: usize) { + let counts: Vec<(usize, usize)> = (0..harness.cluster_size()) + .map(|node| { + ( + node, + harness.node(node).stdout_occurrences(REDISPATCH_MARKER), + ) + }) + .collect(); + let on_backups: usize = counts + .iter() + .filter(|(node, _)| *node != leader) + .map(|(_, count)| *count) + .sum(); + assert!( + on_backups > 0, + "no backup logged {REDISPATCH_MARKER:?} (leader is node {leader}, per-node counts \ + {counts:?}); either the produce never raced materialisation, in which case this test \ + proves nothing, or `RUST_LOG` in the environment overrode the debug level the marker \ + needs" + ); +} + +fn assert_no_degraded_park_paths(harness: &TestHarness) { + for node in 0..harness.cluster_size() { + let server = harness.node(node); + for marker in DEGRADED_MARKERS { + assert_eq!( + server.stdout_occurrences(marker), + 0, + "node {node} logged {marker:?}: the park buffer degraded instead of converging" + ); + } + assert_eq!( + server.stdout_occurrences(GAP_MARKER), + 0, + "node {node} logged {GAP_MARKER:?}: a re-dispatched prepare lost its arrival \ + position, and the partition plane has no normal-status repair driver to refetch it" + ); + } +} diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index e9114dabdb..b846a1fbdc 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -40,20 +40,19 @@ //! "unroutable", and falls back to `calculate_shard_assignment`. The frame //! always reaches the shard that will own the partition. //! - `IggyShard::park_if_unmaterialised` holds it there until the matching -//! `InsertOwned` lands, then re-queues it onto this shard's inbox -- but not to -//! a DIFFERENT incarnation than the one it was addressed to. Each parked frame +//! `InsertOwned` lands, then hands it back to the pump -- but not to a +//! DIFFERENT incarnation than the one it was addressed to. Each parked frame //! carries the committed `created_revision` observed when it was parked, and a //! drain whose epoch disagrees with that stamp answers the client instead of //! serving it: recycled slab keys make the namespace byte-identical, so such a //! frame would otherwise land a dead topic's write inside the topic that -//! replaced it. One gap, recorded below: the stamp is re-derived if the frame -//! re-enters the park path from the inbox. A frame parked with NO stamp is -//! served; see `redispatch_parked_frames` for why a missing committed revision -//! is not evidence of a prior incarnation. Re-queuing appends, so a parked -//! frame is ordered behind whatever is already in the inbox. A frame the inbox -//! refuses is re-parked rather than answered, since the deny would ride the -//! same full sender, and the pump re-drives it (`retry_reparked_frames`) once -//! a slot frees. +//! replaced it. The stamp survives a second park, so a re-delivered frame +//! cannot be re-stamped with the incarnation that replaced its own. A frame +//! parked with NO stamp is served; see `redispatch_parked_frames` for why a +//! missing committed revision is not evidence of a prior incarnation. Delivery +//! is the pump's: `drain_redispatched_frames` runs before it reads the inbox +//! again, so a parked op is not ordered behind a later op of the same +//! partition already queued there, which the plane's gap check would drop. //! - `IggyShard::serves_committed_incarnation` refuses a namespace whose //! committed `created_revision` disagrees with the epoch on the local row, so //! a request arriving mid-teardown cannot be acked against the incarnation @@ -124,26 +123,6 @@ //! shrinks the exposure to two cases, a genuinely exhausted byte budget and a //! namespace this shard cannot serve, but only the repair driver removes it. //! -//! TODO(krishna): the park stamp is not stable across re-entry. A re-dispatched -//! frame still in the inbox when a delete + recreate completes (`ConfirmRemove` -//! removes and untombstones in one arm, then the rebuild lands) re-enters -//! `park_if_unmaterialised` and is re-stamped with the NEW revision and -//! `passes: 0`, then served against the replacement: the write the stamp exists -//! to block. Narrow (a full delete + recreate has to finish while one frame -//! waits), but the guarantee is not absolute the way the bullet above reads. -//! Closing it needs the frame to carry provenance through the inbox instead of -//! re-deriving it on arrival. -//! -//! TODO(krishna): re-dispatch APPENDS to the inbox, so a parked prepare loses its -//! arrival position. `router.rs`'s `select_biased!` puts the consensus tick (which -//! runs `apply_reconcile_ops`, and with it the re-dispatch) above the inbox arm, -//! so a parked op N is re-queued *behind* an op N+1 that was already sitting in -//! the inbox. The partition plane then sees N+1 first, rejects it against its -//! backup gap check, and N+1 is gone -- with no normal-status repair driver to -//! refetch it (see the TODO above). Ordering has to be restored at the plane, by -//! buffering out-of-order prepares rather than dropping them, or by re-dispatching -//! through a priority path that preserves op order. -//! //! TODO(krishna): `serves_committed_incarnation` and the park stamp both call //! `Streams::created_revision_for_namespace`, now on the per-request fence path. //! It indexes directly and falls back to a scan only if partition ids are not @@ -803,12 +782,11 @@ async fn reconcile_additions( /// snapshotted before `reconcile_additions` awaits `build_partition_fresh`, so a /// topic committing during those awaits is judged against a stale set. /// -/// Everything else is aged: building, backed off, still committing, genuinely -/// deleted, or materialised with frames the inbox refused. -/// [`shard::IggyShard::age_parked_partition_frames`] answers CLIENT REQUESTS past -/// `MAX_PARKED_PASSES` and leaves prepares alone, so no client waits out its read -/// timeout and no committed op dies on a local-convergence signal. Residency -/// only; see `ParkedFrame::passes`. +/// Everything else is aged: building, backed off, still committing, or +/// genuinely deleted. [`shard::IggyShard::age_parked_partition_frames`] answers +/// CLIENT REQUESTS past `MAX_PARKED_PASSES` and leaves prepares alone, so no +/// client waits out its read timeout and no committed op dies on a +/// local-convergence signal. Residency only; see `ParkedFrame::passes`. /// /// A namespace with a staged, unapplied `InsertOwned` is exempt: its partition /// is on the way but reads as un-materialised here. The queue is asked per @@ -845,11 +823,10 @@ fn reconcile_parked_frames(ctx: &ReconcilerCtx, counters: &mut PassCounters) { counters.parked_reclaimed += 1; continue; } - // Materialised with frames still parked means the re-dispatch hit a full - // inbox and re-parked them. The pump retries every iteration, so aging is - // only the backstop for an inbox that never drains. Without it they have - // no exit: `reconcile_additions` stages no second `InsertOwned` for a - // namespace already in `IggyPartitions`. + // Un-materialised and still ours: the build is on the way or backed off, + // so age the requests rather than hold them for the process lifetime. + // Materialisation hands its frames straight to the pump, so a namespace + // in `IggyPartitions` no longer reaches here with any parked. if ctx.shard.age_parked_partition_frames(ns) > 0 { counters.parked_reclaimed += 1; } @@ -1650,19 +1627,20 @@ mod tests { } /// [`build_test_shard`] with a sender mesh, for tests asserting on work - /// handed back to the pump (transient denies, parked-frame re-dispatch). + /// handed back to the pump (transient denies, frames queued on the inbox). /// Caller must keep the returned receiver alive; dropping it turns every /// `try_send` into `Disconnected`. /// /// Mesh covers `0..=shard_id` since consumers index `senders[shard_id]`. /// Peer receivers are dropped, so a misroute fails loudly instead of landing /// in this shard's inbox and reading as success. - /// Both receiving ends of a test shard's own sender-ring slot: parked - /// frames re-dispatch onto the main lane, staged client answers onto the - /// reply lane. + /// A test shard's own sender-ring slot: both receiving ends plus the + /// sending end, so a test can also put a frame on the main lane the way a + /// peer shard would. struct TestLanes { main: shard::Receiver, reply: shard::Receiver, + main_tx: shard::TaggedSender, } fn build_test_shard_with_inbox( @@ -1675,13 +1653,14 @@ mod tests { let mut own_rx = None; for peer in 0..=shard_id { let (tx, rx, reply_rx) = shard::shard_channel(peer, capacity, capacity); - senders.push(tx); if peer == shard_id { own_rx = Some(TestLanes { main: rx, reply: reply_rx, + main_tx: tx.clone(), }); } + senders.push(tx); } let mut shard = Rc::into_inner(build_test_shard(shard_id, config, mux)) .expect("freshly built shard is uniquely owned"); @@ -1692,8 +1671,8 @@ mod tests { ) } - /// Drain a test shard's lanes into `(re-dispatched frames, staged client - /// sends)`: served parked frames vs answers headed for a client. + /// Drain a test shard's lanes into `(consensus frames, staged client + /// sends)`: work headed for the pump vs answers headed for a client. fn drain_inbox(lanes: &TestLanes) -> (usize, usize) { let mut served = 0; while let Ok(frame) = lanes.main.try_recv() { @@ -1720,6 +1699,22 @@ mod tests { drain_inbox(lanes).1 } + /// Take the prepares off a test shard's main lane in arrival order, as the + /// pump would read them, and return their op numbers. + fn drain_main_lane_prepare_ops(lanes: &TestLanes) -> Vec { + let mut ops = Vec::new(); + while let Ok(frame) = lanes.main.try_recv() { + if let shard::ShardFrame::Consensus { + message: MessageBag::Prepare(prepare), + .. + } = frame + { + ops.push(prepare.header().op); + } + } + ops + } + fn make_ctx( shard: Rc, total_shards: u16, @@ -3533,10 +3528,15 @@ mod tests { 0, "materialisation must drain the park entry" ); + assert_eq!( + shard.redispatched_frame_count(), + 1, + "the unstamped frame must be staged for the pump, not rejected" + ); let (served, answered) = drain_inbox(&inbox); assert_eq!( - served, 1, - "the unstamped frame must be re-dispatched onto the pump, not rejected" + served, 0, + "the queue is the handoff, so nothing may be appended to the inbox" ); assert_eq!( answered, 0, @@ -3646,11 +3646,16 @@ mod tests { ); reconcile_pass(&ctx).await; + assert_eq!( + shard.redispatched_frame_count(), + 1, + "the parked prepare must be staged for re-dispatch; discarding it is \ + silent committed-data loss, since a prepare has no client to answer" + ); let (served, answered) = drain_inbox(&inbox); assert_eq!( - served, 1, - "the parked prepare must be re-dispatched; discarding it is silent \ - committed-data loss, since a prepare has no client to answer" + served, 0, + "the queue is the handoff, so nothing may be appended to the inbox" ); assert_eq!(answered, 0, "a prepare has no client deny to send"); assert_eq!( @@ -3703,6 +3708,149 @@ mod tests { ); } + /// The stamp has to survive re-delivery, not just the first drain. A + /// materialisation stages the frame, the delete half of a recreate then + /// leaves the namespace un-materialised, and the drain parks it a second + /// time: deriving a fresh stamp there picks up the replacement's committed + /// revision, and the rebuild would serve a dead incarnation's op into the + /// topic that recycled its slab keys. + #[compio::test] + async fn given_a_staged_frame_when_a_recreate_lands_before_the_drain_should_reject_it_as_stale() + { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-restamp"); + seed_topic(&mux, 2, 0, "topic-restamp-first", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + // Parked against the FIRST incarnation, so it carries that revision. + park_one_prepare(&shard, ns, 7).await; + assert_eq!(shard.parked_frame_count(ns), 1); + + // The matching incarnation materialises, so the frame stages. Nothing + // has drained it yet: that is the pump's next step. + reconcile_pass(&ctx).await; + assert_eq!( + shard.redispatched_frame_count(), + 1, + "a stamp matching the materialised epoch must stage for the pump" + ); + + // Delete + recreate the same tuple. The teardown pass removes and + // untombstones, which is the state that makes the drain below re-park. + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-restamp-second", + vec![assignment(0, 2)], + ); + reconcile_pass(&ctx).await; + assert!( + !shard.plane.partitions().contains(&ns), + "the teardown pass must have dropped the first incarnation" + ); + assert!( + !shard.plane.partitions().is_tombstoned(&ns), + "and lifted the tombstone, or the drain takes the tombstone path" + ); + + shard.drain_redispatched_frames().await; + assert_eq!( + shard.parked_frame_count(ns), + 1, + "an un-materialised namespace must park the re-delivered frame again" + ); + assert_eq!(shard.redispatched_frame_count(), 0); + + // The rebuild lands at the recreate's revision. + reconcile_pass(&ctx).await; + assert!( + shard.plane.partitions().contains(&ns), + "the replacement must materialise" + ); + assert_eq!( + shard.metrics().partition_frames_rejected_stale_value(), + 1, + "the re-parked frame must keep the stamp it first parked with, so the \ + replacement rejects it instead of serving a dead incarnation's op" + ); + assert_eq!( + shard.redispatched_frame_count(), + 0, + "and it must not be staged for the replacement" + ); + assert_eq!(drain_inbox(&inbox).0, 0, "nothing may reach the pump"); + } + + /// Re-dispatch must not append to the shard's own inbox. `select_biased!` + /// ranks the consensus tick, which is where materialisation runs, above the + /// inbox arm, so a later op of the same partition can already be queued + /// there. Appended, the parked op lands behind it and the plane's backup gap + /// check drops the later one for not being `current_op + 1`, with nothing to + /// refetch it. + /// + /// The pump's order is the queue, then the inbox, so the assertions below + /// are on where each op sits at that moment. Whether the plane accepts them + /// is out of reach here: a solo primary never runs the gap check, and a + /// synthetic prepare cannot be applied. + #[compio::test] + async fn given_a_later_op_on_the_inbox_when_the_namespace_materialises_should_hand_back_the_parked_op_first() + { + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-order"); + seed_topic(&mux, 2, 0, "topic-order", vec![assignment(0, 1)]); + + let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 8); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let ns = IggyNamespace::new(0, 0, 0); + + park_one_prepare(&shard, ns, 5).await; + assert_eq!(shard.parked_frame_count(ns), 1); + + // Op 6 reaches the inbox while op 5 is still parked, as it does whenever + // the primary keeps replicating through the convergence window. + inbox + .main_tx + .try_send(shard::ShardFrame::consensus( + 0, + build_partition_prepare(ns, 6), + )) + .expect("capacity for one queued prepare"); + + reconcile_pass(&ctx).await; + + assert_eq!( + shard.parked_frame_count(ns), + 0, + "materialisation must drain the park entry" + ); + assert_eq!( + drain_main_lane_prepare_ops(&inbox), + vec![6], + "op 5 must not be appended behind op 6; the plane would then see 6 \ + first and drop it as a gap" + ); + assert_eq!( + shard.redispatched_frame_count(), + 1, + "op 5 must be handed back through the queue the pump drains before it \ + reads the inbox" + ); + assert_eq!( + park_dropped_count(&shard), + 0, + "and neither op may be counted as a drop" + ); + } + /// Parking does not bump `Streams::revision` and does not wake the reconciler, /// so a frame that parks in a converged steady state would be held for the /// process lifetime if the revision fast-skip could still fire. A non-empty @@ -4319,62 +4467,9 @@ mod tests { assert_eq!(park_overflow_count(&shard), 2); } - /// A refused re-dispatch re-parks the frame, and by then the namespace is - /// materialised, closing every other exit: the sweep skips a namespace in - /// `IggyPartitions` and `reconcile_additions` stages no second - /// `InsertOwned`. Before the pump retry the frame sat until a topic delete, - /// unanswered, with its bytes charged and the fast-skip never re-arming. - #[compio::test] - async fn a_re_parked_frame_is_re_dispatched_once_the_inbox_drains() { - let tmp = TempDir::new().expect("tempdir for system path"); - let config = test_config(&tmp); - let mux = TestMux::default(); - seed_stream(&mux, 1, "stream-repark"); - seed_topic(&mux, 2, 0, "topic-repark", vec![assignment(0, 1)]); - - // Capacity 1: `enqueue_reconcile_op`'s `ReconcileApply` marker takes the - // only slot, so the re-dispatch that follows is refused with `Full`. - let (shard, inbox) = build_test_shard_with_inbox(0, &config, mux, 1); - let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); - let ns = IggyNamespace::new(0, 0, 0); - - park_one_request(&shard, ns).await; - reconcile_pass(&ctx).await; - - assert!( - shard.plane.partitions().contains(&ns), - "the namespace must have materialised" - ); - assert_eq!( - shard.parked_frame_count(ns), - 1, - "the full inbox must have re-parked the frame rather than dropping it" - ); - - // What the pump does every iteration: consume a frame, then re-drive. - let (_served, _answered) = drain_inbox(&inbox); - shard.apply_reconcile_ops(); - - assert_eq!( - shard.parked_frame_count(ns), - 0, - "the freed slot must let the retry drain the entry" - ); - assert!( - !shard.has_parked_partition_frames(), - "and the byte budget must return, so the revision fast-skip can re-arm" - ); - assert_eq!( - drain_inbox(&inbox).0, - 1, - "the frame must reach the pump as a consensus frame, not be answered away" - ); - } - - /// A namespace mid-teardown is still in `IggyPartitions`, so it reads as - /// materialised while the fence forbids serving it. `ConfirmRemove` would - /// answer its frames, but a disk delete that keeps failing never enqueues - /// one, so the sweep has to. + /// A frame parks while the namespace is un-materialised, then teardown + /// fences it. `ConfirmRemove` would answer the frame, but a disk delete that + /// keeps failing never enqueues one, so the sweep has to. #[compio::test] async fn parked_frames_of_a_tombstoned_namespace_are_reclaimed_without_confirm_remove() { let tmp = TempDir::new().expect("tempdir for system path"); @@ -4383,20 +4478,20 @@ mod tests { seed_stream(&mux, 1, "stream-tombstone-park"); seed_topic(&mux, 2, 0, "topic-tombstone-park", vec![assignment(0, 1)]); - let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 1); + let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 8); let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); let ns = IggyNamespace::new(0, 0, 0); park_one_request(&shard, ns).await; - reconcile_pass(&ctx).await; assert_eq!( shard.parked_frame_count(ns), 1, - "the full inbox must have re-parked the frame" + "the request must park while the namespace is un-materialised" ); // Teardown's synchronous fence, without the `ConfirmRemove` a wedged - // disk delete never reaches. + // disk delete never reaches. It also stops the pass below from building + // the namespace, which is what would otherwise drain the entry. shard.plane.partitions().tombstone(ns); shard.shards_table().remove(&ns); @@ -4412,35 +4507,6 @@ mod tests { ); } - /// Residency backstop: an inbox that never drains must not hold a re-parked - /// frame forever, so `MAX_PARKED_PASSES` covers a materialised namespace too. - #[compio::test] - async fn a_re_parked_frame_ages_out_when_the_inbox_never_drains() { - let tmp = TempDir::new().expect("tempdir for system path"); - let config = test_config(&tmp); - let mux = TestMux::default(); - seed_stream(&mux, 1, "stream-repark-age"); - seed_topic(&mux, 2, 0, "topic-repark-age", vec![assignment(0, 1)]); - - let (shard, _inbox) = build_test_shard_with_inbox(0, &config, mux, 1); - let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); - let ns = IggyNamespace::new(0, 0, 0); - - park_one_request(&shard, ns).await; - reconcile_pass(&ctx).await; - assert_eq!(shard.parked_frame_count(ns), 1, "re-parked on a full inbox"); - - for _ in 0..=PARK_MAX_PASSES { - reconcile_pass(&ctx).await; - } - assert_eq!( - shard.parked_frame_count(ns), - 0, - "a materialised namespace must still be aged, or the frame is stranded" - ); - assert!(!shard.has_parked_partition_frames()); - } - /// Mirrors `MAX_PARKED_PER_NAMESPACE` in `shard::park_if_unmaterialised`. const PARK_CAP: usize = 128; /// Mirrors `MAX_PARKED_BYTES`. diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 2decb39341..96bb5de003 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -37,7 +37,6 @@ use consensus::{ }; #[cfg(any(test, feature = "simulator"))] use crossfire::AsyncRxTrait; -use crossfire::TrySendError; use futures::FutureExt; use iggy_binary_protocol::{ CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusHeader, DoViewChangeHeader, @@ -70,7 +69,7 @@ use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen}; use shards_table::ShardsTable; use std::cell::{Cell, RefCell}; -use std::collections::{BTreeMap, BTreeSet, HashMap, VecDeque}; +use std::collections::{BTreeMap, HashMap, VecDeque}; use std::future::Future; use std::rc::Rc; #[cfg(feature = "simulator")] @@ -1395,18 +1394,17 @@ where /// admission, on the reactor thread inside the map's `borrow_mut`. parked_partition_bytes: Cell, - /// Namespaces holding frames [`Self::redispatch_parked_frames`] could not - /// re-queue, for the pump to retry. + /// Frames [`Self::redispatch_parked_frames`] handed back for the pump to + /// deliver, in park order. /// - /// Without it a re-parked frame has no exit: its namespace is materialised - /// by then, so the sweep skips it and `reconcile_additions` stages no second - /// `InsertOwned`. Only a topic delete would reach it. The pump drains the - /// inbox, so a refusal usually clears on its next iteration. - /// - /// [`BTreeSet`] for the reason [`Self::pending_partition_frames`] is a - /// [`BTreeMap`]: fixed-seed simulator replay needs iteration order to be a - /// function of the namespaces alone. - reparked_partition_namespaces: RefCell>, + /// Staging exists because re-dispatch runs inside the synchronous + /// [`Self::apply_reconcile_ops`] while the plane is reachable only through + /// an async path. The pump drains this before its next `inbox.recv()`, so a + /// parked op reaches the plane ahead of a later op already sitting on the + /// inbox (the plane's backup gap check drops anything that is not + /// `current_op + 1`), and each frame keeps the provenance it was parked + /// with instead of having it re-derived against newer committed state. + redispatch_queue: RefCell>, /// Set while the shard-wide budget is shedding for namespaces holding no /// park entry of their own, which have no [`ParkEntry::shed`] to warn once @@ -1589,7 +1587,7 @@ where reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(BTreeMap::new()), parked_partition_bytes: Cell::new(0), - reparked_partition_namespaces: RefCell::new(BTreeSet::new()), + redispatch_queue: RefCell::new(VecDeque::new()), shard_park_shedding: Cell::new(false), metadata_repair: RefCell::new(None), metadata_transfer: RefCell::new(None), @@ -1919,7 +1917,7 @@ where reconcile_queue: RefCell::new(VecDeque::new()), pending_partition_frames: RefCell::new(BTreeMap::new()), parked_partition_bytes: Cell::new(0), - reparked_partition_namespaces: RefCell::new(BTreeSet::new()), + redispatch_queue: RefCell::new(VecDeque::new()), shard_park_shedding: Cell::new(false), metadata_repair: RefCell::new(None), metadata_transfer: RefCell::new(None), @@ -2109,35 +2107,6 @@ where })); } - /// Re-drive the re-dispatch for namespaces whose frames the inbox refused. - /// - /// Runs on the pump, wherever [`Self::apply_reconcile_ops`] does, so it - /// fires right after a frame was consumed and a slot freed. Only another - /// refusal puts a namespace back, so the set empties itself. - /// - /// Epoch comes from the routing row, which `InsertOwned` writes alongside - /// the partition. Skipped when the row is gone or the namespace is fenced; - /// teardown does both, and the reconciler sweep retires the frames. - fn retry_reparked_frames(&self) { - let pending: Vec = { - let mut reparked = self.reparked_partition_namespaces.borrow_mut(); - if reparked.is_empty() { - return; - } - std::mem::take(&mut *reparked).into_iter().collect() - }; - let partitions = self.plane.partitions(); - for namespace in pending { - if partitions.is_tombstoned(&namespace) { - continue; - } - let Some(epoch) = self.shards_table.epoch_for(namespace) else { - continue; - }; - self.redispatch_parked_frames(namespace, epoch); - } - } - /// Drain and apply staged [`ReconcileOp`]s on the pump task. /// Synchronous: every arm is in-memory only. `ConfirmRemove`'s fsync + /// blocking close is offloaded to a detached task so the pump doesn't @@ -2146,9 +2115,6 @@ where where B: MessageBus + 'static, { - // Ahead of the staged ops and outside their empty-queue early return: a - // re-parked frame waits on inbox capacity, not on a reconcile op. - self.retry_reparked_frames(); let staged: Vec> = { let mut q = self.reconcile_queue.borrow_mut(); if q.is_empty() { @@ -2430,6 +2396,30 @@ impl ParkedFrame { } } +/// What a frame keeps when the pump re-delivers it and it parks again. +/// +/// Re-deriving the epoch on re-entry re-stamps the frame with whatever +/// incarnation is committed NOW, so a delete + recreate that landed while the +/// frame was in flight gets served the dead incarnation's write: the exact case +/// the stamp exists to block. `None` therefore stays `None` - a frame that +/// parked without a committed revision must not acquire one later, since the +/// drain serves an unstamped frame (see +/// [`IggyShard::redispatch_parked_frames`]). +#[derive(Clone, Copy)] +struct ParkProvenance { + epoch: Option, + passes: u32, +} + +impl From<&ParkedFrame> for ParkProvenance { + fn from(frame: &ParkedFrame) -> Self { + Self { + epoch: frame.epoch, + passes: frame.passes, + } + } +} + /// One namespace's parked frames plus their running footprint. /// /// Carried, not re-summed: `park_if_unmaterialised` reads it per arriving frame @@ -2548,6 +2538,99 @@ where /// once per hop rather than once for routing and again for dispatch. #[allow(clippy::future_not_send)] pub async fn on_message(&self, message: MessageBag) + where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: StateMachine< + Input = Message, + Output = metadata::stm::result::ApplyReply, + Error = iggy_common::IggyError, + > + StreamsFrontend + + metadata::stm::snapshot::RestoreSnapshotInPlace< + metadata::stm::snapshot::MetadataSnapshot, + >, + T: ShardsTable, + { + self.dispatch_message(message, None).await; + } + + /// Deliver what [`Self::redispatch_parked_frames`] staged, in park order. + /// + /// The pump runs this after every `apply_reconcile_ops` and before it reads + /// the inbox again, which is what puts a parked op ahead of a later op of + /// the same partition already queued there. + /// + /// One frame per borrow: the dispatch below can park the frame again, and a + /// guard held across the `.await` would panic when it does. + #[allow(clippy::future_not_send)] + pub async fn drain_redispatched_frames(&self) + where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: StateMachine< + Input = Message, + Output = metadata::stm::result::ApplyReply, + Error = iggy_common::IggyError, + > + StreamsFrontend + + metadata::stm::snapshot::RestoreSnapshotInPlace< + metadata::stm::snapshot::MetadataSnapshot, + >, + T: ShardsTable, + { + loop { + let Some(frame) = self.redispatch_queue.borrow_mut().pop_front() else { + return; + }; + let provenance = ParkProvenance::from(&frame); + // Parked frames are stored generic (the buffer holds every variant + // in one Vec), so re-entering the pump costs one classify. That is + // the rare path -- a post-`CreateTopic` convergence window, not the + // per-message steady state the bag handoff exists for. + let bag = match MessageBag::try_from(frame.message) { + Ok(bag) => bag, + Err(error) => { + // The frame classified once already, on the way in, so this + // is unreachable short of memory corruption. Dropping it + // costs a client retry; panicking on the pump would take the + // shard down. + tracing::error!( + shard = self.id, + %error, + "re-dispatched partition frame no longer classifies; dropping it" + ); + continue; + } + }; + self.dispatch_message(bag, Some(provenance)).await; + } + } + + /// Retire staged frames the pump is no longer going to deliver, on its way + /// out. Client requests get a transient deny; the rest are counted as drops, + /// which is the only record a replicated frame leaves. + pub fn retire_redispatched_frames(&self) { + let staged: Vec = self.redispatch_queue.borrow_mut().drain(..).collect(); + if staged.is_empty() { + return; + } + let (answered, dropped) = self.retire_parked_frames(staged); + tracing::debug!( + shard = self.id, + answered, + dropped, + "retiring re-dispatched partition frames the pump will not deliver" + ); + } + + /// [`Self::on_message`] carrying the park provenance of a frame the pump is + /// re-delivering, so a second park keeps the stamp and age the first one + /// derived instead of deriving them again against newer committed state. + #[allow(clippy::future_not_send)] + async fn dispatch_message(&self, message: MessageBag, provenance: Option) where B: MessageBus + 'static, MJ: JournalHandle, @@ -2570,7 +2653,7 @@ where let header = request.header(); (header.operation, header.group) }; - match self.park_if_unmaterialised(request, routing.0, routing.1) { + match self.park_if_unmaterialised(request, routing.0, routing.1, provenance) { // The incarnation fence runs only here, on client traffic. // A backup denying what the primary admitted would diverge // the replicas, so replicated frames are never fenced. @@ -2601,7 +2684,7 @@ where // A tombstoned prepare still flows to the plane: replicated // traffic has no client awaiting a reply on this node, and // the plane's own tombstone guard drops it. - match self.park_if_unmaterialised(prepare, routing.0, routing.1) { + match self.park_if_unmaterialised(prepare, routing.0, routing.1, provenance) { ParkOutcome::Deliver(prepare) | ParkOutcome::Tombstoned(prepare) => { self.on_replicate(prepare).await; // A follower learns the cluster commit point from the @@ -2806,13 +2889,10 @@ where } } - /// Remove a namespace's entry, debiting [`Self::parked_partition_bytes`] and - /// disarming the pump retry. Single place an entry leaves the map, so - /// neither can drift out of step with it. + /// Remove a namespace's entry, debiting [`Self::parked_partition_bytes`]. + /// Single place an entry leaves the map, so the two cannot drift out of step + /// with each other. fn take_parked_partition_frames(&self, namespace: IggyNamespace) -> Option> { - self.reparked_partition_namespaces - .borrow_mut() - .remove(&namespace); let (entry, converged) = { let mut pending = self.pending_partition_frames.borrow_mut(); let entry = pending.remove(&namespace)?; @@ -2877,9 +2957,8 @@ where .collect() } - /// Re-queue the frames parked for `namespace` now that its partition exists - /// at `epoch`, onto this shard's own inbox so the pump serves them after the - /// current drain. + /// Hand the frames parked for `namespace` back to the pump, in park order, + /// now that its partition exists at `epoch`. /// /// A frame stamped with a DIFFERENT incarnation never makes it back: the /// namespace is byte-identical across incarnations, so serving it would land @@ -2901,25 +2980,16 @@ where /// discriminator (see the `TODO(krishna)` in /// `partition_reconciler`'s module docs), not a `None`-means-stale rule. /// - /// A frame the inbox refuses is re-parked: retained, so not counted as a - /// drop. Re-queuing appends, so a pass materialising many namespaces can - /// overrun the inbox; staging a deny is futile because it rides the same - /// sender with no await in between. One namespace alone can now do it, since - /// [`MAX_PARKED_BYTES_PER_NAMESPACE`] admits 1024 header-only prepares - /// against a default `inbox_capacity` of 1024. That costs other namespaces a - /// later convergence, not a frame. The first `Full` ends the loop, since - /// the sole consumer of `senders[self.id]` is the pump task running this - /// call and no later frame can find a slot the first one could not. + /// Staged onto [`Self::redispatch_queue`] rather than sent: the shard's own + /// inbox can already hold a LATER op of this partition, and the plane's + /// backup gap check drops anything that is not `current_op + 1`, so + /// appending would strand the parked op behind an op that will be dropped + /// for arriving too early. [`Self::drain_redispatched_frames`] runs before + /// the pump reads the inbox again. /// - /// [`MAX_PARKED_PASSES`] does not bound a re-parked frame: the sweep ages a - /// namespace only while un-materialised, and by here it is materialised. - /// [`Self::repark_partition_frames`] arms the pump retry instead; - /// `partition_reconciler::reconcile_parked_frames` is the backstop for an - /// inbox that never drains. - fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64) - where - B: MessageBus + 'static, - { + /// [`MAX_PARKED_PASSES`] does not bound a staged frame: it has left the park + /// buffer, and the pump drains the queue on the iteration that filled it. + fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64) { let Some(frames) = self.take_parked_partition_frames(namespace) else { return; }; @@ -2930,8 +3000,6 @@ where epoch, "re-dispatching parked partition frames after materialisation" ); - // Incarnation filter first, independent of the sender: a prior - // incarnation is rejected whether or not this shard can re-queue. let mut servable: Vec = Vec::with_capacity(frames.len()); for frame in frames { // Only a stamp that exists and disagrees is evidence of a prior @@ -2944,108 +3012,7 @@ where servable.push(frame); } } - let Some(sender) = self.senders.get(self.id as usize) else { - self.retire_parked_frames(servable); - return; - }; - let mut refused_frames: Vec = Vec::new(); - let mut remaining = servable.into_iter(); - while let Some(frame) = remaining.next() { - let passes = frame.passes; - let parked_epoch = frame.epoch; - // Parked frames are stored generic (the buffer holds every variant - // in one Vec), so re-entering the pump costs one classify. That is - // the rare path -- a post-`CreateTopic` convergence window, not the - // per-message steady state the bag handoff exists for. - let bag = match MessageBag::try_from(frame.message) { - Ok(bag) => bag, - Err(error) => { - // The frame classified once already, on the way in, so this - // is unreachable short of memory corruption. Dropping it - // costs a client retry; panicking on the reconciler's path - // would take the shard down. - tracing::error!( - shard = self.id, - namespace_raw = namespace.inner(), - %error, - "parked partition frame no longer classifies; dropping it" - ); - continue; - } - }; - let Err(error) = sender.try_send(ShardFrame::consensus(self.id, bag)) else { - continue; - }; - let (refused, disconnected) = match error { - TrySendError::Full(frame) => (frame, false), - TrySendError::Disconnected(frame) => (frame, true), - }; - let ShardFrame::Consensus { message, .. } = refused else { - unreachable!("try_send returns the frame it was handed"); - }; - let refused_frame = ParkedFrame { - epoch: parked_epoch, - passes, - message: message.into_generic(), - }; - if disconnected { - // Pump gone: re-parking holds the frame until process exit, and - // every later send hits the same dead channel. - self.metrics.record_frame_drop( - crate::metrics::frame_drop_variant::PARTITION, - crate::metrics::frame_drop_reason::DISCONNECTED, - ); - tracing::warn!( - shard = self.id, - namespace_raw = namespace.inner(), - "re-dispatch of parked partition frames refused: inbox disconnected" - ); - let mut stranded = vec![refused_frame]; - stranded.extend(remaining); - self.retire_parked_frames(stranded); - return; - } - refused_frames.push(refused_frame); - refused_frames.extend(remaining); - tracing::debug!( - shard = self.id, - namespace_raw = namespace.inner(), - count = refused_frames.len(), - passes, - "re-parking parked partition frames: inbox full" - ); - break; - } - if !refused_frames.is_empty() { - self.repark_partition_frames(namespace, refused_frames); - } - } - - /// Put frames back under `namespace` after a refused re-dispatch, keeping - /// [`Self::parked_partition_bytes`] in step and arming the pump-side retry. - /// - /// Deliberately not budget-checked: these bytes were already counted while - /// parked, so re-admitting them cannot grow the total past what it held a - /// moment ago, and shedding here would answer a frame the inbox merely - /// deferred. - /// - /// Arming [`Self::reparked_partition_namespaces`] is what makes it a - /// deferral. Every other exit is closed once materialised: the sweep only - /// ages a namespace it has not built, and `reconcile_additions` stages no - /// second `InsertOwned` for one already in `IggyPartitions`. - fn repark_partition_frames(&self, namespace: IggyNamespace, frames: Vec) { - let restored: usize = frames.iter().map(ParkedFrame::footprint).sum(); - let mut pending = self.pending_partition_frames.borrow_mut(); - let entry = pending.entry(namespace).or_default(); - for frame in frames { - entry.push(frame); - } - drop(pending); - self.parked_partition_bytes - .set(self.parked_partition_bytes.get().saturating_add(restored)); - self.reparked_partition_namespaces - .borrow_mut() - .insert(namespace); + self.redispatch_queue.borrow_mut().extend(servable); } /// Age every frame under `namespace` by one pass, answering CLIENT REQUESTS @@ -3073,7 +3040,7 @@ where let emptied = entry.frames.is_empty(); drop(pending); if emptied { - // Through the shared remover so the pump-retry set is disarmed + // Through the shared remover so the shed-episode flag clears // with it; the entry is already empty, so this only unhooks it. self.take_parked_partition_frames(namespace); } @@ -3112,6 +3079,17 @@ where .map_or(0, |entry| entry.frames.len()) } + /// How many frames are staged for the pump to re-deliver. + /// + /// Test/simulator accessor, gated for the same reason as + /// [`Self::parked_frame_count`]: the pump drains the queue on the iteration + /// that filled it, so no production caller has a depth to branch on. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn redispatched_frame_count(&self) -> usize { + self.redispatch_queue.borrow().len() + } + /// Retire a frame that will never be served: a client request gets a /// transient deny, replicated traffic is destroyed. Returns `true` only when /// a reply reached the pump. @@ -3186,16 +3164,21 @@ where /// disk delete) report [`ParkOutcome::Tombstoned`] so the caller can deny /// client requests instead of feeding them to the plane's silent-drop /// guard, while replicated traffic still flows there. Parked frames are - /// re-dispatched by [`Self::apply_reconcile_ops`] once the matching + /// staged for the pump by [`Self::apply_reconcile_ops`] once the matching /// `ReconcileOp::InsertOwned` lands, and only if the epoch stamped here /// still matches (see [`ParkedFrame`]); a full buffer reports /// [`ParkOutcome::Overflow`] so the caller can answer rather than shed /// silently. + /// + /// `provenance` is `None` for a frame arriving off the wire and `Some` for + /// one the pump is re-delivering, which must keep the stamp and the age it + /// parked with (see [`ParkProvenance`]). fn park_if_unmaterialised( &self, message: Message, operation: Operation, namespace_raw: u64, + provenance: Option, ) -> ParkOutcome where H: iggy_binary_protocol::ConsensusHeader, @@ -3217,13 +3200,18 @@ where } // Read the committed revision before taking the borrow below: the frame // is stamped with the incarnation it was addressed to, so a later drain - // can tell it apart from a same-key replacement. - let epoch = self - .plane - .metadata() - .mux_stm - .streams() - .created_revision_for_namespace(namespace); + // can tell it apart from a same-key replacement. A re-delivered frame + // brings its own, since by now the committed revision can describe the + // replacement rather than the incarnation the frame was addressed to. + let ParkProvenance { epoch, passes } = provenance.unwrap_or_else(|| ParkProvenance { + epoch: self + .plane + .metadata() + .mux_stm + .streams() + .created_revision_for_namespace(namespace), + passes: 0, + }); let frame_cost = parked_footprint(message.as_slice().len()); let replicated = message.header().command() != Command::Request; let mut pending = self.pending_partition_frames.borrow_mut(); @@ -3317,7 +3305,7 @@ where ); pending.entry(namespace).or_default().push(ParkedFrame { epoch, - passes: 0, + passes, message: message.into_generic(), }); drop(pending); diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index cd05dba02a..81608ba166 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -27,8 +27,8 @@ //! by plane. //! - `IggyShard::park_if_unmaterialised` - partition frames shed because the //! park buffer is at its frame or byte cap. -//! - `IggyShard::apply_reconcile_ops` - parked frames whose re-dispatch onto -//! this shard's own inbox was refused. +//! - `IggyShard::retire_parked_frames` - parked frames retired with no client +//! to answer. //! //! The counter uses atomic interior mutability, safe to bump from `!Send` //! compio reactor contexts. Each shard owns its own instance, and the server @@ -77,8 +77,9 @@ pub struct FrameDropLabel { /// /// `PARTITION` covers the partition plane: a frame shed because the namespace /// had not materialised and its park buffer was at capacity -/// (`reason=park_overflow`), a re-dispatch the shard's own inbox refused, or a -/// routing send the target inbox refused. A shed client request is answered with +/// (`reason=park_overflow`), a parked frame retired with no client to answer +/// (`reason=park_dropped`), or a routing send the target inbox refused. A shed +/// client request is answered with /// a retriable status, so the client recovers -- but a shed *prepare* is not /// covered by retransmit once its op has reached quorum /// (`consensus::retransmit_targets` skips `ok_quorum_received`, and the diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index c98e585303..f2b28f8c6e 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -346,6 +346,11 @@ where // a quiet shard until the next inbound frame's tail // drain; parked partition frames then never re-dispatch. self.apply_reconcile_ops(); + // Before the next `inbox.recv()`, always: a materialisation + // hands its parked frames back here, and the inbox may + // already hold a LATER op of the same partition, which the + // plane's gap check drops unless the parked one lands first. + self.drain_redispatched_frames().await; consensus_tick.set(rearm_tick()); } frame = self.inbox.recv().fuse() => { @@ -356,6 +361,7 @@ where self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await; // Tail drain catches reconcile ops whose marker was dropped. self.apply_reconcile_ops(); + self.drain_redispatched_frames().await; } // Guaranteed reply-lane service: `select_biased!` // polls the main lane first, so a saturated main @@ -418,6 +424,10 @@ where } } } + // Retired, not delivered: the pump is going away, so a staged frame + // has no later drain to reach the plane through. Runs before the + // reply-lane drain below, which is what carries the denies out. + self.retire_redispatched_frames(); } if fatal.is_none() { while let Ok(frame) = self.reply_inbox.try_recv() { @@ -527,6 +537,10 @@ where async fn process_lifecycle(&self, payload: LifecycleFrame) where B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: RestorableMetadataStm, { match payload { LifecycleFrame::ReplicaInboundSetup { fd, slot } => { @@ -676,6 +690,7 @@ where } LifecycleFrame::ReconcileApply => { self.apply_reconcile_ops(); + self.drain_redispatched_frames().await; } LifecycleFrame::CleanPartition { namespace, From 4bae9d7c47894e152006eb23c2441eab5cd73702 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 31 Aug 2026 16:32:00 +0200 Subject: [PATCH 2/4] fix doctest --- core/shard/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 96bb5de003..40c742cf81 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -2556,7 +2556,7 @@ where self.dispatch_message(message, None).await; } - /// Deliver what [`Self::redispatch_parked_frames`] staged, in park order. + /// Deliver what `redispatch_parked_frames` staged, in park order. /// /// The pump runs this after every `apply_reconcile_ops` and before it reads /// the inbox again, which is what puts a parked op ahead of a later op of From 249d69122b65c4859c454a08c5cfeb4907fcd9ba Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 31 Aug 2026 14:39:15 +0200 Subject: [PATCH 3/4] fix(shard): drive partition repair and the commit walk from the tick --- .../tests/cluster/parked_frame_redispatch.rs | 6 +- core/partitions/src/iggy_partition.rs | 29 + core/server/src/partition_reconciler.rs | 22 +- core/shard/src/lib.rs | 448 ++++++++++++ core/shard/src/metrics.rs | 43 +- core/simulator/src/lib.rs | 643 ++++++++++++++++++ 6 files changed, 1168 insertions(+), 23 deletions(-) diff --git a/core/integration/tests/cluster/parked_frame_redispatch.rs b/core/integration/tests/cluster/parked_frame_redispatch.rs index f5939a92dc..2d2240f153 100644 --- a/core/integration/tests/cluster/parked_frame_redispatch.rs +++ b/core/integration/tests/cluster/parked_frame_redispatch.rs @@ -97,8 +97,10 @@ 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 -/// is gone: nothing refetches it while the replica's status is normal. +/// 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. const GAP_MARKER: &str = "dropping out-of-order prepare (gap)"; fn topic_name(index: u32) -> String { diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 93eaf156f0..4ece3674b0 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -132,6 +132,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, + /// 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 @@ -477,6 +487,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, @@ -998,6 +1010,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. + pub 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`]). @@ -2378,6 +2406,7 @@ where .with_operation(header.operation) .with_op(header.op), ); + self.note_prepare_gap_drop(); return; } } else { diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index b846a1fbdc..93ca117289 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -108,21 +108,23 @@ //! `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. 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: //! -//! TODO(krishna): a shed or discarded *prepare* has no recovery once its op has -//! reached quorum. `consensus::retransmit_targets` skips entries with -//! `ok_quorum_received`, and the partition plane creates a repair session only -//! in `on_start_view` -- `tick_partitions` re-drives an existing session but -//! cannot open one -- so the backup stays behind `commit_max` until an unrelated -//! view change. It needs a normal-status repair driver. The park policy above -//! shrinks the exposure to two cases, a genuinely exhausted byte budget and a -//! namespace this shard cannot serve, but only the repair driver removes it. -//! //! TODO(krishna): `serves_committed_incarnation` and the park stamp both call //! `Streams::created_revision_for_namespace`, now on the per-request fence path. //! It indexes directly and falls back to a scan only if partition ids are not diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 40c742cf81..c1bff0dc07 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -6495,6 +6495,9 @@ where // mid-sweep is seen on the next tick, the same latency a capped arm // already accepts. let mut transfers_inflight: Option = None; + // Repair sessions this sweep has opened, against + // `PARTITION_REPAIR_ARMS_PER_TICK_MAX`. + let mut repair_arms = 0usize; let mut fatal: Option = None; for namespace in namespace_scratch.drain(..) { @@ -6625,6 +6628,72 @@ where .await; } + // Level-triggered gap detector. Every other partition arming site + // is edge-triggered and the edges are starvable: the commit-heartbeat + // backstop needs `CommitOutcome::Advanced`, and a follower has + // already advanced `commit_max` from each prepare header in + // `replicate_preflight` before the gap check dropped the prepare, so + // under produce load the heartbeat lands as `Accepted` and the gap + // wedges until an unrelated view change. Those edges stay the fast + // path; this is the ~1s floor under them. + let (gap_arm, walk_stalled) = { + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let gap_drops = partition.take_prepare_gap_drops(); + if gap_drops > 0 { + self.metrics.record_partition_prepare_gap_drops(gap_drops); + } + let probe = partition_gap_probe(partition); + let gap_arm = drive_partition_gap_debounce( + &probe, + &mut partition.gap_ticks, + repair_retry_ticks, + repair_arms, + ); + (gap_arm, partition_is_walk_stalled(&probe)) + }; + if gap_arm { + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let consensus = partition.consensus(); + let peer = consensus.primary_index(consensus.view()); + let commit_min = consensus.commit_min(); + let commit_max = consensus.commit_max(); + tracing::info!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min, + commit_max, + peer, + "partition gap-stopped past the debounce; arming repair from the primary" + ); + self.maybe_request_partition_repair(partition, peer).await; + repair_arms += 1; + } + + // Undebounced and uncapped, unlike the repair arm: the predicate + // guarantees the walk finds at least the next op (it and + // `collect_committable_from_journal` read the same `header_by_op`), + // so it cannot spin, and the apply is the same local, already-owed + // work the prepare arm runs inline uncapped. + if walk_stalled { + let config = partitions.config(); + let Some(partition) = partitions.get_mut_by_ns(&namespace) else { + continue; + }; + let consensus = partition.consensus(); + tracing::info!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min = consensus.commit_min(), + commit_max = consensus.commit_max(), + "partition commit walk parked over resident committed ops; resuming" + ); + partition.commit_journal(config).await; + } + // Transfer stall retry: descriptor and chunk frames are // fire-and-forget, so a lost one must not wedge the session (and // the rejoin behind it) forever. Budget-bounded: a peer that died @@ -8831,6 +8900,124 @@ fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) -> u64 requested_to_op.min(commit_max.max(head)) } +/// Repair sessions the partition tick sweep will OPEN per pass. Sibling of +/// `IggyShard::PARTITION_TRANSFERS_INFLIGHT_MAX`: one arm is a `RequestPrepares` +/// plus a repair stream the serving peer walks synchronously, and a node-wide +/// gap (a rejoin, a lossy link) makes every group on this shard due in the same +/// tick. Over-cap groups stay due and arm on a later pass. +const PARTITION_REPAIR_ARMS_PER_TICK_MAX: usize = 3; + +/// What already owns a partition's recovery, if anything. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RecoveryOwner { + /// Nothing owns it: the sweep may arm repair. + Nobody, + Repair, + Transfer, + /// A scheduled transfer re-arm counting down. Arming repair over it would + /// defeat its backoff, as `arm_partition_transfer` documents. + TransferRearm, +} + +/// What the tick sweep reads off one partition to decide whether it is +/// gap-stopped. Split out so the guards, the debounce and the per-tick cap are +/// testable without a shard, a bus, or a journal. +#[derive(Debug, Clone, Copy)] +struct GapProbe { + normal: bool, + transferring: bool, + recovery: RecoveryOwner, + commit_min: u64, + commit_max: u64, + /// Whether `commit_min + 1` is resident in the local journal. + next_op_resident: bool, +} + +/// Whether this replica holds committed ops it cannot walk to, because the op +/// one past its commit frontier is missing from its journal. +/// +/// The journal-hole half is not redundant: a follower advances `commit_max` +/// from every prepare header in `replicate_preflight`, so `commit_min < +/// commit_max` is transiently true on every healthy pipelined tick and a bare +/// lag test would arm repair against ordinary produce. +const fn partition_is_gap_stopped(probe: &GapProbe) -> bool { + probe.normal + && !probe.transferring + && matches!(probe.recovery, RecoveryOwner::Nobody) + && probe.commit_min < probe.commit_max + && !probe.next_op_resident +} + +/// The gap predicate's complement: everything the walk needs is resident, it +/// just never ran (a heartbeat carrying a known commit is `Accepted`, and an +/// idle group offers no other edge). Not gated on `RecoveryOwner`: repair +/// fetches bodies without walking them, so gating parks the walk all session. +const fn partition_is_walk_stalled(probe: &GapProbe) -> bool { + probe.normal + && !probe.transferring + && probe.commit_min < probe.commit_max + && probe.next_op_resident +} + +/// Count one sweep tick against `gap_ticks` and answer whether this partition +/// may arm repair now. +/// +/// Level-triggered, because every edge-triggered arming site is starvable: the +/// commit-heartbeat backstop fires only on `CommitOutcome::Advanced`, and under +/// sustained produce the prepares consume the advance in preflight before the +/// gap check drops them, so the heartbeat lands as `Accepted` and the gap wedges +/// until an unrelated view change. +/// +/// A capped-out arm keeps its debounce satisfied rather than starting over, so +/// the group arms on the next pass with a slot free. +const fn drive_partition_gap_debounce( + probe: &GapProbe, + gap_ticks: &mut u32, + debounce_ticks: u32, + arms_this_tick: usize, +) -> bool { + if !partition_is_gap_stopped(probe) { + *gap_ticks = 0; + return false; + } + *gap_ticks = gap_ticks.saturating_add(1); + *gap_ticks >= debounce_ticks && arms_this_tick < PARTITION_REPAIR_ARMS_PER_TICK_MAX +} + +/// Read the gap probe off a live partition. +fn partition_gap_probe(partition: &IggyPartition) -> GapProbe +where + B: MessageBus, + SB: SuperblockStore, +{ + let consensus = partition.consensus(); + let commit_min = consensus.commit_min(); + // Transfer first: it supersedes repair, so naming it is the truthful + // diagnostic when both happen to be set. + let recovery = if partition.transfer.is_some() { + RecoveryOwner::Transfer + } else if partition.transfer_rearm.is_some() { + RecoveryOwner::TransferRearm + } else if partition.repair.is_some() { + RecoveryOwner::Repair + } else { + RecoveryOwner::Nobody + }; + GapProbe { + normal: consensus.is_normal(), + transferring: consensus.is_transferring(), + recovery, + commit_min, + commit_max: consensus.commit_max(), + next_op_resident: partition + .log + .journal() + .inner + .header_by_op(commit_min.saturating_add(1)) + .is_some(), + } +} + /// Whether the parked `StartView` log names every op in the uncommitted /// suffix `(commit_max, head]`, in descending order. Only this canonical list /// makes fetching bodies above the commit point safe. @@ -10064,3 +10251,264 @@ mod superblock_fail_stop_tests { assert!(superblock_wedged(121, 120)); } } + +#[cfg(test)] +mod gap_detector_tests { + //! The level-triggered repair arm the partition tick sweep runs. + //! + //! Its whole reason to exist is that the edge-triggered arming sites are + //! starvable, so the guards it shares with them and the debounce that keeps + //! it off healthy traffic are the parts worth pinning. + + use super::{ + GapProbe, PARTITION_REPAIR_ARMS_PER_TICK_MAX, RecoveryOwner, drive_partition_gap_debounce, + partition_is_gap_stopped, partition_is_walk_stalled, + }; + + const DEBOUNCE: u32 = 100; + + /// A gap-stopped follower: committed through op 10, walkable only to 5, + /// because op 6 is not in its journal. + const fn gap_stopped() -> GapProbe { + GapProbe { + normal: true, + transferring: false, + recovery: RecoveryOwner::Nobody, + commit_min: 5, + commit_max: 10, + next_op_resident: false, + } + } + + /// A walk-stalled follower: the same lag, but op 6 IS in its journal, so + /// nothing needs fetching and the walk just has to run. + fn walk_stalled() -> GapProbe { + GapProbe { + next_op_resident: true, + ..gap_stopped() + } + } + + #[test] + fn given_a_lagging_follower_with_the_next_op_resident_when_probed_should_not_be_gap_stopped() { + // The half that keeps the predicate honest. A follower advances + // commit_max from every prepare header in preflight, so commit_min < + // commit_max is transiently true on any pipelined tick; without the + // journal-hole test the driver would request repair against ordinary + // produce, on every partition, forever. + let healthy = GapProbe { + next_op_resident: true, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&healthy)); + assert!(partition_is_gap_stopped(&gap_stopped())); + } + + #[test] + fn given_a_caught_up_follower_when_probed_should_not_be_gap_stopped() { + let caught_up = GapProbe { + commit_min: 10, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&caught_up)); + } + + #[test] + fn given_a_replica_outside_normal_status_when_probed_should_not_be_gap_stopped() { + // A view change owns the log while it runs, and `maybe_request_partition_repair` + // refuses outside Normal anyway; arming here would only burn a nonce. + let electing = GapProbe { + normal: false, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&electing)); + + let installing = GapProbe { + transferring: true, + ..gap_stopped() + }; + assert!(!partition_is_gap_stopped(&installing)); + } + + #[test] + fn given_recovery_already_owned_when_probed_should_not_be_gap_stopped() { + for owner in [ + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ] { + let owned = GapProbe { + recovery: owner, + ..gap_stopped() + }; + assert!( + !partition_is_gap_stopped(&owned), + "{owner:?} owns the recovery; a second session would race it" + ); + } + } + + #[test] + fn given_a_gap_stopped_follower_when_debouncing_should_arm_only_at_the_threshold() { + let probe = gap_stopped(); + let mut gap_ticks = 0; + for tick in 1..DEBOUNCE { + assert!( + !drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, 0), + "armed at tick {tick}, before the debounce elapsed" + ); + } + assert!(drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + 0 + )); + } + + #[test] + fn given_a_debounce_in_progress_when_the_gap_closes_should_reset_the_counter() { + let stopped = gap_stopped(); + let walkable = GapProbe { + next_op_resident: true, + ..stopped + }; + let mut gap_ticks = 0; + for _ in 0..DEBOUNCE - 1 { + drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0); + } + assert_eq!(gap_ticks, DEBOUNCE - 1); + + assert!(!drive_partition_gap_debounce( + &walkable, + &mut gap_ticks, + DEBOUNCE, + 0 + )); + assert_eq!(gap_ticks, 0, "progress must restart the debounce"); + assert!( + !drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0), + "a fresh gap must serve its own debounce, not inherit the old count" + ); + } + + #[test] + fn given_a_follower_with_resident_committed_ops_when_probed_should_be_walk_stalled() { + assert!(partition_is_walk_stalled(&walk_stalled())); + assert!( + !partition_is_walk_stalled(&gap_stopped()), + "a missing next op is repair's job; a walk over it would stop dead" + ); + } + + #[test] + fn given_a_caught_up_follower_when_probed_should_not_be_walk_stalled() { + let caught_up = GapProbe { + commit_min: 10, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&caught_up)); + } + + #[test] + fn given_a_replica_outside_normal_status_when_probed_should_not_be_walk_stalled() { + let electing = GapProbe { + normal: false, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&electing)); + + // Same gate as the on-commit arm: a walk during a transfer can advance + // commit_min past the incoming frontier. + let installing = GapProbe { + transferring: true, + ..walk_stalled() + }; + assert!(!partition_is_walk_stalled(&installing)); + } + + #[test] + fn given_recovery_already_owned_when_the_next_op_is_resident_should_still_be_walk_stalled() { + // Deliberate: `apply_repaired_prepare` journals without walking, so a + // gated walk would sit parked for the whole session while the resident + // prefix is already applicable. + for owner in [ + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ] { + let owned = GapProbe { + recovery: owner, + ..walk_stalled() + }; + assert!( + partition_is_walk_stalled(&owned), + "{owner:?} owns the fetch, not the resident prefix" + ); + } + } + + #[test] + fn given_any_probe_when_evaluated_should_never_be_both_gap_stopped_and_walk_stalled() { + // The two halves split on `next_op_resident`; if they ever overlap, one + // tick both arms repair and walks the window it is fetching. + let owners = [ + RecoveryOwner::Nobody, + RecoveryOwner::Repair, + RecoveryOwner::Transfer, + RecoveryOwner::TransferRearm, + ]; + for normal in [false, true] { + for transferring in [false, true] { + for recovery in owners { + for (commit_min, commit_max) in [(5, 10), (10, 10)] { + for next_op_resident in [false, true] { + let probe = GapProbe { + normal, + transferring, + recovery, + commit_min, + commit_max, + next_op_resident, + }; + assert!( + !(partition_is_gap_stopped(&probe) + && partition_is_walk_stalled(&probe)), + "both predicates claim {probe:?}" + ); + } + } + } + } + } + } + + #[test] + fn given_the_per_tick_cap_reached_when_debouncing_should_defer_without_losing_the_debounce() { + let probe = gap_stopped(); + let mut gap_ticks = DEBOUNCE; + assert!( + !drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + PARTITION_REPAIR_ARMS_PER_TICK_MAX + ), + "the cap must refuse the arm" + ); + assert!( + gap_ticks > DEBOUNCE, + "a capped-out group stays due; restarting its debounce would push the \ + arm a whole interval out per contended tick" + ); + assert!( + drive_partition_gap_debounce( + &probe, + &mut gap_ticks, + DEBOUNCE, + PARTITION_REPAIR_ARMS_PER_TICK_MAX - 1 + ), + "the same group arms on the next pass with a slot free" + ); + } +} diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index 81608ba166..cefc3d4458 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -79,17 +79,12 @@ pub struct FrameDropLabel { /// had not materialised and its park buffer was at capacity /// (`reason=park_overflow`), a parked frame retired with no client to answer /// (`reason=park_dropped`), or a routing send the target inbox refused. A shed -/// client request is answered with -/// a retriable status, so the client recovers -- but a shed *prepare* is not -/// covered by retransmit once its op has reached quorum -/// (`consensus::retransmit_targets` skips `ok_quorum_received`, and the -/// partition plane creates a repair session only in `on_start_view`), so it -/// leaves that backup behind until an unrelated view change. -// -// TODO(krishna): give the partition plane a normal-status repair driver so a -// shed or refused prepare is repaired without waiting for a view change. Until -// then `variant=partition` is the only signal that a backup may be stranded -// behind `commit_max`. +/// client request is answered with a retriable status, so the client recovers. +/// A shed *prepare* has nobody to answer and is not covered by retransmit once +/// its op has reached quorum (`consensus::retransmit_targets` skips +/// `ok_quorum_received`), so the backup gap-stops; `tick_partitions`' sweep is +/// what repairs it, and `partition_prepare_gap_drops_total` is what counts the +/// prepares that reached the gap check. pub mod frame_drop_variant { pub const CONSENSUS: &str = "consensus"; pub const FD_TRANSFER: &str = "fd_transfer"; @@ -204,6 +199,7 @@ pub struct ShardMetrics { partition_frames_rejected_ahead_total: Counter, partition_requests_denied_transient_total: Counter, partition_repair_serves_deferred_purge_total: Counter, + partition_prepare_gap_drops_total: Counter, } impl ShardMetrics { @@ -228,6 +224,7 @@ impl ShardMetrics { partition_frames_rejected_ahead_total: Counter::default(), partition_requests_denied_transient_total: Counter::default(), partition_repair_serves_deferred_purge_total: Counter::default(), + partition_prepare_gap_drops_total: Counter::default(), } } @@ -401,6 +398,25 @@ impl ShardMetrics { self.partition_repair_serves_deferred_purge_total.get() } + /// Add the prepares a partition's backup gap check destroyed since the last + /// sweep, drained per tick from `IggyPartition::take_prepare_gap_drops`. + /// + /// Deliberately NOT a `frame_drops_total{variant=partition}` reason: that + /// family means the bus or the router shed a frame, and the simulator + /// asserts it stays at zero on runs with no injected loss. A gap drop is a + /// protocol-ordering drop that any real loss produces, and the sweep repairs + /// it, so folding the two would turn a routing-fault alert into noise. + pub fn record_partition_prepare_gap_drops(&self, drops: u64) { + self.partition_prepare_gap_drops_total.inc_by(drops); + } + + /// Snapshot of `partition_prepare_gap_drops_total`. Test/simulator accessor. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn partition_prepare_gap_drops_value(&self) -> u64 { + self.partition_prepare_gap_drops_total.get() + } + /// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator /// accessor, readable from any crate under those cfgs so the crates that /// drive the reconciler can assert a reject did not happen. @@ -492,6 +508,11 @@ impl ShardMetrics { "partition repair serves or completions deferred until a committed purge applies", self.partition_repair_serves_deferred_purge_total.clone(), ); + registry.register( + "partition_prepare_gap_drops", + "replicated prepares dropped out of order by a backup's gap check", + self.partition_prepare_gap_drops_total.clone(), + ); } } diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 128278899a..44421de461 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -4657,3 +4657,646 @@ mod repair_frontier_tests { ); } } + +#[cfg(test)] +mod partition_repair_driver_tests { + //! A backup that missed a committed partition prepare recovers in Normal + //! status, without waiting for a view change. + //! + //! Every edge-triggered arming site is starvable. A follower advances + //! `commit_max` from each prepare header in `replicate_preflight`, before + //! the gap check drops the prepare, so under produce the primary's commit + //! heartbeat lands as `CommitOutcome::Accepted` and the backstop that would + //! arm repair on `Advanced` never runs. These tests starve that edge + //! outright (no commit heartbeat for the group reaches the lagging replica) + //! so nothing but the level-triggered detector in `tick_partitions` can + //! close the gap. + + use super::*; + use bytes::Bytes; + use consensus::Status; + use iggy_binary_protocol::{ + CommitHeader, PrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader, + }; + use packet::Packet; + use std::sync::atomic::{AtomicU64, Ordering}; + + /// Chain replication runs 0 -> 1 -> 2 and stops before the primary, so + /// replica 2 forwards to nobody: it is the only replica whose losses do not + /// also starve its successor, and therefore the group, of quorum. + const LAGGING: u8 = 2; + + const CLIENT_ID: u128 = 1; + + /// Ops that replicate cleanly before the fault, so the gap opens above a + /// committed prefix rather than at the group's first op. + const WARMUP_SENDS: usize = 3; + + /// Ticks stepped after each produce. Keeps one send's round trip inside its + /// own window so the workload is legible tick by tick. + const STEPS_PER_SEND: usize = 12; + + /// Produces issued with the fault standing. `partitions::REPAIR_RETRY_TICKS` + /// is 100, so this must carry the run past the debounce while prepares keep + /// consuming the `commit_max` advance the heartbeat backstop needs. + const GAP_SENDS: usize = 12; + + /// Quiet ticks after the produce stops, for the repair stream to land. + /// + /// Budgeted against `NORMAL_HEARTBEAT_TICKS` (500): with this group's commit + /// heartbeats withheld, the lagging replica elects once that timer fires, and + /// an election would heal the gap through `on_start_view` instead. Every + /// phase after the fault is installed has to fit inside it. + const QUIET_STEPS: usize = 160; + + /// Budget for the group to settle once the fault is lifted. Generous rather + /// than tuned: the drain loop breaks on convergence, and the tail above the + /// repaired window waits on whichever interval-driven site picks it up. + const DRAIN_STEPS: usize = 600; + + /// Ticks of healthy load in the no-false-positive test, several debounce + /// intervals' worth so the sweep gets many chances to arm. + const LOAD_TICKS: usize = 4 * partitions::REPAIR_RETRY_TICKS as usize; + + /// Paced at a round trip rather than one submit per tick: the pipeline caps + /// at `PIPELINE_PREPARE_QUEUE_MAX`, so submitting faster than the group + /// commits just collects transient rejections and the group goes quiet. + const TICKS_PER_SEND: usize = 4; + + /// Ops the healthy run must have committed for its verdict to mean anything. + /// A produce's round trip is four one-way hops plus tick granularity, so this + /// network commits on the order of one op per fifteen ticks however hard the + /// client pushes; the floor only has to prove the group was live across + /// several debounce intervals, not that it was saturated. + const COMMITTED_MIN: u64 = 20; + + /// Produces issued with the fault standing in the walk-starvation test: few + /// enough that the gap debounce fires only after produce stops, so the + /// repair window closes at the last carried commit and lands fully resident. + const STRAND_SENDS: usize = 3; + + /// Quiet budget for the walk-starvation test: debounce, repair stream, then + /// the drain, kept under `NORMAL_HEARTBEAT_TICKS` so an election cannot be + /// the healer. + const STRAND_QUIET_STEPS: usize = 300; + + /// The partition-plane prepare a packet carries, if it carries one for + /// `group`. + fn prepare_for(packet: &Packet, group: u64) -> Option { + if packet.message.header().command != Command::Prepare { + return None; + } + let header: &PrepareHeader = + bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::()]); + (header.group == group).then_some(*header) + } + + /// Whether a packet is a commit heartbeat for `group`. + fn is_commit_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::Commit { + return false; + } + let header: &CommitHeader = + bytemuck::checked::from_bytes(&packet.message.as_slice()[..size_of::()]); + header.group == group + } + + /// Whether a packet is a repair request for `group`. + fn is_request_prepares_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::RequestPrepares { + return false; + } + let header: &RequestPreparesHeader = bytemuck::checked::from_bytes( + &packet.message.as_slice()[..size_of::()], + ); + header.group == group + } + + /// Whether a packet is a repair stream terminator for `group`. + fn is_repair_done_for(packet: &Packet, group: u64) -> bool { + if packet.message.header().command != Command::RepairDone { + return false; + } + let header: &RepairRangeReplyHeader = bytemuck::checked::from_bytes( + &packet.message.as_slice()[..size_of::()], + ); + header.group == group + } + + fn cluster(seed: u64) -> (Simulator, SimClient) { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let replica_count: u8 = 3; + let network_opts = packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed, + ..packet::PacketSimulatorOptions::default() + }; + let sim = Simulator::new( + replica_count as usize, + std::iter::once(CLIENT_ID), + network_opts, + ); + (sim, SimClient::new(CLIENT_ID)) + } + + /// Submit `sends` produces against `namespace`, stepping between each. + fn produce( + sim: &mut Simulator, + client: &SimClient, + namespace: IggyNamespace, + sends: usize, + tag: &str, + ) { + for index in 0..sends { + let msg = client.send_messages(namespace, &[Bytes::from(format!("{tag}-{index}"))]); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + for _ in 0..STEPS_PER_SEND { + sim.step(); + } + } + } + + /// `(status, view, commit_min, commit_max)` of one replica's partition group. + fn group_state( + sim: &Simulator, + replica: u8, + namespace: IggyNamespace, + ) -> (Status, u32, u64, u64) { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + let partition = shard + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the replica hosts the group"); + let consensus = partition.consensus(); + ( + consensus.status(), + consensus.view(), + consensus.commit_min(), + consensus.commit_max(), + ) + } + + fn journal_holds(sim: &Simulator, replica: u8, namespace: IggyNamespace, op: u64) -> bool { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + shard + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| partition.log.journal().inner.header_by_op(op).is_some()) + } + + fn gap_drops(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> u64 { + sim.replicas[replica as usize] + .partition_shard(namespace) + .metrics() + .partition_prepare_gap_drops_value() + } + + fn transfer_armed(sim: &Simulator, replica: u8, namespace: IggyNamespace) -> bool { + let shard = sim.replicas[replica as usize].partition_shard(namespace); + shard + .plane + .partitions() + .get_by_ns(&namespace) + .is_some_and(|partition| { + partition.transfer.is_some() + || partition.consensus().state_transfer_stage() + != consensus::StateTransferStage::Idle + }) + } + + #[test] + fn given_a_backup_that_dropped_a_committed_prepare_when_heartbeat_advances_are_starved_should_repair_in_normal_status() + { + // Statics, not captures: the link hooks are bare `fn` pointers. Declared + // inside the test because the sibling tests in this binary run in + // parallel and would otherwise share them. + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first partition prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold this group's commit heartbeats, so the + /// `Advanced` backstop can never run, and withhold retransmits of the + /// dropped op. The retransmit half stands in for production behaviour + /// rather than adding a fault: `consensus::retransmit_targets` skips an + /// op that already reached quorum, and this op reaches quorum on 0 and 1 + /// alone. Without it the retry timer could heal the gap and the test + /// would pass with no repair driver at all. + fn starve_commit_edge(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if let Some(header) = prepare_for(packet, group) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0232); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + produce(&mut sim, &client, namespace, GAP_SENDS, "gap"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + assert!( + gap_drops(&sim, LAGGING, namespace) > 0, + "the lagging replica never reached its backup gap check, so the \ + prepares after the withheld op were not dropped as a gap" + ); + + for _ in 0..QUIET_STEPS { + sim.step(); + } + + // Judged with the blockade still standing, so the tick driver is the only + // thing that can have armed the repair: no commit heartbeat for this + // group has reached the replica since the gap opened, and `on_commit` is + // where the `Advanced` backstop lives. + let (status, view, commit_min, _) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change healed the gap instead of the repair driver; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + assert!( + journal_holds(&sim, LAGGING, namespace, withheld), + "op {withheld} was never repaired back into the lagging replica's journal" + ); + assert!( + commit_min >= withheld, + "the commit walk never crossed the repaired hole: stopped at \ + {commit_min}, the withheld op is {withheld}" + ); + + // Lift the blockade and let the group settle. The commit walk is driven + // by arriving frames, so with this group's heartbeats withheld the tail + // above the repaired window has nothing to advance it; that is the + // injected fault, not the gap under test. + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = None; + for _ in 0..DRAIN_STEPS { + sim.step(); + let (_, _, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + if commit_min == commit_max { + break; + } + } + let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + assert_eq!((status, view), (Status::Normal, 0)); + assert_eq!( + commit_min, commit_max, + "the lagging replica is still gap-stopped: committed through \ + {commit_max} but walkable only to {commit_min}" + ); + } + + #[test] + fn given_a_repair_armed_by_the_tick_driver_when_the_range_is_evicted_should_convert_to_state_transfer() + { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + fn starve_commit_edge(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if let Some(header) = prepare_for(packet, group) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0233); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + produce(&mut sim, &client, namespace, GAP_SENDS, "gap"); + assert_ne!( + WITHHELD_OP.load(Ordering::Relaxed), + 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + + // Compact the serving side past the gap. This plane's journal is + // memory-only, so wiping it IS its retention floor moving: the serve + // path reads `repair_retained_from` as `None` and reports eviction from + // its own commit frontier, which is exactly what a peer that + // checkpointed past the requested window answers. + { + let primary = sim.replicas[0].partition_shard(namespace); + let partition = primary + .plane + .partitions() + .get_by_ns(&namespace) + .expect("the primary hosts the group"); + partition.log.journal().inner.clear_all(); + } + + for _ in 0..QUIET_STEPS { + sim.step(); + if transfer_armed(&sim, LAGGING, namespace) { + break; + } + } + + let (status, view, ..) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change armed the recovery instead of the tick-armed repair session" + ); + assert!( + transfer_armed(&sim, LAGGING, namespace), + "the tick-armed repair session hit an evicted range but never converted \ + to a state transfer (status {status:?})" + ); + } + + #[test] + fn given_healthy_pipelined_traffic_when_no_gap_exists_should_not_arm_repair() { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static REPAIR_REQUESTS: AtomicU64 = AtomicU64::new(0); + + /// Observer, not a fault: counts this group's repair requests and passes + /// every packet through. + fn count_repair_requests(packet: &Packet) -> bool { + if is_request_prepares_for(packet, GAP_NS.load(Ordering::Relaxed)) { + REPAIR_REQUESTS.fetch_add(1, Ordering::Relaxed); + } + false + } + + // TWO replicas, so quorum spans both: no op can commit without the + // backup's ack, every reordering-induced gap therefore blocks quorum, and + // `consensus::retransmit_targets` refills it. At three, the network's + // per-tick delivery shuffle lets an op commit on the primary and its + // first chain hop while the last replica loses it for good, which is the + // very fault the sibling tests inject -- it would then be repaired here, + // correctly, and this assertion would fire on a healthy driver. + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let replica_count: u8 = 2; + let mut sim = Simulator::new( + replica_count as usize, + std::iter::once(CLIENT_ID), + packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed: 0x5EED_0234, + ..packet::PacketSimulatorOptions::default() + }, + ); + let client = SimClient::new(CLIENT_ID); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + REPAIR_REQUESTS.store(0, Ordering::Relaxed); + + for (from, to) in [(0u8, 1u8), (1, 0)] { + *sim.network + .link_drop_packet_fn(ProcessId::Replica(from), ProcessId::Replica(to)) = + Some(count_repair_requests); + } + + // Sustained, not bursty: a produce every tick, for several debounce + // intervals, so the sweep gets many chances to arm against ordinary + // pipelining. The backup's lag is sampled per tick, and the run asserts + // it went somewhere, so a green result cannot come from a workload that + // never loaded the group. + let mut lag_run = 0u32; + let mut longest_lag_run = 0u32; + let sample = |sim: &Simulator, lag_run: &mut u32, longest: &mut u32| { + let (_, _, commit_min, commit_max) = group_state(sim, 1, namespace); + if commit_min < commit_max { + *lag_run += 1; + *longest = (*longest).max(*lag_run); + } else { + *lag_run = 0; + } + }; + for tick in 0..LOAD_TICKS { + if tick % TICKS_PER_SEND == 0 { + let msg = + client.send_messages(namespace, &[Bytes::from(format!("healthy-{tick}"))]); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + } + sim.step(); + sample(&sim, &mut lag_run, &mut longest_lag_run); + } + for _ in 0..QUIET_STEPS { + sim.step(); + sample(&sim, &mut lag_run, &mut longest_lag_run); + } + + let committed = group_state(&sim, 1, namespace).2; + let sends = LOAD_TICKS / TICKS_PER_SEND; + assert!( + committed >= COMMITTED_MIN, + "the backup committed only {committed} ops across {sends} sends, so the \ + sweep was never driven over a loaded group" + ); + // Recorded, not asserted: a two-replica group's backup keeps pace tick for + // tick, so the lag a naive `commit_min < commit_max` detector would read + // as a gap does not arise from healthy partition traffic at all. What that + // detector actually mistakes for a gap is a walk BLOCKED over resident ops + // (a deferred purge, a transfer install), which no healthy workload + // reaches; `gap_detector_tests` pins the predicate directly instead. + assert_eq!( + longest_lag_run, 0, + "healthy two-replica traffic left the backup lagging for \ + {longest_lag_run} consecutive ticks; if this ever becomes non-zero \ + the predicate's journal-hole half is load-bearing HERE too and this \ + test should assert against it rather than record it" + ); + for replica in 0..replica_count { + let (status, view, commit_min, commit_max) = group_state(&sim, replica, namespace); + assert_eq!( + (status, view), + (Status::Normal, 0), + "replica {replica} left view 0 / Normal, so a view change could \ + account for repair traffic" + ); + // Residual lag is transient at worst: a heartbeat carrying a commit + // the replica already knows is `CommitOutcome::Accepted`, so a backup + // can sit with committed-but-unwalked ops until the tick sweep's + // walk-stalled backstop resumes the walk. Every one of them is + // RESIDENT, which is what separates it from a gap. + if commit_min < commit_max { + assert!( + journal_holds(&sim, replica, namespace, commit_min + 1), + "replica {replica} lags at {commit_min} of {commit_max} with op \ + {} missing, so a no-loss run produced a real hole", + commit_min + 1 + ); + } + } + assert_eq!( + REPAIR_REQUESTS.load(Ordering::Relaxed), + 0, + "the tick driver requested repair on a healthy group; its gap predicate \ + is reading ordinary commit lag as a journal hole" + ); + } + + #[test] + fn given_a_backup_holding_resident_committed_ops_when_every_walk_edge_is_starved_should_drain_in_normal_status() + { + static GAP_NS: AtomicU64 = AtomicU64::new(0); + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + static WITHHELD_DONES: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first partition prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, GAP_NS.load(Ordering::Relaxed)) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold every direct prepare (live ones ride the + /// chain, so this starves only retransmit heals), the group's commit + /// heartbeats, and its repair terminators. The repaired ops themselves + /// pass, so the window lands resident while `complete_repair`, the walk + /// the terminator would run, never fires. + fn starve_walk_edges(packet: &Packet) -> bool { + let group = GAP_NS.load(Ordering::Relaxed); + if prepare_for(packet, group).is_some() { + return true; + } + if is_repair_done_for(packet, group) { + WITHHELD_DONES.fetch_add(1, Ordering::Relaxed); + return true; + } + is_commit_for(packet, group) + } + + let (mut sim, client) = cluster(0x5EED_0235); + let namespace = IggyNamespace::new(1, 1, 0); + sim.init_partition(namespace); + sim.register_client_with_primary(&client); + GAP_NS.store(namespace.inner(), Ordering::Relaxed); + WITHHELD_OP.store(0, Ordering::Relaxed); + WITHHELD_DONES.store(0, Ordering::Relaxed); + + produce(&mut sim, &client, namespace, WARMUP_SENDS, "warmup"); + let (_, _, warm_commit_min, _) = group_state(&sim, LAGGING, namespace); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_walk_edges); + + produce(&mut sim, &client, namespace, STRAND_SENDS, "strand"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no partition prepare crossed the chain link, so the fault never armed" + ); + + for _ in 0..STRAND_QUIET_STEPS { + sim.step(); + let (_, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + if view != 0 || (commit_min >= withheld && commit_min == commit_max) { + break; + } + } + + assert!( + WITHHELD_DONES.load(Ordering::Relaxed) > 0, + "no repair terminator was withheld, so the walk was never starved and \ + a green run would not prove the tick backstop" + ); + assert!( + journal_holds(&sim, LAGGING, namespace, withheld), + "op {withheld} was never repaired back into the lagging replica's journal" + ); + let (status, view, commit_min, commit_max) = group_state(&sim, LAGGING, namespace); + assert_eq!( + view, 0, + "a view change drained the walk instead of the tick backstop; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + for op in commit_min + 1..=commit_max { + assert!( + journal_holds(&sim, LAGGING, namespace, op), + "op {op} is not resident, so this run stranded on a repair gap, \ + not a parked walk" + ); + } + assert_eq!( + commit_min, commit_max, + "the walk never resumed over resident committed ops: walkable to \ + {commit_min}, committed through {commit_max}, every op between resident" + ); + } +} From d5c8ba1906bb2d344afb6eecd6232bc6b4ee1540 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 31 Aug 2026 15:47:38 +0200 Subject: [PATCH 4/4] fix(shard): drive metadata repair and the commit walk from the tick --- core/metadata/src/impls/metadata.rs | 45 ++- core/shard/src/lib.rs | 269 ++++++++++---- core/shard/src/metrics.rs | 24 ++ core/simulator/src/lib.rs | 523 +++++++++++++++++++++++++++- 4 files changed, 787 insertions(+), 74 deletions(-) diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index b7b7f83375..217e03c69b 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -769,6 +769,11 @@ pub struct IggyMetadata { /// whole snapshot on shard 0's pump, and hands each requester its own /// multi-MB copy. transfer_offer_cache: RefCell>>, + /// 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. The partition twin is `IggyPartition::prepare_gap_drops`. + prepare_gap_drops: Cell, } impl IggyMetadata @@ -808,11 +813,27 @@ where commit_notifier: RefCell::new(None), client_table_frontier: Cell::new(0), transfer_offer_cache: RefCell::new(None), + prepare_gap_drops: Cell::new(0), } } } impl IggyMetadata { + /// Record one prepare destroyed by the backup gap check. Drained by + /// `tick_metadata` into `metadata_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. + pub fn note_prepare_gap_drop(&self) { + self.prepare_gap_drops + .set(self.prepare_gap_drops.get().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(&self) -> u64 { + self.prepare_gap_drops.replace(0) + } + /// Slot capacity of the LIVE client table, i.e. the largest transferred /// table this replica can absorb. /// @@ -1219,6 +1240,7 @@ where sequencer_op = current_op, "on_replicate: dropping out-of-order prepare (gap)" ); + self.note_prepare_gap_drop(); return; } } else { @@ -3535,10 +3557,10 @@ where let Some(header) = journal.handle().header(op as usize) else { // Gap-stop: the walk halts at the first missing prepare and - // resumes once it is refilled. Live drops refill via the - // primary's prepare retransmit; a replica behind at recovery - // or after StartView adoption arms a `MetadataRepairSession` - // (shard) that re-requests the missing window. + // resumes once it is refilled -- by the primary's retransmit + // while the op lacks quorum, else by a `MetadataRepairSession` + // (armed at StartView adoption or by `tick_metadata`'s gap + // detector, since retransmit skips quorum-acked ops). break; }; let header = *header; @@ -4172,6 +4194,21 @@ mod tests { IggyMetadata::new(None, None, None, None, TestMux::default(), None) } + #[test] + fn take_prepare_gap_drops_drains_the_count() { + let md = peer_metadata(); + assert_eq!(md.take_prepare_gap_drops(), 0); + + md.note_prepare_gap_drop(); + md.note_prepare_gap_drop(); + assert_eq!(md.take_prepare_gap_drops(), 2); + assert_eq!( + md.take_prepare_gap_drops(), + 0, + "a second drain must not re-report drops the metrics already counted" + ); + } + #[test] fn commit_notifier_fires_with_received_operation() { let md = peer_metadata(); diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index c1bff0dc07..f0e007dbac 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -1259,6 +1259,14 @@ where /// repair takes over at install. See [`MetadataTransferSession`]. metadata_transfer: RefCell>, + /// Consecutive ticks the metadata group has been seen gap-stopped + /// (committed ops it cannot walk to, because the op at its commit frontier + /// plus one is missing from the WAL). Debounces `tick_metadata`'s + /// level-triggered repair arm; the partition twin is + /// `IggyPartition::gap_ticks`. One metadata group per node, so shard-level + /// state suffices (precedent: [`Self::metadata_transfer_attempts`]). + metadata_gap_ticks: Cell, + /// Serving-side cache of state-transfer offers, both planes, keyed by /// `(namespace, requester replica id)`. Bounded by the replica count times /// the groups this shard serves; replaced per fresh nonce. @@ -1591,6 +1599,7 @@ where shard_park_shedding: Cell::new(false), metadata_repair: RefCell::new(None), metadata_transfer: RefCell::new(None), + metadata_gap_ticks: Cell::new(0), state_transfer_offers: RefCell::new(HashMap::new()), partition_offer_builds: RefCell::new(HashMap::new()), served_segment_cache: RefCell::new(ServedSegmentCache::default()), @@ -1921,6 +1930,7 @@ where shard_park_shedding: Cell::new(false), metadata_repair: RefCell::new(None), metadata_transfer: RefCell::new(None), + metadata_gap_ticks: Cell::new(0), state_transfer_offers: RefCell::new(HashMap::new()), partition_offer_builds: RefCell::new(HashMap::new()), served_segment_cache: RefCell::new(ServedSegmentCache::default()), @@ -4080,17 +4090,16 @@ where // its own WAL (a late joiner missed the ops below the // primary's active window; the primary only retransmits // uncommitted ops, never the committed prefix). Without - // this, such a replica learns it is behind and does - // nothing about it -- metadata repair is otherwise only - // rooted at StartView adoption, which a same-view - // late joiner never sees. Request repair from the - // primary; if it has checkpointed past the gap the - // repair floor evicts and the handler above converts to - // state transfer. Idempotent: `maybe_request_metadata_repair` - // no-ops when caught up, already transferring, or a - // session is live, so a caught-up replica and a - // cold-start node (commit_max == commit_min == 0) both - // skip it. + // this, such a replica waits out `tick_metadata`'s + // debounced gap detector; this edge is the fast path + // when a heartbeat does land as `Advanced`. Request + // repair from the primary; if it has checkpointed past + // the gap the repair floor evicts and the handler above + // converts to state transfer. Idempotent: + // `maybe_request_metadata_repair` no-ops when caught + // up, already transferring, or a session is live, so a + // caught-up replica and a cold-start node + // (commit_max == commit_min == 0) both skip it. self.maybe_request_metadata_repair(consensus, header.replica) .await; } @@ -5130,6 +5139,18 @@ where consensus.group(), ) .await; + } else { + // The window is fully walked; only the `RepairDone` that clears + // the session was lost, or raced a walk that finished without + // it. Nothing is left to request, so close the session here or + // it blocks every future arm forever. + tracing::info!( + shard = self.id, + to_op, + peer, + "metadata repair window fully walked; closing the stalled session" + ); + *self.metadata_repair.borrow_mut() = None; } } } @@ -5403,8 +5424,10 @@ where } /// Start metadata tail journal-repair from `peer` when the commit walk - /// gap-stopped below the known frontier. Shared by `StartView` adoption - /// and the post-install step of a state transfer. + /// gap-stopped below the known frontier. Every arming site funnels through + /// here: `StartView` adoption, the commit-heartbeat backstop, the + /// state-transfer fallbacks, and `tick_metadata`'s gap detector, whose + /// idempotence these guards provide. #[allow(clippy::future_not_send)] async fn maybe_request_metadata_repair

(&self, consensus: &VsrConsensus, peer: u8) where @@ -6645,13 +6668,13 @@ where self.metrics.record_partition_prepare_gap_drops(gap_drops); } let probe = partition_gap_probe(partition); - let gap_arm = drive_partition_gap_debounce( + let gap_arm = drive_gap_debounce( &probe, &mut partition.gap_ticks, repair_retry_ticks, repair_arms, ); - (gap_arm, partition_is_walk_stalled(&probe)) + (gap_arm, is_walk_stalled(&probe)) }; if gap_arm { let Some(partition) = partitions.get_mut_by_ns(&namespace) else { @@ -8500,7 +8523,36 @@ where } } - #[allow(clippy::future_not_send)] + /// Read the gap probe off the metadata plane; `partition_gap_probe`'s twin. + /// A shard method because the recovery slots live here, not on the plane. + fn metadata_gap_probe

(&self, consensus: &VsrConsensus, journal: &MJ) -> GapProbe + where + B: MessageBus, + P: Pipeline, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + { + let commit_min = consensus.commit_min(); + GapProbe { + normal: consensus.is_normal(), + transferring: consensus.is_transferring(), + recovery: metadata_recovery_owner( + self.metadata_transfer.borrow().is_some(), + consensus.is_transferring(), + self.metadata_repair.borrow().is_some(), + ), + commit_min, + commit_max: consensus.commit_max(), + // The same query the commit walk gap-stops on. Safe against the + // snapshot floor: a checkpoint drains only to `commit_min`, so + // `commit_min + 1` never sits below it and a `None` is a real hole. + #[allow(clippy::cast_possible_truncation)] + next_op_resident: journal.handle().header((commit_min + 1) as usize).is_some(), + } + } + + #[allow(clippy::future_not_send, clippy::too_many_lines)] pub async fn tick_metadata(&self) where B: MessageBus, @@ -8565,6 +8617,56 @@ where self.advance_pending_metadata_view().await; self.expire_idle_state_transfer_offers(); + // Level-triggered gap detector, the metadata twin of the one in + // `tick_partitions`; the starvation is the same (`replicate_preflight` + // advances `commit_max` before the gap check drops the prepare, so the + // `Advanced`-gated arm in `on_commit` never fires under sustained + // traffic). Before the transfer-stall block: its exhausted branch + // returns early and would skip a detector placed after it. + if let Some(journal) = metadata.journal.as_ref() { + let gap_drops = metadata.take_prepare_gap_drops(); + if gap_drops > 0 { + self.metrics.record_metadata_prepare_gap_drops(gap_drops); + } + let probe = self.metadata_gap_probe(consensus, journal); + let mut gap_ticks = self.metadata_gap_ticks.get(); + // Zero arms used: one metadata group per node, no per-tick cap. + let gap_arm = + drive_gap_debounce(&probe, &mut gap_ticks, self.repair_retry_ticks.get(), 0); + self.metadata_gap_ticks.set(gap_ticks); + if gap_arm { + let peer = consensus.primary_index(consensus.view()); + tracing::info!( + shard = self.id, + commit_min = probe.commit_min, + commit_max = probe.commit_max, + peer, + "metadata gap-stopped past the debounce; arming repair from the primary" + ); + // Always repair, never classify the gap up front: a window + // below the peer's retention floor is answered `RangeEvicted`, + // and `on_repair_range_reply` converts that to a state + // transfer. The floor is only learned through that refusal. + self.maybe_request_metadata_repair(consensus, peer).await; + } + // Undebounced and uncapped, like the partition walk arm. Follower + // only: a backup's `commit_journal` ships no wire replies, while a + // stranded primary is `resume_stranded_commits`' job above, which + // does. Not gated on `RecoveryOwner` (repaired prepares are + // journaled without walking, so gating parks the walk all + // session); `is_walk_stalled` itself refuses mid-transfer, where a + // walk past the incoming `snapshot_seq` breaks the install. + if is_walk_stalled(&probe) && consensus.is_follower() { + tracing::info!( + shard = self.id, + commit_min = probe.commit_min, + commit_max = probe.commit_max, + "metadata commit walk parked over resident committed ops; resuming" + ); + metadata.commit_journal().await; + } + } + // Stall retry for an in-flight state transfer: descriptor or chunk // frames are fire-and-forget, so a lost one must not wedge the // session (and the boot flow behind it) forever. @@ -8907,19 +9009,20 @@ fn repair_serve_ceiling(requested_to_op: u64, commit_max: u64, head: u64) -> u64 /// tick. Over-cap groups stay due and arm on a later pass. const PARTITION_REPAIR_ARMS_PER_TICK_MAX: usize = 3; -/// What already owns a partition's recovery, if anything. +/// What already owns a group's recovery, if anything. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum RecoveryOwner { - /// Nothing owns it: the sweep may arm repair. + /// Nothing owns it: the tick may arm repair. Nobody, Repair, Transfer, /// A scheduled transfer re-arm counting down. Arming repair over it would - /// defeat its backoff, as `arm_partition_transfer` documents. + /// defeat its backoff, as `arm_partition_transfer` documents. Partition + /// plane only; metadata has no re-arm state. TransferRearm, } -/// What the tick sweep reads off one partition to decide whether it is +/// What a tick driver reads off one consensus group to decide whether it is /// gap-stopped. Split out so the guards, the debounce and the per-tick cap are /// testable without a shard, a bus, or a journal. #[derive(Debug, Clone, Copy)] @@ -8939,8 +9042,8 @@ struct GapProbe { /// The journal-hole half is not redundant: a follower advances `commit_max` /// from every prepare header in `replicate_preflight`, so `commit_min < /// commit_max` is transiently true on every healthy pipelined tick and a bare -/// lag test would arm repair against ordinary produce. -const fn partition_is_gap_stopped(probe: &GapProbe) -> bool { +/// lag test would arm repair against ordinary traffic. +const fn is_gap_stopped(probe: &GapProbe) -> bool { probe.normal && !probe.transferring && matches!(probe.recovery, RecoveryOwner::Nobody) @@ -8952,31 +9055,32 @@ const fn partition_is_gap_stopped(probe: &GapProbe) -> bool { /// just never ran (a heartbeat carrying a known commit is `Accepted`, and an /// idle group offers no other edge). Not gated on `RecoveryOwner`: repair /// fetches bodies without walking them, so gating parks the walk all session. -const fn partition_is_walk_stalled(probe: &GapProbe) -> bool { +const fn is_walk_stalled(probe: &GapProbe) -> bool { probe.normal && !probe.transferring && probe.commit_min < probe.commit_max && probe.next_op_resident } -/// Count one sweep tick against `gap_ticks` and answer whether this partition -/// may arm repair now. +/// Count one tick against `gap_ticks` and answer whether this group may arm +/// repair now. /// /// Level-triggered, because every edge-triggered arming site is starvable: the /// commit-heartbeat backstop fires only on `CommitOutcome::Advanced`, and under -/// sustained produce the prepares consume the advance in preflight before the +/// sustained traffic the prepares consume the advance in preflight before the /// gap check drops them, so the heartbeat lands as `Accepted` and the gap wedges /// until an unrelated view change. /// /// A capped-out arm keeps its debounce satisfied rather than starting over, so -/// the group arms on the next pass with a slot free. -const fn drive_partition_gap_debounce( +/// the group arms on the next pass with a slot free. The cap only bounds the +/// partition sweep; the metadata driver has one group and passes zero arms. +const fn drive_gap_debounce( probe: &GapProbe, gap_ticks: &mut u32, debounce_ticks: u32, arms_this_tick: usize, ) -> bool { - if !partition_is_gap_stopped(probe) { + if !is_gap_stopped(probe) { *gap_ticks = 0; return false; } @@ -9018,6 +9122,25 @@ where } } +/// Map the metadata plane's recovery slots onto [`RecoveryOwner`]. Transfer +/// first, as in `partition_gap_probe`: it supersedes repair, so naming it is +/// the truthful diagnostic when both are set. The stage counts as a transfer +/// too, since `begin_state_transfer_await` owns the recovery before a session +/// exists. +const fn metadata_recovery_owner( + transfer_session: bool, + stage_transferring: bool, + repair_session: bool, +) -> RecoveryOwner { + if transfer_session || stage_transferring { + RecoveryOwner::Transfer + } else if repair_session { + RecoveryOwner::Repair + } else { + RecoveryOwner::Nobody + } +} + /// Whether the parked `StartView` log names every op in the uncommitted /// suffix `(commit_max, head]`, in descending order. Only this canonical list /// makes fetching bodies above the commit point safe. @@ -10254,15 +10377,16 @@ mod superblock_fail_stop_tests { #[cfg(test)] mod gap_detector_tests { - //! The level-triggered repair arm the partition tick sweep runs. + //! The level-triggered repair arm the partition and metadata tick drivers + //! share. //! //! Its whole reason to exist is that the edge-triggered arming sites are //! starvable, so the guards it shares with them and the debounce that keeps //! it off healthy traffic are the parts worth pinning. use super::{ - GapProbe, PARTITION_REPAIR_ARMS_PER_TICK_MAX, RecoveryOwner, drive_partition_gap_debounce, - partition_is_gap_stopped, partition_is_walk_stalled, + GapProbe, PARTITION_REPAIR_ARMS_PER_TICK_MAX, RecoveryOwner, drive_gap_debounce, + is_gap_stopped, is_walk_stalled, metadata_recovery_owner, }; const DEBOUNCE: u32 = 100; @@ -10300,8 +10424,8 @@ mod gap_detector_tests { next_op_resident: true, ..gap_stopped() }; - assert!(!partition_is_gap_stopped(&healthy)); - assert!(partition_is_gap_stopped(&gap_stopped())); + assert!(!is_gap_stopped(&healthy)); + assert!(is_gap_stopped(&gap_stopped())); } #[test] @@ -10310,7 +10434,7 @@ mod gap_detector_tests { commit_min: 10, ..gap_stopped() }; - assert!(!partition_is_gap_stopped(&caught_up)); + assert!(!is_gap_stopped(&caught_up)); } #[test] @@ -10321,13 +10445,13 @@ mod gap_detector_tests { normal: false, ..gap_stopped() }; - assert!(!partition_is_gap_stopped(&electing)); + assert!(!is_gap_stopped(&electing)); let installing = GapProbe { transferring: true, ..gap_stopped() }; - assert!(!partition_is_gap_stopped(&installing)); + assert!(!is_gap_stopped(&installing)); } #[test] @@ -10342,7 +10466,7 @@ mod gap_detector_tests { ..gap_stopped() }; assert!( - !partition_is_gap_stopped(&owned), + !is_gap_stopped(&owned), "{owner:?} owns the recovery; a second session would race it" ); } @@ -10354,16 +10478,11 @@ mod gap_detector_tests { let mut gap_ticks = 0; for tick in 1..DEBOUNCE { assert!( - !drive_partition_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, 0), + !drive_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, 0), "armed at tick {tick}, before the debounce elapsed" ); } - assert!(drive_partition_gap_debounce( - &probe, - &mut gap_ticks, - DEBOUNCE, - 0 - )); + assert!(drive_gap_debounce(&probe, &mut gap_ticks, DEBOUNCE, 0)); } #[test] @@ -10375,28 +10494,23 @@ mod gap_detector_tests { }; let mut gap_ticks = 0; for _ in 0..DEBOUNCE - 1 { - drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0); + drive_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0); } assert_eq!(gap_ticks, DEBOUNCE - 1); - assert!(!drive_partition_gap_debounce( - &walkable, - &mut gap_ticks, - DEBOUNCE, - 0 - )); + assert!(!drive_gap_debounce(&walkable, &mut gap_ticks, DEBOUNCE, 0)); assert_eq!(gap_ticks, 0, "progress must restart the debounce"); assert!( - !drive_partition_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0), + !drive_gap_debounce(&stopped, &mut gap_ticks, DEBOUNCE, 0), "a fresh gap must serve its own debounce, not inherit the old count" ); } #[test] fn given_a_follower_with_resident_committed_ops_when_probed_should_be_walk_stalled() { - assert!(partition_is_walk_stalled(&walk_stalled())); + assert!(is_walk_stalled(&walk_stalled())); assert!( - !partition_is_walk_stalled(&gap_stopped()), + !is_walk_stalled(&gap_stopped()), "a missing next op is repair's job; a walk over it would stop dead" ); } @@ -10407,7 +10521,7 @@ mod gap_detector_tests { commit_min: 10, ..walk_stalled() }; - assert!(!partition_is_walk_stalled(&caught_up)); + assert!(!is_walk_stalled(&caught_up)); } #[test] @@ -10416,7 +10530,7 @@ mod gap_detector_tests { normal: false, ..walk_stalled() }; - assert!(!partition_is_walk_stalled(&electing)); + assert!(!is_walk_stalled(&electing)); // Same gate as the on-commit arm: a walk during a transfer can advance // commit_min past the incoming frontier. @@ -10424,7 +10538,7 @@ mod gap_detector_tests { transferring: true, ..walk_stalled() }; - assert!(!partition_is_walk_stalled(&installing)); + assert!(!is_walk_stalled(&installing)); } #[test] @@ -10442,7 +10556,7 @@ mod gap_detector_tests { ..walk_stalled() }; assert!( - partition_is_walk_stalled(&owned), + is_walk_stalled(&owned), "{owner:?} owns the fetch, not the resident prefix" ); } @@ -10472,8 +10586,7 @@ mod gap_detector_tests { next_op_resident, }; assert!( - !(partition_is_gap_stopped(&probe) - && partition_is_walk_stalled(&probe)), + !(is_gap_stopped(&probe) && is_walk_stalled(&probe)), "both predicates claim {probe:?}" ); } @@ -10488,7 +10601,7 @@ mod gap_detector_tests { let probe = gap_stopped(); let mut gap_ticks = DEBOUNCE; assert!( - !drive_partition_gap_debounce( + !drive_gap_debounce( &probe, &mut gap_ticks, DEBOUNCE, @@ -10502,7 +10615,7 @@ mod gap_detector_tests { arm a whole interval out per contended tick" ); assert!( - drive_partition_gap_debounce( + drive_gap_debounce( &probe, &mut gap_ticks, DEBOUNCE, @@ -10511,4 +10624,34 @@ mod gap_detector_tests { "the same group arms on the next pass with a slot free" ); } + + #[test] + fn given_metadata_recovery_slots_when_mapped_should_name_transfer_over_repair() { + // (transfer session, stage transferring, repair session), as + // `metadata_gap_probe` reads them off `metadata_transfer`, the + // consensus stage and `metadata_repair`. + assert_eq!( + metadata_recovery_owner(false, false, false), + RecoveryOwner::Nobody + ); + assert_eq!( + metadata_recovery_owner(false, false, true), + RecoveryOwner::Repair + ); + // A stage past `Idle` owns the recovery before any session exists. + assert_eq!( + metadata_recovery_owner(false, true, false), + RecoveryOwner::Transfer + ); + assert_eq!( + metadata_recovery_owner(true, false, false), + RecoveryOwner::Transfer + ); + assert_eq!( + metadata_recovery_owner(true, true, true), + RecoveryOwner::Transfer, + "a transfer supersedes a lingering repair session; naming Repair \ + would read as the tick may re-arm it" + ); + } } diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index cefc3d4458..29f0a47293 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -200,6 +200,7 @@ pub struct ShardMetrics { partition_requests_denied_transient_total: Counter, partition_repair_serves_deferred_purge_total: Counter, partition_prepare_gap_drops_total: Counter, + metadata_prepare_gap_drops_total: Counter, } impl ShardMetrics { @@ -225,6 +226,7 @@ impl ShardMetrics { partition_requests_denied_transient_total: Counter::default(), partition_repair_serves_deferred_purge_total: Counter::default(), partition_prepare_gap_drops_total: Counter::default(), + metadata_prepare_gap_drops_total: Counter::default(), } } @@ -417,6 +419,23 @@ impl ShardMetrics { self.partition_prepare_gap_drops_total.get() } + /// Add the prepares the metadata backup gap check destroyed since the last + /// tick, drained from `IggyMetadata::take_prepare_gap_drops`. A sibling of + /// `partition_prepare_gap_drops_total`, and NOT a `frame_drops_total` + /// reason for the same cause: gap drops are protocol-ordering drops that + /// the tick driver repairs, not routing faults, and the simulator asserts + /// `frame_drops_total` stays at zero on runs with no injected loss. + pub fn record_metadata_prepare_gap_drops(&self, drops: u64) { + self.metadata_prepare_gap_drops_total.inc_by(drops); + } + + /// Snapshot of `metadata_prepare_gap_drops_total`. Test/simulator accessor. + #[cfg(any(test, feature = "simulator"))] + #[must_use] + pub fn metadata_prepare_gap_drops_value(&self) -> u64 { + self.metadata_prepare_gap_drops_total.get() + } + /// Snapshot of `partition_frames_rejected_stale_total`. Test/simulator /// accessor, readable from any crate under those cfgs so the crates that /// drive the reconciler can assert a reject did not happen. @@ -513,6 +532,11 @@ impl ShardMetrics { "replicated prepares dropped out of order by a backup's gap check", self.partition_prepare_gap_drops_total.clone(), ); + registry.register( + "metadata_prepare_gap_drops", + "replicated metadata prepares dropped out of order by a backup's gap check", + self.metadata_prepare_gap_drops_total.clone(), + ); } } diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 44421de461..44abeb4983 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -4740,9 +4740,10 @@ mod partition_repair_driver_tests { /// the healer. const STRAND_QUIET_STEPS: usize = 300; - /// The partition-plane prepare a packet carries, if it carries one for - /// `group`. - fn prepare_for(packet: &Packet, group: u64) -> Option { + /// The prepare a packet carries, if it carries one for `group`. Shared + /// with `metadata_repair_driver_tests`, as are the three below and + /// [`cluster`]: both planes ride the same commands, keyed by group. + pub fn prepare_for(packet: &Packet, group: u64) -> Option { if packet.message.header().command != Command::Prepare { return None; } @@ -4752,7 +4753,7 @@ mod partition_repair_driver_tests { } /// Whether a packet is a commit heartbeat for `group`. - fn is_commit_for(packet: &Packet, group: u64) -> bool { + pub fn is_commit_for(packet: &Packet, group: u64) -> bool { if packet.message.header().command != Command::Commit { return false; } @@ -4762,7 +4763,7 @@ mod partition_repair_driver_tests { } /// Whether a packet is a repair request for `group`. - fn is_request_prepares_for(packet: &Packet, group: u64) -> bool { + pub fn is_request_prepares_for(packet: &Packet, group: u64) -> bool { if packet.message.header().command != Command::RequestPrepares { return false; } @@ -4773,7 +4774,7 @@ mod partition_repair_driver_tests { } /// Whether a packet is a repair stream terminator for `group`. - fn is_repair_done_for(packet: &Packet, group: u64) -> bool { + pub fn is_repair_done_for(packet: &Packet, group: u64) -> bool { if packet.message.header().command != Command::RepairDone { return false; } @@ -4783,7 +4784,7 @@ mod partition_repair_driver_tests { header.group == group } - fn cluster(seed: u64) -> (Simulator, SimClient) { + pub fn cluster(seed: u64) -> (Simulator, SimClient) { server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), @@ -5300,3 +5301,511 @@ mod partition_repair_driver_tests { ); } } + +#[cfg(test)] +mod metadata_repair_driver_tests { + //! A backup that missed a committed metadata prepare recovers in Normal + //! status, without waiting for a view change. + //! + //! The metadata twin of `partition_repair_driver_tests`, driven by the + //! detector in `tick_metadata`: the same preflight `commit_max` advance + //! starves the `Advanced`-gated arm in `on_commit`, and + //! `retry_stalled_metadata_repair` re-drives only a session that already + //! exists. Every fault here is keyed on `METADATA_GROUP`, since both + //! planes share `Prepare`/`Commit` on the same links. + + use super::partition_repair_driver_tests::{ + cluster, is_commit_for, is_repair_done_for, is_request_prepares_for, prepare_for, + }; + use super::*; + use consensus::Status; + use journal::Journal; + use packet::Packet; + use server_common::sharding::METADATA_GROUP; + use std::sync::atomic::{AtomicU64, Ordering}; + + /// Chain replication runs 0 -> 1 -> 2 and stops before the primary, so + /// replica 2 is the only one whose losses cannot starve the group of + /// quorum (see `partition_repair_driver_tests::LAGGING`). + const LAGGING: u8 = 2; + + const CLIENT_ID: u128 = 1; + + /// Ops committed cleanly before the fault, so the gap opens above a + /// committed prefix rather than at the group's first op. + const WARMUP_SENDS: usize = 3; + + /// Ticks stepped after each stream creation, one round trip's worth. + const STEPS_PER_SEND: usize = 12; + + /// Creations issued with the fault standing in the gap test. Long enough + /// that prepares keep consuming the `commit_max` advance the heartbeat + /// backstop needs; the debounce may elapse mid-produce, which the verdict + /// tolerates (the withheld heartbeats mean only the tick driver can arm). + const GAP_SENDS: usize = 12; + + /// Creations issued with the fault standing in the eviction and + /// walk-starvation tests: few enough (under `partitions::REPAIR_RETRY_TICKS` + /// ticks) that the debounce fires only after produce stops, so the floor + /// stamp / the starved walk edge is in place before the arm runs. + const SHORT_GAP_SENDS: usize = 3; + + /// Quiet ticks for the repair stream to land, kept under + /// `NORMAL_HEARTBEAT_TICKS` (500) so no election can be the healer. + const QUIET_STEPS: usize = 160; + + /// Budget for the group to settle once the fault is lifted; the drain + /// loop breaks on convergence. + const DRAIN_STEPS: usize = 600; + + /// Quiet budget for the walk-starvation test: debounce, repair stream, + /// then the drain, still under `NORMAL_HEARTBEAT_TICKS`. + const STRAND_QUIET_STEPS: usize = 300; + + /// Ticks of healthy load in the no-false-positive test, several debounce + /// intervals' worth so the driver gets many chances to arm. + const LOAD_TICKS: usize = 4 * partitions::REPAIR_RETRY_TICKS as usize; + + /// Paced at a fraction of a round trip; faster submission only collects + /// transient rejections once the prepare pipeline fills. + const TICKS_PER_SEND: usize = 4; + + /// Ops the healthy run must have committed for its verdict to mean + /// anything: enough to prove the group was live across several debounce + /// intervals, not that it was saturated. + const COMMITTED_MIN: u64 = 20; + + /// `(status, view, commit_min, commit_max)` of one replica's metadata group. + fn metadata_state(sim: &Simulator, replica: u8) -> (Status, u32, u64, u64) { + let metadata = sim.replicas[replica as usize].shards[0].plane.metadata(); + let consensus = metadata + .consensus + .as_ref() + .expect("shard 0 owns metadata consensus"); + ( + consensus.status(), + consensus.view(), + consensus.commit_min(), + consensus.commit_max(), + ) + } + + #[allow(clippy::cast_possible_truncation)] + fn journal_holds(sim: &Simulator, replica: u8, op: u64) -> bool { + sim.replicas[replica as usize] + .metadata_journal + .header(op as usize) + .is_some() + } + + fn gap_drops(sim: &Simulator, replica: u8) -> u64 { + sim.replicas[replica as usize].shards[0] + .metrics() + .metadata_prepare_gap_drops_value() + } + + fn transfer_armed(sim: &Simulator, replica: u8) -> bool { + let metadata = sim.replicas[replica as usize].shards[0].plane.metadata(); + metadata.consensus.as_ref().is_some_and(|consensus| { + consensus.state_transfer_stage() != consensus::StateTransferStage::Idle + }) + } + + /// Submit `sends` stream creations to the primary, stepping between each. + fn create_streams(sim: &mut Simulator, client: &SimClient, sends: usize, tag: &str) { + for index in 0..sends { + let msg = client.create_stream(&format!("{tag}-{index}")); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + for _ in 0..STEPS_PER_SEND { + sim.step(); + } + } + } + + #[test] + fn given_a_backup_that_dropped_a_committed_metadata_prepare_when_heartbeat_advances_are_starved_should_repair_in_normal_status() + { + // Statics, not captures: the link hooks are bare `fn` pointers, + // declared inside the test so parallel siblings cannot share them. + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first metadata prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, METADATA_GROUP) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold this group's commit heartbeats, so the + /// `Advanced` backstop can never run, and withhold retransmits of the + /// dropped op. The retransmit half stands in for production behaviour: + /// `consensus::retransmit_targets` skips an op that already reached + /// quorum, and this op reaches quorum on 0 and 1 alone. + fn starve_commit_edge(packet: &Packet) -> bool { + if let Some(header) = prepare_for(packet, METADATA_GROUP) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, METADATA_GROUP) + } + + let (mut sim, client) = cluster(0x5EED_0240); + sim.register_client_with_primary(&client); + WITHHELD_OP.store(0, Ordering::Relaxed); + + create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm"); + let (_, _, warm_commit_min, _) = metadata_state(&sim, LAGGING); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + create_streams(&mut sim, &client, GAP_SENDS, "md-gap"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no metadata prepare crossed the chain link, so the fault never armed" + ); + assert!( + gap_drops(&sim, LAGGING) > 0, + "the lagging replica never reached its backup gap check, so the \ + prepares after the withheld op were not dropped as a gap" + ); + + for _ in 0..QUIET_STEPS { + sim.step(); + } + + // Judged with the blockade still standing: no commit heartbeat for + // this group has reached the replica since the gap opened, so only + // the tick driver can have armed the repair. + let (status, view, commit_min, _) = metadata_state(&sim, LAGGING); + assert_eq!( + view, 0, + "a view change healed the gap instead of the repair driver; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + assert!( + journal_holds(&sim, LAGGING, withheld), + "op {withheld} was never repaired back into the lagging replica's WAL" + ); + assert!( + commit_min >= withheld, + "the commit walk never crossed the repaired hole: stopped at \ + {commit_min}, the withheld op is {withheld}" + ); + + // Lift the blockade and let the group settle; the tail above the + // repaired window waits on the heartbeats the fault withheld. + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = None; + for _ in 0..DRAIN_STEPS { + sim.step(); + let (_, _, commit_min, commit_max) = metadata_state(&sim, LAGGING); + if commit_min == commit_max { + break; + } + } + let (status, view, commit_min, commit_max) = metadata_state(&sim, LAGGING); + assert_eq!((status, view), (Status::Normal, 0)); + assert_eq!( + commit_min, commit_max, + "the lagging replica is still gap-stopped: committed through \ + {commit_max} but walkable only to {commit_min}" + ); + } + + #[test] + fn given_a_metadata_repair_armed_by_the_tick_driver_when_the_floor_is_evicted_should_convert_to_state_transfer() + { + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, METADATA_GROUP) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + fn starve_commit_edge(packet: &Packet) -> bool { + if let Some(header) = prepare_for(packet, METADATA_GROUP) { + return header.op == WITHHELD_OP.load(Ordering::Relaxed); + } + is_commit_for(packet, METADATA_GROUP) + } + + let (mut sim, client) = cluster(0x5EED_0241); + sim.register_client_with_primary(&client); + WITHHELD_OP.store(0, Ordering::Relaxed); + + create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm"); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_commit_edge); + + create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-gap"); + assert_ne!( + WITHHELD_OP.load(Ordering::Relaxed), + 0, + "no metadata prepare crossed the chain link, so the fault never armed" + ); + + // Move the primary's retention floor past the whole gap window before + // the debounce can arm (`SHORT_GAP_SENDS`): the serve path reads only + // the snapshot watermark, so the request is answered `RangeEvicted` + // (see `stamp_metadata_snapshot`). + let primary_commit_min = metadata_state(&sim, 0).2; + sim.stamp_metadata_snapshot(0, primary_commit_min); + + for _ in 0..QUIET_STEPS { + sim.step(); + if transfer_armed(&sim, LAGGING) { + break; + } + } + + let (status, view, ..) = metadata_state(&sim, LAGGING); + assert_eq!( + view, 0, + "a view change armed the recovery instead of the tick-armed repair session" + ); + assert!( + transfer_armed(&sim, LAGGING), + "the tick-armed repair session hit an evicted floor but never converted \ + to a state transfer (status {status:?})" + ); + } + + #[test] + fn given_a_backup_holding_resident_committed_metadata_ops_when_every_walk_edge_is_starved_should_drain_in_normal_status() + { + static WITHHELD_OP: AtomicU64 = AtomicU64::new(0); + static WITHHELD_DONES: AtomicU64 = AtomicU64::new(0); + + /// Chain link 1 -> 2: swallow the first metadata prepare, once. + fn withhold_one_prepare(packet: &Packet) -> bool { + let Some(header) = prepare_for(packet, METADATA_GROUP) else { + return false; + }; + WITHHELD_OP + .compare_exchange(0, header.op, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + } + + /// Primary -> 2: withhold every direct prepare (live ones ride the + /// chain, so this starves only retransmit heals), the group's commit + /// heartbeats, and its repair terminators. The repaired ops themselves + /// pass, so the window lands resident while the walk `RepairDone` + /// would run never fires. + fn starve_walk_edges(packet: &Packet) -> bool { + if prepare_for(packet, METADATA_GROUP).is_some() { + return true; + } + if is_repair_done_for(packet, METADATA_GROUP) { + WITHHELD_DONES.fetch_add(1, Ordering::Relaxed); + return true; + } + is_commit_for(packet, METADATA_GROUP) + } + + let (mut sim, client) = cluster(0x5EED_0242); + sim.register_client_with_primary(&client); + WITHHELD_OP.store(0, Ordering::Relaxed); + WITHHELD_DONES.store(0, Ordering::Relaxed); + + create_streams(&mut sim, &client, WARMUP_SENDS, "md-warm"); + let (_, _, warm_commit_min, _) = metadata_state(&sim, LAGGING); + assert!( + warm_commit_min > 0, + "the lagging replica committed nothing before the fault, so the gap \ + below would open at the group's first op" + ); + + *sim.network + .link_drop_packet_fn(ProcessId::Replica(1), ProcessId::Replica(LAGGING)) = + Some(withhold_one_prepare); + *sim.network + .link_drop_packet_fn(ProcessId::Replica(0), ProcessId::Replica(LAGGING)) = + Some(starve_walk_edges); + + create_streams(&mut sim, &client, SHORT_GAP_SENDS, "md-strand"); + + let withheld = WITHHELD_OP.load(Ordering::Relaxed); + assert_ne!( + withheld, 0, + "no metadata prepare crossed the chain link, so the fault never armed" + ); + + for _ in 0..STRAND_QUIET_STEPS { + sim.step(); + let (_, view, commit_min, commit_max) = metadata_state(&sim, LAGGING); + if view != 0 || (commit_min >= withheld && commit_min == commit_max) { + break; + } + } + + assert!( + WITHHELD_DONES.load(Ordering::Relaxed) > 0, + "no repair terminator was withheld, so the walk was never starved and \ + a green run would not prove the tick backstop" + ); + assert!( + journal_holds(&sim, LAGGING, withheld), + "op {withheld} was never repaired back into the lagging replica's WAL" + ); + let (status, view, commit_min, commit_max) = metadata_state(&sim, LAGGING); + assert_eq!( + view, 0, + "a view change drained the walk instead of the tick backstop; the test \ + proves nothing about normal status" + ); + assert_eq!(status, Status::Normal, "the replica left Normal status"); + for op in commit_min + 1..=commit_max { + assert!( + journal_holds(&sim, LAGGING, op), + "op {op} is not resident, so this run stranded on a repair gap, \ + not a parked walk" + ); + } + assert_eq!( + commit_min, commit_max, + "the walk never resumed over resident committed ops: walkable to \ + {commit_min}, committed through {commit_max}, every op between resident" + ); + } + + /// Regression canary, expected green even without the tick driver: a + /// healthy metadata backup walks at every accepted prepare's tail + /// (`on_replicate`), so per-tick lag never survives to quiescence and the + /// journal-hole half of the predicate is pinned by `gap_detector_tests` + /// instead. What this run pins is that the driver stays silent under + /// sustained pipelined load. + #[test] + fn given_healthy_metadata_traffic_when_no_gap_exists_should_not_arm_repair() { + static REPAIR_REQUESTS: AtomicU64 = AtomicU64::new(0); + + /// Observer, not a fault: counts this group's repair requests and + /// passes every packet through. + fn count_repair_requests(packet: &Packet) -> bool { + if is_request_prepares_for(packet, METADATA_GROUP) { + REPAIR_REQUESTS.fetch_add(1, Ordering::Relaxed); + } + false + } + + // TWO replicas, as in the partition twin: quorum spans both, so no op + // can commit while the backup misses it, and every reordering-induced + // gap blocks quorum until retransmit refills it. + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { + enabled: false, + size: iggy_common::IggyByteSize::from(0u64), + bucket_capacity: 1, + }); + let replica_count: u8 = 2; + let mut sim = Simulator::new( + replica_count as usize, + std::iter::once(CLIENT_ID), + packet::PacketSimulatorOptions { + node_count: replica_count, + client_count: 1, + seed: 0x5EED_0243, + ..packet::PacketSimulatorOptions::default() + }, + ); + let client = SimClient::new(CLIENT_ID); + sim.register_client_with_primary(&client); + REPAIR_REQUESTS.store(0, Ordering::Relaxed); + + for (from, to) in [(0u8, 1u8), (1, 0)] { + *sim.network + .link_drop_packet_fn(ProcessId::Replica(from), ProcessId::Replica(to)) = + Some(count_repair_requests); + } + + // Sustained, not bursty, and the run asserts the load went somewhere, + // so a green result cannot come from a workload that never loaded the + // group. + let mut lag_run = 0u32; + let mut longest_lag_run = 0u32; + let sample = |sim: &Simulator, lag_run: &mut u32, longest: &mut u32| { + let (_, _, commit_min, commit_max) = metadata_state(sim, 1); + if commit_min < commit_max { + *lag_run += 1; + *longest = (*longest).max(*lag_run); + } else { + *lag_run = 0; + } + }; + for tick in 0..LOAD_TICKS { + if tick % TICKS_PER_SEND == 0 { + let msg = client.create_stream(&format!("healthy-{tick}")); + sim.submit_request(client.client_id(), 0, msg.into_generic()); + } + sim.step(); + sample(&sim, &mut lag_run, &mut longest_lag_run); + } + for _ in 0..QUIET_STEPS { + sim.step(); + sample(&sim, &mut lag_run, &mut longest_lag_run); + } + + let committed = metadata_state(&sim, 1).2; + let sends = LOAD_TICKS / TICKS_PER_SEND; + assert!( + committed >= COMMITTED_MIN, + "the backup committed only {committed} ops across {sends} sends, so the \ + driver was never ticked over a loaded group" + ); + // Recorded, not load-bearing: the tail walk in `on_replicate` keeps a + // healthy backup caught up at quiescence, so a naive lag detector has + // nothing to misread here; `gap_detector_tests` pins the predicate. + assert_eq!( + longest_lag_run, 0, + "healthy two-replica metadata traffic left the backup lagging for \ + {longest_lag_run} consecutive ticks; if this ever becomes non-zero \ + the predicate's journal-hole half is load-bearing HERE too and this \ + test should assert against it rather than record it" + ); + for replica in 0..replica_count { + let (status, view, commit_min, commit_max) = metadata_state(&sim, replica); + assert_eq!( + (status, view), + (Status::Normal, 0), + "replica {replica} left view 0 / Normal, so a view change could \ + account for repair traffic" + ); + if commit_min < commit_max { + assert!( + journal_holds(&sim, replica, commit_min + 1), + "replica {replica} lags at {commit_min} of {commit_max} with op \ + {} missing, so a no-loss run produced a real hole", + commit_min + 1 + ); + } + } + assert_eq!( + REPAIR_REQUESTS.load(Ordering::Relaxed), + 0, + "the tick driver requested repair on a healthy group; its gap predicate \ + is reading ordinary commit lag as a journal hole" + ); + } +}