You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/09/01 14:14:06 UTC
(iggy) branch master updated: fix(shard): keep a parked prepare's stamp and op order on re-dispatch (#4005)
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3aed9f458 fix(shard): keep a parked prepare's stamp and op order on re-dispatch (#4005)
3aed9f458 is described below
commit 3aed9f458e545cb0de77bbe040e68ab5d5579a5a
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Tue Sep 1 16:14:01 2026 +0200
fix(shard): keep a parked prepare's stamp and op order on re-dispatch (#4005)
Partition prepares can arrive before their namespace is
materialised. Re-dispatch through the shard inbox lost each
frame's original epoch stamp and placed it behind newer frames.
A delete and recreate could therefore serve an old prepare
against the replacement, while the backup gap check could drop
a later operation before the parked prefix reached the plane.
Stage parked frames in a shard-local FIFO and carry their epoch
and park-pass provenance. A biased pump arm processes one
staged frame per iteration after ticks and before inbox work.
This preserves order without delaying heartbeats behind a
bounded queue, and loopback runs after every delivery so a solo
primary commits re-dispatched requests immediately. Shutdown
uses the same ordering.
Simulator materialisation now passes its resolved epoch and
wakes the pump after off-pump staging. Seeded three-replica and
solo-primary tests pin queue ordering, wakeup, loopback, and
deterministic replay. Quiescence rejects stranded redispatch
work.
The integration oracle now filters gap markers to the partition
plane, removes ambient RUST_LOG when an explicit logging level
is configured, and falls back to the server log when verbose
mode inherits stdout. Dropped or rejected staged frames are
counted, and recovery documentation reflects same-view repair.
The shared leader-redirection BDD fixture could elect replica 1
when a loaded runner delayed replica 0 during concurrent
startup. Start replica 1 only after replica 0 is healthy and use
a wider fixture heartbeat window so the scenarios test client
redirection instead of container scheduling.
---------
Co-authored-by: Piotr Gankiewicz <pi...@gmail.com>
---
bdd/docker-compose.cluster.yml | 6 +
core/integration/src/harness/handle/server.rs | 49 +-
core/integration/tests/cluster/mod.rs | 1 +
.../tests/cluster/parked_frame_redispatch.rs | 363 +++++++++++++++
core/server/src/partition_reconciler.rs | 396 +++++++++-------
core/shard/src/lib.rs | 508 +++++++++++----------
core/shard/src/metrics.rs | 35 +-
core/shard/src/router.rs | 109 ++++-
core/simulator/src/lib.rs | 271 ++++++++++-
9 files changed, 1266 insertions(+), 472 deletions(-)
diff --git a/bdd/docker-compose.cluster.yml b/bdd/docker-compose.cluster.yml
index 9976d342b..df773b8a5 100644
--- a/bdd/docker-compose.cluster.yml
+++ b/bdd/docker-compose.cluster.yml
@@ -56,6 +56,9 @@ x-cluster-topology: &cluster-topology
IGGY_ROOT_PASSWORD: iggy
IGGY_CLUSTER_ENABLED: "true"
IGGY_CLUSTER_NAME: test-cluster
+ # Keep role-specific scenarios on view 0 when a loaded CI runner delays
+ # one replica during startup.
+ IGGY_CLUSTER_HEARTBEAT_TIMEOUT: "30s"
IGGY_CLUSTER_NODES_0_NAME: leader-node
IGGY_CLUSTER_NODES_0_IP: 172.28.0.101
IGGY_CLUSTER_NODES_0_REPLICA_ID: "0"
@@ -115,6 +118,9 @@ services:
iggy-follower:
<<: *cluster-node
+ depends_on:
+ iggy-leader:
+ condition: service_healthy
command: [ "--replica-id", "1" ]
healthcheck:
test: [ "CMD", "/usr/local/bin/iggy", "--tcp-server-address", "172.28.0.102:8092", "ping" ]
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index ed015a1ec..088b3b261 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -187,35 +187,46 @@ impl ServerHandle {
self.stdout_occurrences(marker) > 0
}
- /// Number of times `marker` appears in this node's stdout log. The log
- /// file is TRUNCATED on every (re)start (it captures one process run),
- /// so counts never carry across a restart; a marker seen after
- /// `restart_server` was logged by the new process.
+ /// Number of times `marker` appears in this node's captured stdout, or in
+ /// its own logs when verbose mode makes the child inherit stdout. Captured
+ /// stdout is truncated on restart. The server log may span restarts, so
+ /// restart-sensitive callers must compare against a pre-restart baseline.
///
/// ANSI escape sequences are stripped before matching: the server colors
/// its tracing fields, so a `key=value` marker never matches the raw
/// bytes (`key\x1b[0m\x1b[2m=\x1b[0m value`).
#[must_use]
pub fn stdout_occurrences(&self, marker: &str) -> usize {
- self.stdout_path
- .as_ref()
- .and_then(|path| fs::read_to_string(path).ok())
- .map_or(0, |log| strip_ansi(&log).matches(marker).count())
+ self.stdout_plain().matches(marker).count()
}
- /// This node's stdout log with ANSI escapes stripped, the same text
- /// [`Self::stdout_occurrences`] matches against. Empty when the log is
- /// missing or unreadable.
+ /// This node's captured stdout with ANSI escapes stripped, or its own logs
+ /// when verbose mode disables the capture. Empty when neither source is
+ /// readable.
///
/// For callers that need to PARSE a marker's fields (`checkpoint_op=193`
/// reaches the file as `checkpoint_op\x1b[0m\x1b[2m=\x1b[0m193`) rather
/// than just count occurrences of it.
#[must_use]
pub fn stdout_plain(&self) -> String {
- self.stdout_path
+ let captured = self
+ .stdout_path
.as_ref()
.and_then(|path| fs::read_to_string(path).ok())
- .map_or_else(String::new, |log| strip_ansi(&log))
+ .map(|log| strip_ansi(&log));
+ captured.unwrap_or_else(|| {
+ let own_logs = self.data_path().join("logs");
+ let Ok(entries) = fs::read_dir(own_logs) else {
+ return String::new();
+ };
+ let mut log = String::new();
+ for entry in entries.flatten() {
+ if let Ok(contents) = fs::read_to_string(entry.path()) {
+ log.push_str(&contents);
+ }
+ }
+ log
+ })
}
/// Returns a `ClientBuilder` using the test transport.
@@ -882,6 +893,18 @@ impl TestBinary for ServerHandle {
{
command.env("IGGY_SHARD_RUNTIME_CAPACITY", "256");
}
+ // An explicit config-level override must be the test's logging
+ // contract. The server gives `RUST_LOG` precedence, so inheriting an
+ // ambient value could silently filter out markers the test asserts.
+ // A caller that explicitly puts `RUST_LOG` in `extra_envs` adds it back
+ // through `command.envs` below.
+ if self
+ .config
+ .extra_envs
+ .contains_key("IGGY_SYSTEM_LOGGING_LEVEL")
+ {
+ command.env_remove("RUST_LOG");
+ }
command.envs(&self.envs);
// `--replica-id` is the single identity input expected by the
diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index 940e1fcee..b794264a0 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 000000000..3b1a525ed
--- /dev/null
+++ b/core/integration/tests/cluster/parked_frame_redispatch.rs
@@ -0,0 +1,363 @@
+// 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 creates a replica gap: a client request that never
+//! reaches the plane is answered with a retriable status and the SDK replays it,
+//! while a replicated PREPARE has no client behind it and must wait for a later
+//! commit heartbeat to arm repair if this path drops it.
+//!
+//! 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.
+//!
+//! The harness removes an ambient `RUST_LOG` when this test supplies its explicit
+//! logging level, and the log oracle falls back from captured stdout to the
+//! server's own log file so `IGGY_TEST_VERBOSE` cannot disable it.
+
+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::{ServerHandle, 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
+/// forces repair that correct redispatch ordering should never need.
+const GAP_MARKER: &str = "dropping out-of-order prepare (gap)";
+const PARTITION_PLANE_FIELD: &str = "plane=\"partitions\"";
+
+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<PathBuf> = 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<Vec<String>> {
+ 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<String> {
+ 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<String>], 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<String> = (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,
+ server_log_occurrences(harness.node(node), 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 the configured debug marker was not written"
+ );
+}
+
+fn assert_no_degraded_park_paths(harness: &TestHarness) {
+ for node in 0..harness.cluster_size() {
+ let server = harness.node(node);
+ let log = server_log_plain(server);
+ for marker in DEGRADED_MARKERS {
+ assert_eq!(
+ log.matches(marker).count(),
+ 0,
+ "node {node} logged {marker:?}: the park buffer degraded instead of converging"
+ );
+ }
+ let partition_gaps = log
+ .lines()
+ .filter(|line| line.contains(GAP_MARKER) && line.contains(PARTITION_PLANE_FIELD))
+ .count();
+ assert_eq!(
+ partition_gaps, 0,
+ "node {node} logged {GAP_MARKER:?}: a re-dispatched prepare lost its arrival \
+ position and forced avoidable partition repair"
+ );
+ }
+}
+
+fn server_log_occurrences(server: &ServerHandle, marker: &str) -> usize {
+ server_log_plain(server).matches(marker).count()
+}
+
+/// The shared handle falls back to the server's own appender when
+/// `IGGY_TEST_VERBOSE` makes the child inherit stdout.
+fn server_log_plain(server: &ServerHandle) -> String {
+ server.stdout_plain()
+}
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 5cbd5f19a..5de84bcc1 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -40,20 +40,20 @@
//! "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. Production prevents a second park by ranking redispatch above
+//! inbox work and applying incarnation changes only on the pump. Carrying the
+//! stamp through re-delivery is defence in depth for off-pump staging such as
+//! simulator materialisation. 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. The redispatch select arm takes one frame
+//! per iteration before the inbox arm, so a parked op is not ordered behind a
+//! later op of the same partition already queued there.
//! - `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
@@ -89,9 +89,9 @@
//!
//! They apply asymmetrically, because the two frame classes fail differently. A
//! shed request costs a retry: answered with a retriable status, re-issued by
-//! the SDK. A shed prepare is permanent loss on this replica, with no client to
-//! answer and `consensus::retransmit_targets` skipping any op that already
-//! reached quorum.
+//! the SDK. A shed prepare has no client to retry it and leaves the replica
+//! behind until a later commit heartbeat exposes the gap and arms same-view
+//! journal repair.
//!
//! All three bind a request: refused when admitting it would cross a byte budget
//! or the frame cap, answered past `MAX_PARKED_PASSES`. Only the byte budgets
@@ -115,34 +115,11 @@
//! 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): 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.
+//! A shed prepare is not retransmitted once its op reached quorum, but it is not
+//! stranded until a view change. A later `CommitMessage` that advances the
+//! backup's frontier runs `maybe_request_partition_repair`; an evicted repair
+//! range escalates to partition state transfer. The park policy still avoids
+//! manufacturing that recovery work unless a byte budget is already spent.
//!
//! TODO(krishna): `serves_committed_incarnation` and the park stamp both call
//! `Streams::created_revision_for_namespace`, now on the per-request fence path.
@@ -741,12 +718,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
@@ -783,11 +759,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;
}
@@ -1603,19 +1578,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<shard::ShardFrame>,
reply: shard::Receiver<shard::ShardFrame>,
+ main_tx: shard::TaggedSender,
}
fn build_test_shard_with_inbox(
@@ -1628,13 +1604,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");
@@ -1645,8 +1622,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() {
@@ -1673,6 +1650,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<u64> {
+ 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<TestShard>,
total_shards: u16,
@@ -3408,10 +3401,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,
@@ -3483,9 +3481,8 @@ mod tests {
}
/// 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.
+ /// park critical are worst: a prepare has no client, so discarding it forces
+ /// the backup to wait for a later commit heartbeat and journal repair.
///
/// A backup receives the prepare before its own metadata commits (so the frame
/// parks unstamped), then applies the commit and materialises. The prepare must
@@ -3521,11 +3518,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 \
+ an avoidable gap, since a prepare has no client to retry it"
+ );
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!(
@@ -3576,6 +3578,158 @@ mod tests {
1,
"and the reject must be counted"
);
+ assert_eq!(
+ park_dropped_count(&shard),
+ 1,
+ "a rejected prepare has no client deny, so its destruction must be recorded"
+ );
+ }
+
+ /// Defence-in-depth for off-pump staging: materialisation stages the frame,
+ /// the delete half of a recreate leaves the namespace unmaterialised, and
+ /// explicit test delivery parks it a second time. Production cannot take
+ /// this interleaving because every pump-side reconcile apply returns to the
+ /// higher-ranked redispatch arm before another incarnation change can run.
+ /// The carried stamp still prevents simulator or test staging from deriving
+ /// the replacement's revision on that second park.
+ #[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"
+ );
+
+ assert!(
+ shard.dispatch_one_redispatched_frame_for_test().await,
+ "the defence-in-depth frame must be available for explicit delivery"
+ );
+ 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`. Same-view repair
+ /// can heal that gap later, but redispatch must not manufacture 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,
@@ -3673,7 +3827,7 @@ mod tests {
assert_eq!(
park_dropped_count(&shard),
0,
- "the prepare must be retained: destroying it is unrecoverable"
+ "the prepare must be retained instead of forcing journal repair"
);
// Retained across an unbounded number of further passes.
@@ -3850,9 +4004,9 @@ mod tests {
}
/// The age bound answers requests and steps over prepares. Expiring a
- /// prepare is permanent loss (no client, and `retransmit_targets` skips an op
- /// already at quorum), and passes are commit-driven, so a create burst
- /// elapses four in milliseconds across every parked namespace at once.
+ /// prepare would force same-view repair despite the bytes still being held
+ /// locally, and passes are commit-driven, so a create burst elapses four in
+ /// milliseconds across every parked namespace at once.
#[compio::test]
async fn aging_answers_requests_and_never_expires_a_prepare() {
let tmp = TempDir::new().expect("tempdir for system path");
@@ -3927,8 +4081,8 @@ mod tests {
}
/// A frame larger than the per-namespace cap failed the check even against
- /// an empty entry, so it could never park. Unrecoverable for a prepare:
- /// `retransmit_targets` skips an op already at quorum.
+ /// an empty entry, so it could never park. A prepare already at quorum may
+ /// no longer retransmit, so shedding it forces later same-view repair.
#[compio::test]
async fn a_frame_over_the_namespace_byte_cap_still_parks_into_an_empty_entry() {
const OVER_NAMESPACE_CAP: usize = 5 * 1024 * 1024;
@@ -4194,62 +4348,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");
@@ -4258,20 +4359,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);
@@ -4287,35 +4388,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 98674232c..a2ffd20a3 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")]
@@ -107,6 +106,27 @@ where
pub clock: ConsensusClock,
}
+/// Committed metadata the simulator carries into one partition
+/// materialisation. Named because both values are `u64`-compatible revision or
+/// view stamps and swapping positional arguments would compile.
+#[cfg(feature = "simulator")]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PartitionMaterialisation {
+ epoch: u64,
+ created_view: u32,
+}
+
+#[cfg(feature = "simulator")]
+impl PartitionMaterialisation {
+ #[must_use]
+ pub const fn new(epoch: u64, created_view: u32) -> Self {
+ Self {
+ epoch,
+ created_view,
+ }
+ }
+}
+
/// Replica id + count bundle.
///
/// Adjacent `u8` params (`self_replica_id`, `replica_count`) were a
@@ -1395,18 +1415,17 @@ where
/// admission, on the reactor thread inside the map's `borrow_mut`.
parked_partition_bytes: Cell<usize>,
- /// Namespaces holding frames [`Self::redispatch_parked_frames`] could not
- /// re-queue, for the pump to retry.
- ///
- /// 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.
+ /// Frames [`Self::redispatch_parked_frames`] handed back for the pump to
+ /// deliver, in park order.
///
- /// [`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<BTreeSet<IggyNamespace>>,
+ /// Staging exists because re-dispatch runs inside the synchronous
+ /// [`Self::apply_reconcile_ops`] while the plane is reachable only through
+ /// an async path. A biased select arm takes one staged frame per pump
+ /// iteration and ranks above the inbox, so a parked op reaches the plane
+ /// ahead of a later op already sitting there. One-at-a-time delivery lets
+ /// consensus ticks and commit broadcasts run between frames instead of
+ /// stalling behind the whole bounded queue.
+ redispatch_queue: RefCell<VecDeque<ParkedFrame>>,
/// 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 +1608,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 +1938,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),
@@ -2010,6 +2029,13 @@ where
/// queue never strands ops for longer than one tick.
pub fn enqueue_reconcile_op(&self, op: ReconcileOp<B, SB>) {
self.reconcile_queue.borrow_mut().push_back(op);
+ self.wake_reconcile_apply();
+ }
+
+ /// Wake the pump after off-pump work becomes visible. A refused marker is
+ /// safe because a full inbox has already woken the pump, whose frame and
+ /// tick arms both apply staged reconciliation work.
+ fn wake_reconcile_apply(&self) {
let Some(sender) = self.senders.get(self.id as usize) else {
return;
};
@@ -2109,46 +2135,18 @@ 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<IggyNamespace> = {
- 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
- /// stall on bulk teardown.
+ ///
+ /// Synchronous: every arm is in-memory only. `ConfirmRemove`'s fsync and
+ /// blocking close are offloaded to a detached task so the pump does not
+ /// stall on bulk teardown. An `InsertOwned` can stage parked frames, so a
+ /// live-pump caller must return to the ranked redispatch select arm before
+ /// reading the inbox again. The graceful-shutdown caller drains the queue
+ /// explicitly because it has already left the select loop.
pub fn apply_reconcile_ops(&self)
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<ReconcileOp<B, SB>> = {
let mut q = self.reconcile_queue.borrow_mut();
if q.is_empty() {
@@ -2386,7 +2384,8 @@ enum ParkOutcome<H> {
/// Namespace is unmaterialised and its park buffer is at capacity. Client
/// requests must be denied with a transient status: the frame is gone, and
/// silence would leave a lockstep transport waiting out its response
- /// read-timeout. Replicated traffic is dropped, recovered by retransmit.
+ /// read-timeout. Replicated traffic is dropped and recovers through either
+ /// retransmit or the same-view repair armed by a later commit heartbeat.
Overflow(Message<H>),
}
@@ -2406,9 +2405,9 @@ struct ParkedFrame {
/// answers CLIENT REQUESTS past [`MAX_PARKED_PASSES`], in units the
/// simulator's virtual clock controls.
///
- /// Never expires a replicated prepare: no client to answer, and
- /// `consensus::retransmit_targets` skips an op that already reached quorum,
- /// so expiry is silent permanent loss. Byte budgets bound those instead.
+ /// Never expires a replicated prepare: no client can retry it, and forcing
+ /// the same-view repair backstop to recover a gap is worse than retaining
+ /// the frame. Byte budgets bound those instead.
///
/// Bounds RESIDENCY, not staleness. The SDK replays the identical request
/// for the rest of its response timeout, so an absolute-offset
@@ -2430,6 +2429,20 @@ impl ParkedFrame {
}
}
+/// What a frame keeps if it parks again after the pump re-delivers it.
+///
+/// Production prevents that race by ranking redispatch above inbox work and by
+/// applying reconcile operations only on the pump. Carrying the original stamp
+/// is defence in depth for off-pump staging such as simulator materialisation.
+/// There, re-deriving on a second park could stamp the frame with a replacement
+/// incarnation. `None` also stays `None`, since absence of a committed revision
+/// is not evidence that the frame belongs to a prior incarnation.
+#[derive(Clone, Copy)]
+struct ParkProvenance {
+ epoch: Option<u64>,
+ passes: u32,
+}
+
/// One namespace's parked frames plus their running footprint.
///
/// Carried, not re-summed: `park_if_unmaterialised` reads it per arriving frame
@@ -2492,11 +2505,11 @@ const MAX_PARKED_BYTES: usize = 16 * 1024 * 1024;
/// cannot spend the whole shard's budget and shed everyone else's frames.
///
/// Applied only to an entry that already holds something. Sized against an
-/// empty entry a larger frame could never park at all, and for a prepare that is
-/// unrecoverable loss: `consensus::retransmit_targets` skips an op that already
-/// reached quorum. Shipped `message_bus.max_message_size` is 64 MiB, so an
-/// ordinary batched append exceeds this. Cost of the waiver is one convergence
-/// window of shard budget; cost of the loss is the replica.
+/// empty entry a larger frame could never park at all. For a prepare, shedding
+/// also forces a later commit heartbeat to discover the gap and run same-view
+/// repair. Shipped `message_bus.max_message_size` is 64 MiB, so an ordinary
+/// batched append exceeds this. The waiver costs one convergence window of
+/// shard budget and avoids unnecessary recovery work.
const MAX_PARKED_BYTES_PER_NAMESPACE: usize = MAX_PARKED_BYTES / 4;
/// Resident cost of parking a frame of `len` bytes.
@@ -2553,14 +2566,99 @@ where
MJ: JournalHandle,
<MJ as JournalHandle>::Target:
Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- M: StateMachine<
- Input = Message<PrepareHeader>,
- Output = metadata::stm::result::ApplyReply,
- Error = iggy_common::IggyError,
- > + StreamsFrontend
- + metadata::stm::snapshot::RestoreSnapshotInPlace<
- metadata::stm::snapshot::MetadataSnapshot,
- >,
+ M: RestorableMetadataStm,
+ T: ShardsTable,
+ {
+ self.dispatch_message(message, None).await;
+ }
+
+ /// Remove and classify one staged frame for the pump's ranked redispatch
+ /// arm. The queue borrow ends before dispatch awaits, so simulator
+ /// materialisation can append off-pump without colliding with a suspended
+ /// `RefCell` guard.
+ fn pop_redispatched_frame(&self) -> Option<(MessageBag, ParkProvenance)> {
+ loop {
+ let ParkedFrame {
+ epoch,
+ passes,
+ message,
+ } = self.redispatch_queue.borrow_mut().pop_front()?;
+ let provenance = ParkProvenance { epoch, passes };
+ // 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.
+ match MessageBag::try_from(message) {
+ Ok(bag) => return Some((bag, provenance)),
+ Err(error) => {
+ // The frame classified once already, on the way in, so this
+ // is unreachable short of memory corruption. The consumed
+ // bytes can no longer produce a client deny, but the drop
+ // still needs the same operator-visible record as any other
+ // parked frame retired unserved.
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ tracing::error!(
+ shard = self.id,
+ %error,
+ "re-dispatched partition frame no longer classifies; dropping it"
+ );
+ }
+ }
+ }
+ }
+
+ /// Test-only delivery of one staged frame. Production obtains frames through
+ /// the router's ranked select arm, which also processes loopback after each
+ /// one. This hook exists for the reconciler's defence-in-depth interleaving.
+ #[cfg(feature = "simulator")]
+ #[allow(clippy::future_not_send)]
+ pub async fn dispatch_one_redispatched_frame_for_test(&self) -> bool
+ where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
+ T: ShardsTable,
+ {
+ let Some((message, provenance)) = self.pop_redispatched_frame() else {
+ return false;
+ };
+ self.dispatch_message(message, Some(provenance)).await;
+ true
+ }
+
+ /// 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.
+ fn retire_redispatched_frames(&self) {
+ let staged: Vec<ParkedFrame> = 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<ParkProvenance>)
+ where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
T: ShardsTable,
{
match message {
@@ -2570,7 +2668,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 +2699,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 +2904,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<Vec<ParkedFrame>> {
- 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)?;
@@ -2839,19 +2934,28 @@ where
let mut answered = 0;
let mut dropped = 0;
for frame in frames {
- if self.deny_parked_client_request(frame) {
+ if self.retire_parked_frame(frame) {
answered += 1;
} else {
dropped += 1;
- self.metrics.record_frame_drop(
- crate::metrics::frame_drop_variant::PARTITION,
- crate::metrics::frame_drop_reason::PARK_DROPPED,
- );
}
}
(answered, dropped)
}
+ /// Answer one parked request or count one replicated frame as destroyed.
+ /// Returns `true` only when a client deny reached the pump.
+ fn retire_parked_frame(&self, frame: ParkedFrame) -> bool {
+ if self.deny_parked_client_request(frame) {
+ return true;
+ }
+ self.metrics.record_frame_drop(
+ crate::metrics::frame_drop_variant::PARTITION,
+ crate::metrics::frame_drop_reason::PARK_DROPPED,
+ );
+ false
+ }
+
/// Whether any frame is parked. Cheap enough for the reconciler's per-tick
/// fast-skip guard: a non-empty buffer means the shard is by definition not
/// converged, so the skip must not fire.
@@ -2877,9 +2981,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
@@ -2893,35 +2996,27 @@ where
/// buffer exists to absorb -- the partition primary materialises and
/// 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.
+ /// destroys live traffic and forces the backup to recover a gap that the
+ /// park path could have delivered directly.
/// 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
/// 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. The pump's biased redispatch arm ranks above its
+ /// inbox arm and delivers one staged frame per iteration.
///
- /// [`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 selects the queue on the iteration after it was
+ /// filled. Returns whether at least one frame was staged.
+ fn redispatch_parked_frames(&self, namespace: IggyNamespace, epoch: u64) -> bool {
let Some(frames) = self.take_parked_partition_frames(namespace) else {
- return;
+ return false;
};
tracing::debug!(
shard = self.id,
@@ -2930,8 +3025,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<ParkedFrame> = Vec::with_capacity(frames.len());
for frame in frames {
// Only a stamp that exists and disagrees is evidence of a prior
@@ -2944,119 +3037,21 @@ 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<ParkedFrame> = 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<ParkedFrame>) {
- 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);
+ let staged = !servable.is_empty();
+ self.redispatch_queue.borrow_mut().extend(servable);
+ staged
}
/// Age every frame under `namespace` by one pass, answering CLIENT REQUESTS
/// past `MAX_PARKED_PASSES`. Returns the number answered.
///
- /// Prepares age but never expire. Expiry destroys a committed op with
- /// nothing to recover it (see `ParkedFrame::passes`), and passes are
- /// commit-driven: a non-empty buffer defeats the reconciler fast-skip, so a
- /// create burst elapses four in milliseconds, across every parked namespace
- /// rather than the one it concerns. Byte budgets bound them instead. Only
- /// [`Self::discard_parked_partition_frames`] still destroys a prepare.
+ /// Prepares age but never expire. Expiry would manufacture a gap that a
+ /// later commit heartbeat must repair (see `ParkedFrame::passes`), and
+ /// passes are commit-driven: a non-empty buffer defeats the reconciler
+ /// fast-skip, so a create burst elapses four in milliseconds across every
+ /// parked namespace rather than the one it concerns. Byte budgets bound
+ /// them instead. Only [`Self::discard_parked_partition_frames`] still
+ /// destroys a prepare.
///
/// Passes, not wall-clock, so the simulator's virtual clock governs it.
pub fn age_parked_partition_frames(&self, namespace: IggyNamespace) -> usize {
@@ -3073,7 +3068,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 +3107,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 consumes this queue through a
+ /// dedicated select arm, 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.
@@ -3121,9 +3127,9 @@ where
/// with no sender stages nothing, hence forwarding
/// [`Self::stage_transient_deny`]'s verdict rather than assuming success.
///
- /// No reply must not mean no record: the primary retransmits only what has
- /// not reached quorum, so a destroyed prepare is invisible loss. The `false`
- /// return is what makes callers bump
+ /// No reply must not mean no record: the primary may no longer retransmit an
+ /// op that reached quorum, so a destroyed prepare creates a gap that later
+ /// repair must fill. The `false` return is what makes callers bump
/// `frame_drops_total{variant=partition,reason=park_dropped}`.
fn deny_parked_client_request(&self, frame: ParkedFrame) -> bool {
if frame.message.header().command == Command::Request
@@ -3175,7 +3181,7 @@ where
"rejecting parked partition frame from a prior incarnation"
);
}
- self.deny_parked_client_request(frame);
+ self.retire_parked_frame(frame);
}
/// Park a partition-plane frame whose namespace this shard has not yet
@@ -3186,16 +3192,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<H>(
&self,
message: Message<H>,
operation: Operation,
namespace_raw: u64,
+ provenance: Option<ParkProvenance>,
) -> ParkOutcome<H>
where
H: iggy_binary_protocol::ConsensusHeader,
@@ -3217,13 +3228,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();
@@ -3234,13 +3250,12 @@ 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
- // would cross a budget; a prepare only once one is already spent. Caps
- // prepare residency at one frame of overshoot per budget (worst case
+ // A prepare is never shed on a byte budget before the budget is spent.
+ // It has no client to retry it, and recovery requires a later commit
+ // heartbeat to expose the gap and arm same-view repair. A request costs
+ // only a retry, so it is refused the moment admitting it would cross a
+ // budget. This caps prepare residency at one frame of overshoot per
+ // budget (worst case
// `MAX_PARKED_BYTES` + `max_message_size`, 80 MiB per shard) instead of
// at the budget, and is what makes an oversize frame parkable at all.
let namespace_budget_spent = parked_len > 0
@@ -3317,7 +3332,7 @@ where
);
pending.entry(namespace).or_default().push(ParkedFrame {
epoch,
- passes: 0,
+ passes,
message: message.into_generic(),
});
drop(pending);
@@ -3556,8 +3571,12 @@ where
/// reports `commit_offset` 0, which reads as a regression rather than a
/// harness that discarded the log.
///
- /// `created_view` is the view the metadata plane created the namespace in;
- /// see `fresh_group_start`.
+ /// `materialisation` carries the committed `created_revision` and the view
+ /// the metadata plane created the namespace in; see `fresh_group_start`.
+ ///
+ /// Once inserted, this also runs the same parked-frame redispatch as
+ /// `ReconcileOp::InsertOwned`. The simulator bypasses the reconciler build,
+ /// but it must not bypass the pump handoff that follows materialisation.
// `feature = "simulator"` alone, unlike its neighbours: the body names items
// `partitions` gates the same way, and a `test` arm cannot turn those on.
// Under `cargo test -p shard` that arm fires from shard's own `cfg(test)`
@@ -3572,10 +3591,15 @@ where
recovered_state: Option<consensus::VsrState>,
retained: Option<partitions::RetainedPartitionState>,
restore_frontier: bool,
- created_view: u32,
+ materialisation: PartitionMaterialisation,
) where
- B: MessageBus + Clone,
+ B: MessageBus + Clone + 'static,
+ T: ShardsTable,
{
+ let PartitionMaterialisation {
+ epoch,
+ created_view,
+ } = materialisation;
let partitions = self.plane.partitions();
if partitions.contains(&namespace) {
return;
@@ -3678,6 +3702,12 @@ where
// store resumes minting at 0 while its group is at N.
partition.restore_offset_frontier(recovered_state.as_ref());
partitions.insert(namespace, partition);
+ if self.redispatch_parked_frames(namespace, epoch) {
+ // This mutation occurs outside the pump, unlike production's
+ // `InsertOwned`. Wake the ranked redispatch arm so quiescence does
+ // not leave real work staged without a poll source.
+ self.wake_reconcile_apply();
+ }
}
/// Resolve the single partition a VSR control frame addresses, keyed by
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
index cd05dba02..16007ffec 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,18 +77,14 @@ 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
-/// 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`.
+/// (`reason=park_overflow`), a parked frame retired with no client to answer
+/// (`reason=park_dropped`), an incarnation rejection, or a routing send the
+/// target inbox refused. A shed client request is answered with a retriable
+/// status. A shed prepare may no longer be covered by retransmit once its op
+/// reached quorum, but a later `CommitMessage` that advances the backup's
+/// frontier arms same-view journal repair. If the primary evicted the range,
+/// repair escalates to partition state transfer. The counter therefore signals
+/// a data-plane gap or recovery burden, not a requirement for a view change.
pub mod frame_drop_variant {
pub const CONSENSUS: &str = "consensus";
pub const FD_TRANSFER: &str = "fd_transfer";
@@ -114,11 +110,12 @@ pub mod frame_drop_variant {
/// `PARK_OVERFLOW` ticks when a partition frame arrives for a namespace this
/// shard has not materialised and the per-namespace park buffer is already at
/// its cap, so the frame is shed with no reply. `PARK_DROPPED` ticks when a
-/// frame that did park leaves unserved: its namespace became unreachable, or a
-/// request outlived `MAX_PARKED_PASSES` with no pump to take the deny. A request
-/// also bumps `partition_requests_denied_transient_total` when answered;
-/// replicated traffic has nobody to answer, so this is the only record the op
-/// was destroyed.
+/// frame that did park leaves unserved: its namespace became unreachable, its
+/// incarnation stamp was rejected, reclassification failed, or a request
+/// outlived `MAX_PARKED_PASSES` with no pump to take the deny. A request also
+/// bumps `partition_requests_denied_transient_total` when answered; replicated
+/// traffic has nobody to answer, so this is the direct record that local bytes
+/// were destroyed and repair may be required.
pub mod frame_drop_reason {
/// Operation discriminant unknown to this build: the sender is newer.
///
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index c98e58530..a866a9027 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -30,8 +30,10 @@ use message_bus::{ConnectionInstaller, MessageBus, ReplicaHandshakeDoneFn};
use partitions::FatalCommit;
use server_common::sharding::{IggyNamespace, METADATA_GROUP};
use server_common::{Message, MessageBag};
+use std::future::poll_fn;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
+use std::task::Poll;
/// How often the shard pump drives `VsrConsensus::tick`.
///
@@ -306,7 +308,8 @@ where
// `select_biased!`, not `select!`: the unbiased macro draws its
// arm order from a process-random thread-local PRNG, which the
// deterministic simulator cannot seed. The listed order is the
- // intended priority anyway: stop, then tick, then frames.
+ // intended priority anyway: stop, then tick, redispatch, then
+ // newly received frames.
futures::select_biased! {
_ = stop.recv().fuse() => break,
() = consensus_tick.as_mut() => {
@@ -348,6 +351,27 @@ where
self.apply_reconcile_ops();
consensus_tick.set(rearm_tick());
}
+ (message, provenance) = poll_fn(|_| {
+ self.pop_redispatched_frame().map_or(Poll::Pending, Poll::Ready)
+ }).fuse() => {
+ // One frame per select iteration. Ranking this arm above
+ // the inbox preserves park order without making a full
+ // queue stall ticks and commit broadcasts for other groups.
+ self.dispatch_message(message, Some(provenance)).await;
+ // A request handled by a solo primary self-acks here. If
+ // loopback waited for another inbox frame, the request
+ // would remain uncommitted indefinitely on a quiet shard.
+ self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await;
+ self.apply_reconcile_ops();
+ // Same guaranteed reply-lane service as the inbox arm: this
+ // arm outranks both lanes, so a deep drain would otherwise
+ // starve forwarded client 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;
+ }
+ }
frame = self.inbox.recv().fuse() => {
match frame {
Ok(frame) => {
@@ -406,29 +430,9 @@ where
// 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;
- }
- }
- }
- }
- 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_queued_frames_for_shutdown(&mut loopback_buf, &mut namespace_scratch)
+ .await;
}
if fatal.is_some() {
@@ -459,6 +463,59 @@ where
fatal
}
+ /// Process queued work after the select loop has stopped. Redispatch keeps
+ /// its live-pump rank over the inbox, and every delivered frame gets its
+ /// loopback before another frame can run.
+ #[allow(clippy::future_not_send)]
+ async fn drain_queued_frames_for_shutdown(
+ &self,
+ loopback_buf: &mut Vec<Message<GenericHeader>>,
+ namespace_scratch: &mut Vec<IggyNamespace>,
+ ) -> Option<FatalCommit>
+ where
+ B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
+ {
+ loop {
+ while let Some((message, provenance)) = self.pop_redispatched_frame() {
+ self.dispatch_message(message, Some(provenance)).await;
+ self.process_loopback(loopback_buf, namespace_scratch).await;
+ self.apply_reconcile_ops();
+ if let Some(fault) = self.first_partition_commit_fault() {
+ return Some(fault);
+ }
+ }
+ let Ok(frame) = self.inbox.try_recv() else {
+ break;
+ };
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ self.process_loopback(loopback_buf, namespace_scratch).await;
+ self.apply_reconcile_ops();
+ if let Some(fault) = self.first_partition_commit_fault() {
+ return Some(fault);
+ }
+ }
+ }
+
+ // Retired, not delivered: the pump is going away, so a staged frame has
+ // no later arm to reach the plane through. This runs before the reply
+ // lane drain, which is what carries client denies out.
+ self.retire_redispatched_frames();
+ 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() {
+ return Some(fault);
+ }
+ }
+ }
+ None
+ }
+
/// First partition commit fault currently fenced on this shard.
///
/// The regular path observes faults in `tick_partitions`. This scan covers
@@ -527,6 +584,10 @@ where
async fn process_lifecycle(&self, payload: LifecycleFrame)
where
B: MessageBus + 'static,
+ MJ: JournalHandle,
+ <MJ as JournalHandle>::Target:
+ Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ M: RestorableMetadataStm,
{
match payload {
LifecycleFrame::ReplicaInboundSetup { fd, slot } => {
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 7b722eb9b..fda8e5bcf 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -50,8 +50,8 @@ use replica::{Replica, SIM_INBOX_CAPACITY, new_shard};
use seeds::SimSeeds;
use server_common::Message;
use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
-use shard::CONSENSUS_TICK_INTERVAL;
use shard::shards_table::{ShardsTable, calculate_shard_assignment};
+use shard::{CONSENSUS_TICK_INTERVAL, PartitionMaterialisation};
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
@@ -902,12 +902,13 @@ impl Simulator {
}
}
- /// Lost-wake tripwire. At executor quiescence every live pump must have drained
- /// its inbox; a non-empty one on a non-crashed replica means a frame reached the
- /// channel without waking the target pump. Every pump holds a standing
+ /// Lost-work tripwire. At executor quiescence every live pump must have drained
+ /// both inbox lanes and its parked-frame redispatch queue. A non-empty lane on a
+ /// non-crashed replica means a frame reached the channel without waking the
+ /// target pump. A non-empty redispatch queue means a materialisation staged work
+ /// without the pump's priority drain completing. Every pump holds a standing
/// `CONSENSUS_TICK_INTERVAL` timer, so the next `advance_time` would re-poll and
- /// silently drain it, masking the exact wake-loss class this harness exists to
- /// catch. Trip here instead.
+ /// silently drain either case, masking the fault. Trip here instead.
///
/// Incomplete by construction: only catches a lost wakeup while the un-woken
/// frame is still queued at quiescence, since a later frame that does wake the
@@ -943,6 +944,16 @@ impl Simulator {
self.seed,
self.executor.schedule_hash(),
);
+ let pending_redispatch = shard.redispatched_frame_count();
+ assert_eq!(
+ pending_redispatch,
+ 0,
+ "lost redispatch: replica {replica_id} shard {} redispatch queue holds \
+ {pending_redispatch} frame(s) at quiescence (seed {:#x}, schedule hash {:#x})",
+ shard.id,
+ self.seed,
+ self.executor.schedule_hash(),
+ );
}
}
}
@@ -1459,7 +1470,7 @@ fn materialise_partition(
recovered_state,
retained,
restore_frontier,
- created_view,
+ PartitionMaterialisation::new(epoch, created_view),
);
for shard in &replica.shards {
shard.shards_table().insert(
@@ -2432,11 +2443,10 @@ mod tests {
/// behind VSR retransmit.
///
/// `park_overflow` is deliberately NOT excluded. The reconciler is unwired here
- /// (`init_partition` mirrors its outcome directly), so nothing drains the park
- /// buffer and a parked frame is never re-dispatched or swept. Non-zero
- /// `park_overflow` therefore means frames shed for a namespace that will never
- /// materialise, which is the fault class this catches rather than the
- /// back-pressure it would be in production.
+ /// and only an explicit `init_partition` mirrors its materialisation outcome and
+ /// drains the matching park entry. Non-zero `park_overflow` therefore means
+ /// frames shed for a namespace the driver never materialised, which is the fault
+ /// class this catches rather than the back-pressure it would be in production.
///
/// Same reason the park buffer must be empty at quiescence: a frame still parked
/// has no drainer, so it is neither delivered nor answered.
@@ -2451,7 +2461,8 @@ mod tests {
assert!(
shard.parked_namespaces().is_empty(),
"replica {replica_idx} shard {shard_idx} left partition frames parked; the \
- simulator wires no reconciler, so nothing will deliver or answer them"
+ simulator wires no reconciler, so only explicit materialisation can \
+ deliver or answer them"
);
}
}
@@ -2539,6 +2550,222 @@ mod tests {
);
}
+ fn successful_send_reply_count(replies: &[Message<ReplyHeader>]) -> usize {
+ replies
+ .iter()
+ .filter(|reply| {
+ reply.header().operation == iggy_binary_protocol::Operation::SendMessages
+ && reply.header().status == 0
+ })
+ .count()
+ }
+
+ fn retained_prepare(
+ sim: &Simulator,
+ replica: usize,
+ namespace: IggyNamespace,
+ op: u64,
+ ) -> Message<iggy_binary_protocol::PrepareHeader> {
+ let partitions = sim.replicas[replica]
+ .partition_shard(namespace)
+ .plane
+ .partitions();
+ let partition = partitions.get_by_ns(&namespace).expect("partition");
+ let stored = partition
+ .log
+ .journal()
+ .inner
+ .repair_entry(op)
+ .unwrap_or_else(|| panic!("replica {replica} retains prepare op {op} for repair"));
+ Message::<iggy_binary_protocol::PrepareHeader>::try_from(server_common::iobuf::Owned::<
+ { server_common::MESSAGE_ALIGN },
+ >::copy_from_slice(
+ stored.as_slice()
+ ))
+ .expect("journaled prepare remains a valid frame")
+ }
+
+ /// Leave one backup without the partition while the primary commits two
+ /// sends through the other backup. The lagging replica parks both replicated
+ /// prepares. A third prepare is withheld from it, then placed on its inbox
+ /// after simulator materialisation stages the parked prefix. Running the pump
+ /// without advancing its tick forces the inbox arm to drain the prefix first.
+ fn parked_prepare_redispatch_trace(seed: u64) -> (usize, Vec<PartitionOffsets>, u64) {
+ const CLIENT_ID: u128 = 1;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: 3,
+ client_count: 1,
+ seed,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(3, 1, std::iter::once(CLIENT_ID), network_opts);
+ let namespace = IggyNamespace::new(0, 0, 0);
+ sim.seed_stream_topic_partition(namespace);
+ let created_view = sim.partition_created_views[&namespace];
+
+ // Replica 0 is the view-0 primary. Replica 2 supplies quorum while
+ // replica 1 has committed metadata but no local partition yet.
+ materialise_partition(&sim.replicas[0], namespace, false, created_view);
+ materialise_partition(&sim.replicas[2], namespace, false, created_view);
+
+ let client = SimClient::new(CLIENT_ID);
+ sim.shell_login(&client);
+ for payload in [
+ Bytes::from_static(b"parked-redispatch-0"),
+ Bytes::from_static(b"parked-redispatch-1"),
+ ] {
+ let request = client.send_messages(namespace, std::slice::from_ref(&payload));
+ sim.submit_request(CLIENT_ID, 0, request.into_generic());
+ }
+
+ let lagging_shard = Rc::clone(&sim.replicas[1].shards[0]);
+ let mut successful_replies = 0usize;
+ let mut parked = 0usize;
+ for _ in 0..500 {
+ successful_replies += successful_send_reply_count(&sim.step());
+ parked = lagging_shard.parked_frame_count(namespace);
+ if successful_replies >= 2 && parked >= 2 {
+ break;
+ }
+ }
+ assert!(
+ successful_replies >= 2,
+ "primary never committed both sends"
+ );
+ assert!(parked >= 2, "lagging backup never parked both prepares");
+
+ // Keep the third prepare off replica 1's network path. Replica 0 can
+ // still commit it through replica 2, after which its journal supplies
+ // the exact wire frame to put on the lagging backup's inbox below.
+ sim.network.process_disable(ProcessId::Replica(1));
+ let third = client.send_messages(namespace, &[Bytes::from_static(b"parked-redispatch-2")]);
+ sim.submit_request(CLIENT_ID, 0, third.into_generic());
+ for _ in 0..500 {
+ successful_replies += successful_send_reply_count(&sim.step());
+ if successful_replies >= 3 {
+ break;
+ }
+ }
+ assert!(
+ successful_replies >= 3,
+ "primary never committed the later send"
+ );
+ let later_prepare = retained_prepare(&sim, 0, namespace, 3);
+ sim.network.process_enable(ProcessId::Replica(1));
+
+ materialise_partition(&sim.replicas[1], namespace, false, created_view);
+ assert_eq!(lagging_shard.parked_frame_count(namespace), 0);
+ assert_eq!(
+ lagging_shard.redispatched_frame_count(),
+ parked,
+ "materialisation must move every parked prepare to the pump queue"
+ );
+
+ // No virtual-time advance here. The materialisation marker and inbox
+ // wake poll the pump, whose ranked redispatch arm must put ops 1 and 2
+ // ahead of this op 3 frame.
+ lagging_shard.dispatch(later_prepare.into_generic());
+ sim.run_pumps();
+ let lagging_holds_later = sim.replicas[1]
+ .partition_shard(namespace)
+ .plane
+ .partitions()
+ .get_by_ns(&namespace)
+ .is_some_and(|partition| partition.log.journal().inner.header_by_op(3).is_some());
+ assert!(
+ lagging_holds_later,
+ "inbox op 3 reached the gap check before the redispatched prefix"
+ );
+
+ let expected = sim
+ .offsets(0, namespace)
+ .expect("primary partition offsets");
+ let mut converged = false;
+ for _ in 0..500 {
+ sim.step();
+ converged = (0..3).all(|replica| sim.offsets(replica, namespace) == Some(expected));
+ if converged {
+ break;
+ }
+ }
+ assert!(
+ converged,
+ "redispatched prepares did not converge the backup"
+ );
+ assert_eq!(lagging_shard.redispatched_frame_count(), 0);
+ assert_no_frame_drops(&sim);
+
+ let offsets = (0..3)
+ .map(|replica| sim.offsets(replica, namespace).expect("partition offsets"))
+ .collect();
+ (parked, offsets, sim.schedule_hash())
+ }
+
+ #[test]
+ fn parked_prepare_redispatch_is_seed_replayable() {
+ let first = parked_prepare_redispatch_trace(0x5CED_4005);
+ let second = parked_prepare_redispatch_trace(0x5CED_4005);
+ assert_eq!(first, second, "same seed diverged on parked redispatch");
+ }
+
+ #[test]
+ fn redispatched_solo_request_commits_without_another_inbox_frame() {
+ const CLIENT_ID: u128 = 1;
+
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: 1,
+ client_count: 1,
+ seed: 0x5CED_4006,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(1, 1, std::iter::once(CLIENT_ID), network_opts);
+ let namespace = IggyNamespace::new(0, 0, 0);
+ sim.seed_stream_topic_partition(namespace);
+
+ let client = SimClient::new(CLIENT_ID);
+ sim.shell_login(&client);
+ let request = client.send_messages(namespace, &[Bytes::from_static(b"solo-redispatch")]);
+ let shard = Rc::clone(&sim.replicas[0].shards[0]);
+ let request = server_common::MessageBag::try_from(request.into_generic())
+ .expect("valid send request");
+ futures::executor::block_on(shard.on_message(request));
+ assert_eq!(
+ shard.parked_frame_count(namespace),
+ 1,
+ "the request must park before materialisation"
+ );
+
+ sim.init_partition(namespace);
+ assert_eq!(
+ shard.redispatched_frame_count(),
+ 1,
+ "materialisation must stage and wake the parked request"
+ );
+
+ // No network or virtual-time step follows materialisation. The ranked
+ // redispatch arm must deliver this request and process its self-ack in
+ // the same pump iteration.
+ sim.run_pumps();
+ let state = sim
+ .partition_consensus_state(0, namespace)
+ .expect("the materialised solo partition has consensus state");
+ assert_eq!(state.commit_min, 1, "the self-ack must commit the request");
+ assert_eq!(shard.redispatched_frame_count(), 0);
+ }
+
/// The dispatch shell's reason to exist: detect the PR #3557 async-concurrency
/// class, a partition reference held across an `.await` while a sibling task
/// mutates the partitions vec. Under the deterministic executor a parked read with
@@ -2698,7 +2925,14 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET); // borrow acquired; task parks
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None, None, false, 0);
+ grow.init_partition(
+ ns_grow,
+ None,
+ None,
+ None,
+ false,
+ PartitionMaterialisation::new(0, 0),
+ );
});
executor.run_until_stalled(POLL_BUDGET); // grow while the borrow is live
}))
@@ -2733,7 +2967,14 @@ mod tests {
executor.run_until_stalled(POLL_BUDGET);
let grow = Rc::clone(&sim.replicas[0].shards[0]);
executor.spawn(async move {
- grow.init_partition(ns_grow, None, None, None, false, 0);
+ grow.init_partition(
+ ns_grow,
+ None,
+ None,
+ None,
+ false,
+ PartitionMaterialisation::new(0, 0),
+ );
});
executor.run_until_stalled(POLL_BUDGET);