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 3f741c4ba2ca57bbca37be346ed826440f1ccc8b 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 e6262e94e5893bf55f11f158d93731c9c41df00d Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Wed, 2 Sep 2026 15:35:43 +0200 Subject: [PATCH 4/4] review fixes --- .../tests/cluster/parked_frame_redispatch.rs | 19 +- core/partitions/src/iggy_partition.rs | 6 +- core/partitions/src/journal.rs | 14 + core/server/config.toml | 6 + core/server/src/partition_reconciler.rs | 17 +- core/shard/src/lib.rs | 338 +++++++++++++----- core/shard/src/metrics.rs | 6 + core/shard/src/router.rs | 162 ++++++--- core/simulator/src/lib.rs | 82 +++-- 9 files changed, 472 insertions(+), 178 deletions(-) diff --git a/core/integration/tests/cluster/parked_frame_redispatch.rs b/core/integration/tests/cluster/parked_frame_redispatch.rs index 2d2240f153..12ccec66f9 100644 --- a/core/integration/tests/cluster/parked_frame_redispatch.rs +++ b/core/integration/tests/cluster/parked_frame_redispatch.rs @@ -23,8 +23,9 @@ //! that parked. This test pins the park path positively, and on the replica //! where getting it wrong 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. +//! replicated PREPARE has no client behind it: nothing re-sends it once its op +//! has quorum, so the backup gap-stops and waits out `tick_partitions`' repair +//! debounce before anything refetches it. //! //! What makes the window wide on a backup is the commit broadcast. A backup //! learns a metadata commit from the `commit` field of the next prepare on that @@ -101,7 +102,16 @@ const DEGRADED_MARKERS: [&str; 3] = [ /// 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)"; +/// +/// Names its plane: `stdout_occurrences` counts SUBSTRINGS, and the metadata +/// plane logs its own gap drop at `warn`, which passes this test's `info` +/// filter. A shared wording would fail a partition-plane assertion on a +/// metadata-plane event. +/// +/// Deliberately not scoped to one namespace or node. Chain replication is +/// ordered per connection, so on a healthy three-node cluster with no injected +/// loss no partition-plane gap drop is expected anywhere. +const GAP_MARKER: &str = "dropping out-of-order partition prepare (gap)"; fn topic_name(index: u32) -> String { format!("parked-redispatch-topic-{index}") @@ -344,7 +354,8 @@ fn assert_no_degraded_park_paths(harness: &TestHarness) { 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" + position, and the op it displaced is recoverable only by waiting out the \ + repair driver's debounce" ); } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 4ece3674b0..b1b3aaaced 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -1014,7 +1014,7 @@ where /// 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) { + const fn note_prepare_gap_drop(&mut self) { self.prepare_gap_drops = self.prepare_gap_drops.saturating_add(1); } @@ -2305,7 +2305,7 @@ where return; } - let journal_holds_op = self.log.journal().inner.header_by_op(header.op).is_some(); + let journal_holds_op = self.log.journal().inner.holds_op(header.op); if journal_holds_op { // Retransmit after downstream flap: durable here but commit // hasn't caught up. Re-forward + re-ACK so primary's view of @@ -2401,7 +2401,7 @@ where tracing::Level::WARN, &PartitionDiagEvent::new( self.diag_ctx(), - "dropping out-of-order prepare (gap)", + "dropping out-of-order partition prepare (gap)", ) .with_operation(header.operation) .with_op(header.op), diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs index fcb3a9af78..5cfbd50e7c 100644 --- a/core/partitions/src/journal.rs +++ b/core/partitions/src/journal.rs @@ -762,6 +762,20 @@ where headers.iter().find(|header| header.op == op).copied() } + /// Whether `op` is resident, without reading its header. + /// + /// Equivalent to `header_by_op(op).is_some()` and answers in O(log n) + /// instead of scanning: `append` writes `headers` and + /// `op_to_storage_offset` together and every clear site + /// ([`Self::commit`], [`Self::evict_prefix`], the restore path) clears + /// both, so the two populations cannot diverge. Callers that only need + /// presence must use this - a miss is the common case on the residency + /// checks, and a miss is exactly when the scan walks the whole vec. + pub fn holds_op(&self, op: u64) -> bool { + let op_to_storage_offset = unsafe { &*self.op_to_storage_offset.get() }; + op_to_storage_offset.contains_key(&op) + } + /// Presence and message-carrying shape of the repair window `(floor, to_op]` /// in ONE pass over the header vec. /// diff --git a/core/server/config.toml b/core/server/config.toml index 583df57320..166d625d9b 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -646,6 +646,12 @@ view_probe_attempts_max = 5 # fire-and-forget over the lossy bus, so a session with no retry wedges forever # on a single dropped frame. Paces both the metadata and partition repair loops; # must be nonzero. +# +# Also how long a gap-stopped partition backup waits before OPENING a repair +# session, so shortening this makes the shard react to a replication hole +# sooner as well. That second use carries its own floor +# (PARTITION_GAP_DEBOUNCE_TICKS_MIN), because a value near the tick interval +# would otherwise arm repair against ordinary pipelining. repair_retry_interval = "1s" # Prepares a peer serves per repair round before the requester walks to the next diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 93ca117289..cc875a4579 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -50,9 +50,11 @@ //! 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. +//! is the pump's: staged frames get their own `select` arm, ranked above the +//! inbox, so the inbox is not read while any is outstanding and a parked op is +//! never ordered behind a later op of the same partition already queued there, +//! which the plane's gap check would drop. One frame per poll, so the +//! consensus tick still preempts between them. //! - `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 @@ -3609,10 +3611,11 @@ mod tests { drop(inbox); } - /// The replicated-prepare shape, which no other test covers and where both - /// park critical are worst: a prepare has no client, so `deny_parked_frame` - /// no-ops on it and anything that discards it loses committed data silently, - /// with no normal-status repair driver to refetch it. + /// The replicated-prepare shape, which no other test covers and where the + /// park path's stakes are highest: a prepare has no client, so + /// `deny_parked_client_request` no-ops on it and anything that discards it + /// loses committed data silently, recoverable only once `tick_partitions`' + /// repair driver notices the gap it left. /// /// A backup receives the prepare before its own metadata commits (so the frame /// parks unstamped), then applies the commit and materialises. The prepare must diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index c1bff0dc07..78acedceab 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -2556,15 +2556,76 @@ where self.dispatch_message(message, None).await; } - /// Deliver what `redispatch_parked_frames` staged, in park order. + /// Whether anything is staged for the pump to re-deliver. + #[must_use] + pub(crate) fn has_redispatched_frames(&self) -> bool { + !self.redispatch_queue.borrow().is_empty() + } + + /// Deliver ONE staged frame, in park order. `false` when none was staged. /// - /// 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. + /// The pump serves this from a select arm ranked above the inbox, so the + /// inbox cannot be read while anything is staged: a parked op reaches the + /// plane ahead of a later op of the same partition already queued there, + /// which the plane's backup gap check would drop. One frame per call is + /// what lets the consensus tick preempt between frames; draining to empty + /// here would hold the pump for the whole run. /// /// 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(crate) async fn dispatch_one_redispatched_frame(&self) -> bool + 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, + { + let Some(frame) = self.redispatch_queue.borrow_mut().pop_front() else { + return false; + }; + 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" + ); + return true; + } + }; + self.dispatch_message(bag, Some(provenance)).await; + true + } + + /// Deliver everything staged, in park order. + /// + /// Shutdown only. The pump's steady state serves the queue one frame at a + /// time from its own select arm ([`Self::dispatch_one_redispatched_frame`]) + /// so the tick can preempt; once the select is gone there is nothing left + /// to preempt for, and the alternative is + /// [`Self::retire_redispatched_frames`] destroying committed prepares a + /// stopping node still owes its group. + #[allow(clippy::future_not_send)] pub async fn drain_redispatched_frames(&self) where B: MessageBus + 'static, @@ -2581,38 +2642,13 @@ where >, 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; - } + while self.dispatch_one_redispatched_frame().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) { + pub(crate) fn retire_redispatched_frames(&self) { let staged: Vec = self.redispatch_queue.borrow_mut().drain(..).collect(); if staged.is_empty() { return; @@ -2973,7 +3009,8 @@ where /// replicates as soon as its own metadata commits, well before a lagging /// backup applies the same commit. Treating that as "prior incarnation" /// destroys live traffic: a replicated prepare has no client to answer, so - /// it would be dropped with no recovery until an unrelated view change. + /// it would be dropped and the backup left gap-stopped until + /// `tick_partitions`' level-triggered driver notices and repairs it. /// The residual is unchanged from before the stamp existed -- a frame parked /// while the namespace was absent, then recreated under a new incarnation, /// is served against the replacement -- and closing it needs a wire-level @@ -3082,8 +3119,9 @@ where /// 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. + /// [`Self::parked_frame_count`]: production branches on emptiness alone + /// ([`Self::has_redispatched_frames`], which arms the pump's delivery arm), + /// never on a depth. #[cfg(any(test, feature = "simulator"))] #[must_use] pub fn redispatched_frame_count(&self) -> usize { @@ -3222,11 +3260,11 @@ where let existing = pending.get_mut(&namespace); let parked_len = existing.as_ref().map_or(0, |entry| entry.frames.len()); let namespace_bytes = existing.as_ref().map_or(0, |entry| entry.bytes); - // A prepare is never shed on a byte budget. No client to answer, and no - // recovery: `consensus::retransmit_targets` skips an op that already - // reached quorum and the plane opens a repair session only in - // `on_start_view`, so shedding one is permanent loss where shedding a - // request costs a retry. A request is refused the moment admitting it + // A prepare is never shed on a byte budget. No client to answer, and + // recovery is slow: `consensus::retransmit_targets` skips an op that + // already reached quorum, so shedding one gap-stops the backup until + // `tick_partitions`' driver repairs it, where shedding a request costs + // one retry. A request is refused the moment admitting it // would cross a budget; a prepare only once one is already spent. Caps // prepare residency at one frame of overshoot per budget (worst case // `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead of @@ -6498,9 +6536,24 @@ where // Repair sessions this sweep has opened, against // `PARTITION_REPAIR_ARMS_PER_TICK_MAX`. let mut repair_arms = 0usize; + // Commit walks this sweep has run, against + // `PARTITION_WALKS_PER_TICK_MAX`. + let mut walks = 0usize; let mut fatal: Option = None; for namespace in namespace_scratch.drain(..) { + { + // Ahead of the fence check and every `continue` below: the + // count is the only record those prepares existed, and a + // partition that fences here never ticks again. + 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 Some(partition) = partitions.get_by_ns(&namespace) else { continue; }; @@ -6636,55 +6689,74 @@ where // 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 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( + let walk_stalled = partition_is_walk_stalled(&probe); + if drive_partition_gap_debounce( &probe, &mut partition.gap_ticks, repair_retry_ticks, repair_arms, - ); - (gap_arm, partition_is_walk_stalled(&probe)) + ) { + let consensus = partition.consensus(); + let self_id = consensus.replica(); + let peer = consensus.primary_index(consensus.view()); + let commit_min = consensus.commit_min(); + let commit_max = consensus.commit_max(); + if peer == self_id { + // The primary is its own repair source, so there is + // nobody to ask: the send would fail (no self entry in + // the replica registry) AFTER the session was recorded, + // and a session nothing can answer blocks this detector + // and every edge-triggered arming site until a view + // change. Restart the debounce so the warning repeats at + // its interval rather than every tick. + partition.gap_ticks = 0; + tracing::warn!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min, + commit_max, + "partition primary is gap-stopped below its own commit frontier; \ + repair has no peer to request from" + ); + } else { + tracing::debug!( + shard = self.id, + namespace_raw = namespace.inner(), + commit_min, + commit_max, + peer, + recovery = ?probe.recovery, + "partition gap-stopped past the debounce; arming repair from the primary" + ); + self.maybe_request_partition_repair(partition, peer).await; + repair_arms += 1; + } + } + walk_stalled }; - 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 { + // Capped like the repair arm, and for the same reason: a node-wide + // rejoin leaves every group on the shard walk-stalled in the same + // tick, and each walk reaches a segment flush. Undebounced, though + // -- the predicate guarantees the walk finds at least the next op + // (it and `collect_committable_from_journal` read the same journal + // index), so it cannot spin, and a skipped group is re-evaluated + // unchanged next tick. + if walk_stalled && walks < PARTITION_WALKS_PER_TICK_MAX { let config = partitions.config(); let Some(partition) = partitions.get_mut_by_ns(&namespace) else { continue; }; let consensus = partition.consensus(); - tracing::info!( + // Debug, not info: an in-flight repair journals bodies without + // walking them, so this is the steady state for the whole + // duration of a rejoin and would be one line per group per tick. + tracing::debug!( shard = self.id, namespace_raw = namespace.inner(), commit_min = consensus.commit_min(), @@ -6692,6 +6764,7 @@ where "partition commit walk parked over resident committed ops; resuming" ); partition.commit_journal(config).await; + walks += 1; } // Transfer stall retry: descriptor and chunk frames are @@ -7587,6 +7660,15 @@ where if !consensus.is_normal() || consensus.is_transferring() || partition.repair.is_some() { return; } + // Never against self. The session is recorded below BEFORE the send, + // and a self-addressed `RequestPrepares` cannot be delivered (the + // replica registry holds no entry for this node), so the session would + // stand forever: `repair_finished` needs a `commit_min` only the reply + // can advance, the stall retry re-sends to the same peer, and + // `repair.is_some()` fences every other arming site meanwhile. + if peer == consensus.replica() { + return; + } // The window ends at the group head when suffix bodies are missing, // not at the commit point. A backup that adopted a StartView holds // suffix HEADERS above `commit_max` whose bodies it may never have @@ -7637,6 +7719,7 @@ where from_op, commit_to_op, fetch_to_op, + peer, "partition behind the group frontier; requesting repair" ); self.send_request_prepares( @@ -8900,13 +8983,41 @@ 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. +/// Repair sessions the partition tick sweep will OPEN per pass. +/// +/// A RATE cap, not a concurrency one: unlike +/// `IggyShard::PARTITION_TRANSFERS_INFLIGHT_MAX`, which counts live sessions +/// and refuses over the count, nothing bounds how many partition repair +/// sessions are live at once. This only spreads the opening cost, because 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; +/// Commit walks the partition tick sweep will RUN per pass. +/// +/// Same correlated-fan-out argument as the repair arm, and the walk is the +/// costlier half: `commit_journal` reaches `commit_messages`, which flushes a +/// segment and fsyncs under `enforce_fsync`. Run serially over every group on +/// the shard in one tick body, a node-wide rejoin exceeds the view-change +/// escalation window the superblock pre-pass already chunks to stay inside. +/// +/// Capping cannot wedge a partition: the walk carries no debounce counter and +/// clears its own predicate (a walk either advances `commit_min` or fences the +/// partition), so a skipped group is re-evaluated unchanged on the next tick +/// and the eligible set drains in `ceil(groups / cap)` ticks. +const PARTITION_WALKS_PER_TICK_MAX: usize = 16; + +/// Floor under the gap detector's debounce, in ticks. +/// +/// The debounce reads `repair_retry_interval`, whose primary meaning is how +/// long a STALLED repair stream waits before re-requesting. `duration_to_ticks` +/// floors that at one tick, and one tick of lag is ordinary pipelining, so +/// without a floor of its own an operator shortening the retry interval would +/// also arm repair against a single reordered prepare. +const PARTITION_GAP_DEBOUNCE_TICKS_MIN: u32 = 50; + /// What already owns a partition's recovery, if anything. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum RecoveryOwner { @@ -8930,6 +9041,11 @@ struct GapProbe { commit_min: u64, commit_max: u64, /// Whether `commit_min + 1` is resident in the local journal. + /// + /// Read only when the guards above already hold, and `false` otherwise: + /// both predicates test the lag first, so a probe that fails it is + /// answered without touching the journal at all. See + /// [`partition_gap_probe`]. next_op_resident: bool, } @@ -8948,10 +9064,12 @@ const fn partition_is_gap_stopped(probe: &GapProbe) -> bool { && !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. +/// The gap predicate's disjoint sibling, not its 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). The two split +/// on `next_op_resident`, so they cannot both hold, but both are false whenever +/// a shared guard fails. 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 @@ -8980,6 +9098,11 @@ const fn drive_partition_gap_debounce( *gap_ticks = 0; return false; } + let debounce_ticks = if debounce_ticks < PARTITION_GAP_DEBOUNCE_TICKS_MIN { + PARTITION_GAP_DEBOUNCE_TICKS_MIN + } else { + debounce_ticks + }; *gap_ticks = gap_ticks.saturating_add(1); *gap_ticks >= debounce_ticks && arms_this_tick < PARTITION_REPAIR_ARMS_PER_TICK_MAX } @@ -8992,6 +9115,7 @@ where { let consensus = partition.consensus(); let commit_min = consensus.commit_min(); + let commit_max = consensus.commit_max(); // 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() { @@ -9003,18 +9127,27 @@ where } else { RecoveryOwner::Nobody }; - GapProbe { - normal: consensus.is_normal(), - transferring: consensus.is_transferring(), - recovery, - commit_min, - commit_max: consensus.commit_max(), - next_op_resident: partition + let normal = consensus.is_normal(); + let transferring = consensus.is_transferring(); + // Residency last, and only once the guards both predicates share already + // hold. This runs for every group on the shard on every tick, and the + // caught-up steady state (`commit_min == commit_max`) would otherwise pay + // a journal lookup whose answer both predicates discard. + let next_op_resident = normal + && !transferring + && commit_min < commit_max + && partition .log .journal() .inner - .header_by_op(commit_min.saturating_add(1)) - .is_some(), + .holds_op(commit_min.saturating_add(1)); + GapProbe { + normal, + transferring, + recovery, + commit_min, + commit_max, + next_op_resident, } } @@ -10261,8 +10394,9 @@ mod gap_detector_tests { //! 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_GAP_DEBOUNCE_TICKS_MIN, PARTITION_REPAIR_ARMS_PER_TICK_MAX, + RecoveryOwner, drive_partition_gap_debounce, partition_is_gap_stopped, + partition_is_walk_stalled, }; const DEBOUNCE: u32 = 100; @@ -10511,4 +10645,22 @@ mod gap_detector_tests { "the same group arms on the next pass with a slot free" ); } + + #[test] + fn given_a_debounce_shorter_than_the_floor_when_driven_should_hold_until_the_floor() { + // `repair_retry_interval` is an operator knob whose primary meaning is + // the stalled-stream retry, and `duration_to_ticks` floors it at one + // tick. One tick of lag is ordinary pipelining, so without a floor of + // its own a shortened retry interval would arm repair against a single + // reordered prepare. + let probe = gap_stopped(); + let mut gap_ticks = 0; + for tick in 1..PARTITION_GAP_DEBOUNCE_TICKS_MIN { + assert!( + !drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, 0), + "a 1-tick debounce armed at tick {tick}, under the floor" + ); + } + assert!(drive_partition_gap_debounce(&probe, &mut gap_ticks, 1, 0)); + } } diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs index cefc3d4458..73434a1181 100644 --- a/core/shard/src/metrics.rs +++ b/core/shard/src/metrics.rs @@ -406,6 +406,12 @@ impl ShardMetrics { /// 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. + /// + /// Shard-scoped, with no namespace label: a server runs hundreds of groups + /// per shard, so labelling by namespace is unbounded cardinality, and every + /// other partition counter in this file is shard-scoped for the same + /// reason. The per-group detail is in the arm's log line. The metadata + /// plane's own gap drop is NOT counted here - it has its own recovery path. pub fn record_partition_prepare_gap_drops(&self, drops: u64) { self.partition_prepare_gap_drops_total.inc_by(drops); } diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index f2b28f8c6e..1158f43e42 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -301,6 +301,17 @@ where // simulator (see `MessageBus::sleep`). let rearm_tick = || self.bus.sleep(CONSENSUS_TICK_INTERVAL).fuse(); let mut consensus_tick = std::pin::pin!(rearm_tick()); + // Ready only while a materialisation has staged parked frames, so the + // arm it feeds is skipped entirely in the steady state. Rebuilt per + // iteration, like `inbox.recv()`, because readiness is a snapshot of + // the queue rather than a durable waker. + let staged_frames_ready = || { + if self.has_redispatched_frames() { + futures::future::ready(()).left_future() + } else { + futures::future::pending::<()>().right_future() + } + }; let mut fatal: Option = None; loop { // `select_biased!`, not `select!`: the unbiased macro draws its @@ -346,13 +357,18 @@ 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()); } + () = staged_frames_ready() => { + // Ranked below the tick and ABOVE the inbox, so nothing is + // read off the inbox while a materialisation's parked + // frames are still staged: 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. One frame + // per poll, so the tick still preempts between them. + self.serve_staged_partition_frame(&mut loopback_buf, &mut namespace_scratch) + .await; + } frame = self.inbox.recv().fuse() => { match frame { Ok(frame) => { @@ -360,8 +376,9 @@ where self.process_frame(frame).await; self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await; // Tail drain catches reconcile ops whose marker was dropped. + // Anything it stages is served by the arm above + // on the next pass, before this arm can run again. self.apply_reconcile_ops(); - self.drain_redispatched_frames().await; } // Guaranteed reply-lane service: `select_biased!` // polls the main lane first, so a saturated main @@ -403,43 +420,9 @@ where fatal = self.first_partition_commit_fault(); } - // Drain remaining frames so in-flight requests get a response, and - // the reply lane so already-forwarded replies still reach their - // clients before the bus tears down. Skipped on a commit fault: a - // queued Ack or Commit frame for the fenced partition would re-enter - // the commit path that just failed, and `advance_commit_min` asserts - // on the gap the fault left. Those requests go unanswered and their - // clients time out, which is what a node stopping on a durability - // fault owes them. - if fatal.is_none() { - while let Ok(frame) = self.inbox.try_recv() { - if self.accept_frame_for_self(&frame) { - self.process_frame(frame).await; - self.process_loopback(&mut loopback_buf, &mut namespace_scratch) - .await; - self.apply_reconcile_ops(); - if let Some(fault) = self.first_partition_commit_fault() { - fatal = Some(fault); - break; - } - } - } - // 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() { - if self.accept_frame_for_self(&frame) { - self.process_frame(frame).await; - if let Some(fault) = self.first_partition_commit_fault() { - fatal = Some(fault); - break; - } - } - } - } + fatal = self + .drain_pump_on_exit(fatal, &mut loopback_buf, &mut namespace_scratch) + .await; if fatal.is_some() { // Flipped BEFORE the final flush, not after the pump returns: the @@ -533,14 +516,102 @@ where } } - #[allow(clippy::future_not_send, clippy::too_many_lines)] - async fn process_lifecycle(&self, payload: LifecycleFrame) + /// Drain remaining frames so in-flight requests get a response, and the + /// reply lane so already-forwarded replies still reach their clients before + /// the bus tears down. Returns the fault to exit with. + /// + /// All of it is skipped on a commit fault: a queued Ack or Commit frame for + /// the fenced partition would re-enter the commit path that just failed, + /// and `advance_commit_min` asserts on the gap the fault left. Those + /// requests go unanswered and their clients time out, which is what a node + /// stopping on a durability fault owes them. + #[allow(clippy::future_not_send)] + async fn drain_pump_on_exit( + &self, + mut fatal: Option, + loopback_buf: &mut Vec>, + namespace_scratch: &mut Vec, + ) -> Option where B: MessageBus + 'static, MJ: JournalHandle, ::Target: Journal, Header = PrepareHeader>, M: RestorableMetadataStm, + { + if fatal.is_none() { + while let Ok(frame) = self.inbox.try_recv() { + if self.accept_frame_for_self(&frame) { + self.process_frame(frame).await; + self.apply_reconcile_ops(); + // Here, not after the loop: the select arm that normally + // serves staged frames is gone, and staging without + // delivering would leave a parked op behind a later op + // still on the inbox -- the ordering that arm exists to + // keep. Draining to empty is right once there is no tick + // left to preempt for. + self.drain_redispatched_frames().await; + self.process_loopback(loopback_buf, namespace_scratch).await; + if let Some(fault) = self.first_partition_commit_fault() { + fatal = Some(fault); + break; + } + } + } + // Whatever the loop above could not deliver: it breaks on a commit + // fault, and a frame can park again on a namespace this node no + // longer owns. Client requests get a transient deny; a prepare is + // counted, which is the only record it existed. 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() { + if self.accept_frame_for_self(&frame) { + self.process_frame(frame).await; + if let Some(fault) = self.first_partition_commit_fault() { + fatal = Some(fault); + break; + } + } + } + } + fatal + } + + /// One staged parked frame, plus the follow-up work delivering it owes. + /// + /// Split out of the pump's select arm only to keep that function readable; + /// the ordering guarantee lives in where the arm sits, not here. + #[allow(clippy::future_not_send)] + async fn serve_staged_partition_frame( + &self, + loopback_buf: &mut Vec>, + namespace_scratch: &mut Vec, + ) where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: RestorableMetadataStm, + { + self.dispatch_one_redispatched_frame().await; + // The delivery above can produce a self-addressed `PrepareOk`, which on + // a solo group is the whole quorum and reaches the plane only here. + self.process_loopback(loopback_buf, namespace_scratch).await; + // Same guaranteed reply-lane service the inbox arm owes: without it a + // long staged run starves forwarded replies for its whole duration. + if let Ok(reply) = self.reply_inbox.try_recv() + && self.accept_frame_for_self(&reply) + { + self.process_frame(reply).await; + } + } + + #[allow(clippy::future_not_send, clippy::too_many_lines)] + async fn process_lifecycle(&self, payload: LifecycleFrame) + where + B: MessageBus + 'static, { match payload { LifecycleFrame::ReplicaInboundSetup { fd, slot } => { @@ -690,7 +761,6 @@ where } LifecycleFrame::ReconcileApply => { self.apply_reconcile_ops(); - self.drain_redispatched_frames().await; } LifecycleFrame::CleanPartition { namespace, diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 44421de461..020f73a33c 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -4723,6 +4723,36 @@ mod partition_repair_driver_tests { /// commits just collects transient rejections and the group goes quiet. const TICKS_PER_SEND: usize = 4; + /// What the healthy run saw of the backup's commit lag, sampled per tick. + /// + /// Lag alone is what a naive detector reads as a gap, so the run has to + /// record how much of it there was AND whether the op past the frontier was + /// resident each time, rather than assert the lag away. + #[derive(Default)] + struct LagObservations { + samples: u32, + resident_samples: u32, + longest_run: u32, + current_run: u32, + } + + impl LagObservations { + /// Fold one tick's view of `replica`'s commit frontier. + fn observe(&mut self, sim: &Simulator, replica: u8, namespace: IggyNamespace) { + let (_, _, commit_min, commit_max) = group_state(sim, replica, namespace); + if commit_min >= commit_max { + self.current_run = 0; + return; + } + self.current_run += 1; + self.longest_run = self.longest_run.max(self.current_run); + self.samples += 1; + if journal_holds(sim, replica, namespace, commit_min + 1) { + self.resident_samples += 1; + } + } + } + /// 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 @@ -5122,17 +5152,7 @@ mod partition_repair_driver_tests { // 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; - } - }; + let mut observed = LagObservations::default(); for tick in 0..LOAD_TICKS { if tick % TICKS_PER_SEND == 0 { let msg = @@ -5140,11 +5160,11 @@ mod partition_repair_driver_tests { sim.submit_request(client.client_id(), 0, msg.into_generic()); } sim.step(); - sample(&sim, &mut lag_run, &mut longest_lag_run); + observed.observe(&sim, 1, namespace); } for _ in 0..QUIET_STEPS { sim.step(); - sample(&sim, &mut lag_run, &mut longest_lag_run); + observed.observe(&sim, 1, namespace); } let committed = group_state(&sim, 1, namespace).2; @@ -5154,18 +5174,27 @@ mod partition_repair_driver_tests { "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. + // Lag is RECORDED, not forbidden. Asserting it away would make the + // repair-request assertion below unfalsifiable, since + // `partition_is_gap_stopped` needs `commit_min < commit_max` to fire at + // all. What must hold is that every lag this run saw was backed by a + // RESIDENT next op -- a walk waiting to run, not a hole -- which is the + // distinction a naive detector would miss. + // + // On a two-replica group the backup keeps pace and the tick's own + // walk-stalled backstop drains what little lag a step leaves, so this + // usually samples none and the check costs nothing. It is the guard for + // when that stops being true, not the proof: the predicate itself is + // pinned exhaustively by `gap_detector_tests`, and what this run adds is + // that the DRIVER sends no repair over a live, loaded group. 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" + observed.samples, + observed.resident_samples, + "the backup lagged with a MISSING next op on {} of {} sampled ticks, so a \ + no-loss two-replica run produced a real hole and the repair assertion \ + below would be testing loss recovery instead of false positives", + observed.samples - observed.resident_samples, + observed.samples ); for replica in 0..replica_count { let (status, view, commit_min, commit_max) = group_state(&sim, replica, namespace); @@ -5193,7 +5222,10 @@ mod partition_repair_driver_tests { 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" + is reading ordinary commit lag as a journal hole (backup lagged on {} of \ + the sampled ticks, longest run {})", + observed.samples, + observed.longest_run ); }