You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/05/22 07:27:56 UTC
(iggy) branch crossfire-server-ng updated (e0067ca9a -> f2d557ead)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch crossfire-server-ng
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit e0067ca9a feat(cluster): wire multi-shard cross-shard communication for server-ng
add b7ef1a5b1 ci: remove team from dependabot assignees field (#3295)
add 233788328 chore(deps): bump mongodb from 3.6.0 to 3.7.0 in the minor-and-patch group (#3296)
add 26357402a chore(deps-dev): bump @sveltejs/kit from 2.58.0 to 2.60.1 in /web (#3297)
new f2d557ead feat(cluster): wire multi-shard cross-shard communication for server-ng
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (e0067ca9a)
\
N -- N -- N refs/heads/crossfire-server-ng (f2d557ead)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/dependabot.yml | 4 ---
Cargo.lock | 85 +++++++++++++++++++++++++++++++++++---------------
Cargo.toml | 2 +-
web/package-lock.json | 10 +++---
web/package.json | 2 +-
5 files changed, 66 insertions(+), 37 deletions(-)
(iggy) 01/01: feat(cluster): wire multi-shard cross-shard communication for server-ng
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch crossfire-server-ng
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f2d557eadbe595d9be6f6fdbdc73fc145736063e
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu May 14 09:56:38 2026 +0200
feat(cluster): wire multi-shard cross-shard communication for server-ng
core/server-ng bootstrapped exactly one shard with SHARD_ID=0 and
senders=vec![sender] hardcoded; the multi-shard path was dead code.
Cross-shard primitives copied from legacy core/server also did not
fit core/shard's crossfire model (bounded MTx + try_send-or-drop,
fd-transfer coordinator instead of SO_REUSEPORT).
bootstrap() now spawns N OS threads from sharding.cpu_allocation,
each pinned via nix::sched + hwlocality and running its own compio
runtime + IggyMessageBus + IggyShard for the partitions hashed to
that shard. Cross-thread shutdown rides an Arc<AtomicBool> polled by
a per-shard watchdog, since the bus' Shutdown is !Send and cannot be
triggered from the main thread directly. Partial shard-spawn failure
and shard-thread panic now signal cluster-wide shutdown instead of
hanging.
ShardFrame becomes a concrete enum (Consensus + Lifecycle); the R
generic is lifted off IggyShard. Named routers (route_metadata /
route_partition / route_consensus_control) replace the duplicated
MessageBag match blocks, and a debug_assert at pump entry catches
receiver-thread mis-binding that the ctor's assert_sender_ordering
cannot see.
ShardMetrics records frame_drops_total (counter, variant+reason
labels), bumped at every inter-shard try_send rejection and at the
receiver-side delivery failure paths (partition-namespace miss,
ForwardClientSend failure, consensus misroute). Without it,
drop-and-recover under VSR retransmit is operationally
indistinguishable from a livelock. The counter is atomic, so it is
safe to bump from !Send compio reactor contexts. shard_id is NOT a
label; each shard owns its own Family and the exporter must attach
shard scope at the registry layer.
The legacy shard-mapping broadcast subsystem (periodic snapshot
refresh task, three-state MappingSlot table, ReplicaMappingUpdate /
ReplicaMappingClear frames) is retired entirely. Cross-shard replica
routing now flows through the cluster-shared ReplicaOwnerTable: the
owning shard's installer stamps its slot on a successful registry
insert and CAS-clears it on disconnect, so every bus' send_to_replica
slow path reads authoritative state with no broadcast or reconcile
loop. Builder accepts the coordinator at ctor; IggyShard stays
immutable post-construction.
message_bus forward-fn types widen to carry replica/client id, and
send_to_replica routes via the shared owner table so non-zero shards
reconcile against shared state rather than shard 0's private bus.
Metadata recovery: shard 0 owns the WAL writer (via
PrepareJournal::open) and the metadata consensus replica. Peer shards
never open the WAL; they receive a MetadataHandoff::Waiter factory
bundle broadcast by shard 0 over a single-shot crossfire channel and
reconstruct their mux_stm from the in-memory snapshot it carries.
broadcast_metadata_bundle / await_metadata_bundle poll a shutdown
flag so a shard 0 that panics before broadcasting cannot strand
peers.
PrepareJournal::drain advances snapshot_op only AFTER the WAL rewrite
sequence (tmp create -> write -> fsync -> rename -> fsync parent ->
reopen) is durable. Advancing earlier would leave snapshot_op past
entries still present on disk on any ? failure, letting a future
append() pass the slot collision check and silently evict a live
entry from the index.
server-ng's main installs Ctrl-C with abort-on-failure: without a
working SIGINT handler the shutdown flag would never flip and shard
threads would park indefinitely.
InvalidShardsCount split into ShardsCountZero (allocator produced
zero shards) and ShardsCountOverflow (count past OWNER_NONE - 1) so
the operator-facing message names the actual failure mode.
Bootstrap namespace materialization stream-filters inside read():
heavy (Arc<TopicStats>, Partition) clones are kept only for partitions
owned by this shard; non-owning entries are inserted straight into
the shards_table from the closure. Partial-spawn survivor join fans
out joiner threads so total wait is bounded by the slowest shard,
not N * SHARD_SHUTDOWN_TIMEOUT.
TODOs left in place for follow-ups:
- shutdown watchdog is still .detach()'d; tracking it on the bus
self-deadlocks because the watchdog drives bus.shutdown(). Needs a
shared core/task_registry crate ported from core/server.
- primary IggyMetadata::on_prepare pre-advances sequencer + checksum
in push_prepare_entry before the journal append; an append failure
leaves state pointing at a phantom op. Needs a rollback path in
push_prepare_entry.
Storage milestones (durable PartitionJournal, durable
(view, commit_op) watermark) and SDK (client_id, request_id)
durability across reconnect remain out of scope; tracked separately.
---
Cargo.lock | 4 +-
core/common/src/sharding/mod.rs | 3 +-
core/common/src/sharding/namespace.rs | 61 +-
core/configs/src/server_config/sharding.rs | 124 +-
core/configs/src/server_config/validators.rs | 128 ++-
core/consensus/src/impls.rs | 11 +
core/consensus/src/metadata_helpers.rs | 4 +
core/consensus/src/plane_helpers.rs | 8 +
core/journal/src/file_storage.rs | 3 +-
core/journal/src/prepare_journal.rs | 530 ++++++++-
core/message_bus/src/client_listener/mod.rs | 2 +-
core/message_bus/src/client_listener/tcp.rs | 2 +-
core/message_bus/src/client_listener/tcp_tls.rs | 2 +-
core/message_bus/src/client_listener/ws.rs | 8 +-
core/message_bus/src/connector.rs | 11 +-
core/message_bus/src/error.rs | 25 +-
core/message_bus/src/installer/mod.rs | 18 +-
core/message_bus/src/installer/replica.rs | 52 +-
core/message_bus/src/installer/tcp_tls.rs | 2 +-
core/message_bus/src/installer/wss.rs | 2 +-
core/message_bus/src/lib.rs | 592 ++++++++--
core/message_bus/src/transports/mod.rs | 2 +-
core/metadata/src/impls/metadata.rs | 79 +-
core/metadata/src/impls/recovery.rs | 21 +-
core/metadata/src/stm/mod.rs | 189 +++-
core/metadata/src/stm/mux.rs | 63 ++
core/partitions/src/iggy_partition.rs | 4 +-
core/server-ng/Cargo.toml | 3 +-
core/server-ng/config.toml | 17 +-
core/server-ng/src/bootstrap.rs | 1366 +++++++++++++++++++----
core/server-ng/src/main.rs | 69 +-
core/server-ng/src/server_error.rs | 147 +++
core/shard/Cargo.toml | 2 +
core/shard/src/builder.rs | 206 ++--
core/shard/src/config.rs | 11 -
core/shard/src/coordinator.rs | 555 ++++-----
core/shard/src/lib.rs | 404 ++++---
core/shard/src/metrics.rs | 258 +++++
core/shard/src/router.rs | 442 ++++----
core/shard/src/shards_table.rs | 6 +-
core/simulator/src/bus.rs | 16 +
core/simulator/src/lib.rs | 6 +-
42 files changed, 4120 insertions(+), 1338 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index cc4de3f61..a19c63e07 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11388,6 +11388,7 @@ dependencies = [
"compio",
"configs",
"consensus",
+ "crossfire",
"ctrlc",
"cyper",
"cyper-axum",
@@ -11396,7 +11397,6 @@ dependencies = [
"err_trail",
"error_set",
"figlet-rs",
- "flume 0.12.0",
"fs2",
"hash32 1.0.0",
"human-repr",
@@ -11518,6 +11518,8 @@ dependencies = [
"metadata",
"papaya",
"partitions",
+ "prometheus-client",
+ "thiserror 2.0.18",
"tracing",
]
diff --git a/core/common/src/sharding/mod.rs b/core/common/src/sharding/mod.rs
index 051802ecf..761e41d04 100644
--- a/core/common/src/sharding/mod.rs
+++ b/core/common/src/sharding/mod.rs
@@ -22,7 +22,8 @@ mod shard_id;
pub use local_idx::LocalIdx;
pub use namespace::{
- IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, NamespaceCapacityError, PARTITION_BITS,
+ IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE,
+ NamespaceCapacityError, PACKED_NAMESPACE_BITS, PACKED_NAMESPACE_MAX, PARTITION_BITS,
PARTITION_MASK, PARTITION_SHIFT, STREAM_BITS, STREAM_MASK, STREAM_SHIFT, TOPIC_BITS,
TOPIC_MASK, TOPIC_SHIFT,
};
diff --git a/core/common/src/sharding/namespace.rs b/core/common/src/sharding/namespace.rs
index cf13db5dc..3cb108247 100644
--- a/core/common/src/sharding/namespace.rs
+++ b/core/common/src/sharding/namespace.rs
@@ -51,6 +51,26 @@ pub const PARTITION_MASK: u64 = (1u64 << PARTITION_BITS) - 1;
pub const TOPIC_MASK: u64 = (1u64 << TOPIC_BITS) - 1;
pub const STREAM_MASK: u64 = (1u64 << STREAM_BITS) - 1;
+/// Total bits used by the packed `IggyNamespace` layout (stream | topic | partition).
+/// Any `u64` whose set bits all sit within this range is a legal packable namespace.
+pub const PACKED_NAMESPACE_BITS: u32 = STREAM_BITS + TOPIC_BITS + PARTITION_BITS;
+
+/// Largest `u64` value that can be produced by [`IggyNamespace::new`].
+/// Equivalent to `(1 << PACKED_NAMESPACE_BITS) - 1`.
+pub const PACKED_NAMESPACE_MAX: u64 = (1u64 << PACKED_NAMESPACE_BITS) - 1;
+
+/// Reserved consensus-namespace identifier for the cluster's metadata replica.
+///
+/// The packed `IggyNamespace` layout uses only bits `0..PACKED_NAMESPACE_BITS`,
+/// so the top bit is guaranteed to be unreachable from any
+/// `IggyNamespace::new(stream, topic, partition).inner()`. This lets routers
+/// distinguish metadata's single global consensus group from per-partition
+/// consensus groups by value alone, without needing a wire-format scope tag.
+///
+/// `IggyNamespace::new(0, 0, 0).inner() == 0` is a legal partition, so `0`
+/// cannot serve as a sentinel.
+pub const METADATA_CONSENSUS_NAMESPACE: u64 = 1u64 << 63;
+
#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
pub enum NamespaceCapacityError {
#[error("max_streams must be greater than 0")]
@@ -112,6 +132,15 @@ impl IggyNamespace {
Self(value)
}
+ /// Returns `true` when `value` could have been produced by [`Self::new`].
+ /// False for the metadata sentinel and for any `u64` with bits set above
+ /// the packed range.
+ #[inline]
+ #[must_use]
+ pub const fn is_packable(value: u64) -> bool {
+ value <= PACKED_NAMESPACE_MAX
+ }
+
pub fn validate_capacity(
max_streams: usize,
max_topics: usize,
@@ -157,7 +186,37 @@ impl IggyNamespace {
#[cfg(test)]
mod tests {
- use super::{IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, NamespaceCapacityError};
+ use super::{
+ IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE,
+ NamespaceCapacityError, PACKED_NAMESPACE_BITS, PACKED_NAMESPACE_MAX,
+ };
+
+ // Both static facts (sentinel above the packed range; bit layout matches
+ // module constants) are checked at compile time so any future tweak to
+ // STREAM_BITS / TOPIC_BITS / PARTITION_BITS that closes the gap between
+ // the packed range and the sentinel will fail to build.
+ const _: () = {
+ assert!(METADATA_CONSENSUS_NAMESPACE > PACKED_NAMESPACE_MAX);
+ assert!(PACKED_NAMESPACE_BITS == 12 + 12 + 20);
+ assert!(PACKED_NAMESPACE_MAX == (1u64 << PACKED_NAMESPACE_BITS) - 1);
+ };
+
+ #[test]
+ fn metadata_sentinel_cannot_collide_with_any_packable_namespace() {
+ assert!(!IggyNamespace::is_packable(METADATA_CONSENSUS_NAMESPACE));
+
+ // The (0, 0, 0) corner is intentionally a legal partition, which is
+ // precisely why `0` is unsuitable as the metadata sentinel.
+ let zero = IggyNamespace::new(0, 0, 0);
+ assert_eq!(zero.inner(), 0);
+ assert!(IggyNamespace::is_packable(zero.inner()));
+ assert_ne!(zero.inner(), METADATA_CONSENSUS_NAMESPACE);
+
+ // Maximum packable triple stays inside the packed range.
+ let max = IggyNamespace::new(MAX_STREAMS - 1, MAX_TOPICS - 1, MAX_PARTITIONS - 1);
+ assert!(IggyNamespace::is_packable(max.inner()));
+ assert_ne!(max.inner(), METADATA_CONSENSUS_NAMESPACE);
+ }
#[test]
fn validates_default_namespace_capacity() {
diff --git a/core/configs/src/server_config/sharding.rs b/core/configs/src/server_config/sharding.rs
index 084c984e1..1f52b24fd 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -16,16 +16,138 @@
* under the License.
*/
+use iggy_common::IggyDuration;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde_with::{DisplayFromStr, serde_as};
use std::str::FromStr;
+use std::time::Duration;
use configs::ConfigEnv;
-#[derive(Debug, Deserialize, Serialize, Default, ConfigEnv)]
+/// Default capacity of the per-shard inter-shard inbox channel. Sized
+/// comfortably above the consensus working set, which is roughly
+/// `PIPELINE_PREPARE_QUEUE_MAX (= 8) * replica_count * directions`
+/// frames in flight per shard, without allowing a runaway producer to
+/// eat unbounded memory. Tunable via `[system.sharding] inbox_capacity`
+/// in TOML.
+///
+/// The capacity must also absorb the worst-case cross-shard client
+/// Reply burst. Unlike consensus frames, client Replies have no VSR
+/// retransmit path: a Reply lost on full inbox is gone and the client
+/// times out. A reasonable lower bound is
+/// `max_inflight_client_requests / num_shards` (assuming requests are
+/// distributed evenly across owning shards) plus the consensus
+/// headroom above.
+///
+/// Consensus frames and client-reply forwards share this one channel,
+/// so the two headrooms are not independent: a consensus burst or
+/// retransmit storm can fill the inbox with consensus frames exactly
+/// when a client Reply needs the space. A single `inbox_capacity` knob
+/// cannot isolate the two frame classes - size it for the sum of both
+/// worst cases occurring together. Watch the drop-site `tracing` logs
+/// (and, once a per-shard exporter lands, the `frame_drops_total`
+/// `{variant="forward_client_send"}` counter) to detect when the bound
+/// is too low in production.
+pub const DEFAULT_INBOX_CAPACITY: usize = 1024;
+
+/// Maximum permitted per-shard inbox depth. The channel is allocated
+/// up-front per shard, so a runaway value here OOMs the process at boot.
+/// `1 << 20` (~1M frames) is several orders of magnitude above any
+/// realistic backpressure target and still fits comfortably in process
+/// address space.
+pub const INBOX_CAPACITY_MAX: usize = 1 << 20;
+
+/// Default bus shutdown drain timeout. Sized larger than typical TCP RTT
+/// times in-flight write-batch so writers receive their full last
+/// `write_vectored_all` budget before the connection registry kicks in.
+pub const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// Default watchdog poll cadence for the cross-thread shutdown flag.
+/// 50ms keeps Ctrl-C latency operator-visible without measurable wakeup
+/// overhead.
+pub const DEFAULT_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(50);
+
+/// Hard upper bound on `shutdown_drain_timeout`. A drain that never
+/// completes wedges process exit; capping at 10 minutes guarantees the
+/// watchdog eventually force-tears the bus even with a pathological
+/// config typo.
+pub const SHUTDOWN_DRAIN_TIMEOUT_MAX: Duration = Duration::from_secs(600);
+
+/// Hard upper bound on `shutdown_poll_interval`. A pollerinterval longer
+/// than the drain timeout makes the flag effectively unobservable; cap
+/// at 5s so Ctrl-C latency stays bounded regardless of config.
+pub const SHUTDOWN_POLL_INTERVAL_MAX: Duration = Duration::from_secs(5);
+
+const fn default_inbox_capacity() -> usize {
+ DEFAULT_INBOX_CAPACITY
+}
+
+fn default_shutdown_drain_timeout() -> IggyDuration {
+ IggyDuration::new(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT)
+}
+
+fn default_shutdown_poll_interval() -> IggyDuration {
+ IggyDuration::new(DEFAULT_SHUTDOWN_POLL_INTERVAL)
+}
+
+#[serde_as]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct ShardingConfig {
#[serde(default)]
#[config_env(leaf)]
pub cpu_allocation: CpuAllocation,
+ /// Per-shard inter-shard inbox channel capacity. Bounded by design.
+ /// Drops on full inbox of consensus frames are recovered by VSR
+ /// retransmit. Drops of cross-shard client Reply frames are terminal:
+ /// the client never receives the reply (no in-protocol retransmit).
+ /// Both frame classes share this one channel, so a consensus burst
+ /// can starve client-reply forwards: size against the worst-case sum
+ /// of consensus working set + peak client-reply fan-out per shard
+ /// occurring together; see `DEFAULT_INBOX_CAPACITY` for the
+ /// rationale. Used by `core/server-ng`; the legacy server uses its
+ /// own hard-coded inbox sizing.
+ ///
+ // TODO(hubcio): split into two priority lanes - one bounded queue for
+ // consensus frames (drops recovered by VSR retransmit) and one for
+ // client `Reply` frames (drops terminal, must be sized for worst-case
+ // fan-out). Current single-channel design is the minimum-viable
+ // wiring so `frame_drops_total{variant,reason}` surfaces under load
+ // and yields real numbers to size the split against.
+ #[serde(default = "default_inbox_capacity")]
+ pub inbox_capacity: usize,
+ /// Wall-clock budget for a single shard's bus drain on shutdown.
+ /// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog
+ /// and the parallel-join survivor path. Sized larger than typical
+ /// TCP RTT times in-flight write-batch so writers receive their full
+ /// last `write_vectored_all` budget before the connection registry
+ /// force-tears the bus. Slow-fsync hosts may need to extend this past
+ /// the default; the cap is `SHUTDOWN_DRAIN_TIMEOUT_MAX` so a config
+ /// typo cannot wedge process exit.
+ #[serde(default = "default_shutdown_drain_timeout")]
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub shutdown_drain_timeout: IggyDuration,
+ /// Poll cadence for the cross-thread shutdown flag and for the
+ /// `await_metadata_bundle` / `broadcast_metadata_bundle` poll loops.
+ /// Trades off Ctrl-C latency against idle wakeup cost; the default
+ /// keeps shutdown observably prompt without measurable scheduler
+ /// overhead. Capped at `SHUTDOWN_POLL_INTERVAL_MAX` so the flag
+ /// remains effectively observable regardless of config.
+ #[serde(default = "default_shutdown_poll_interval")]
+ #[serde_as(as = "DisplayFromStr")]
+ #[config_env(leaf)]
+ pub shutdown_poll_interval: IggyDuration,
+}
+
+impl Default for ShardingConfig {
+ fn default() -> Self {
+ Self {
+ cpu_allocation: CpuAllocation::default(),
+ inbox_capacity: DEFAULT_INBOX_CAPACITY,
+ shutdown_drain_timeout: default_shutdown_drain_timeout(),
+ shutdown_poll_interval: default_shutdown_poll_interval(),
+ }
+ }
}
#[derive(Debug, Clone, PartialEq, Default)]
diff --git a/core/configs/src/server_config/validators.rs b/core/configs/src/server_config/validators.rs
index 1512961d6..36405c72e 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -23,7 +23,10 @@ use super::server::{
DataMaintenanceConfig, MessageSaverConfig, MessagesMaintenanceConfig, TelemetryConfig,
};
use super::server::{MemoryPoolConfig, PersonalAccessTokenConfig, ServerConfig};
-use super::sharding::{CpuAllocation, ShardingConfig};
+use super::sharding::{
+ CpuAllocation, INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
+ ShardingConfig,
+};
use super::system::SegmentConfig;
use super::system::{CompressionConfig, LoggingConfig, PartitionConfig};
use crate::ConfigurationError;
@@ -379,6 +382,66 @@ impl Validatable<ConfigurationError> for MemoryPoolConfig {
impl Validatable<ConfigurationError> for ShardingConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
+ if self.inbox_capacity == 0 {
+ eprintln!(
+ "Invalid sharding configuration: inbox_capacity must be > 0 (crossfire silently \
+ rounds 0 to 1, masking config errors)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if self.inbox_capacity > INBOX_CAPACITY_MAX {
+ eprintln!(
+ "Invalid sharding configuration: inbox_capacity {} exceeds the {} cap (each \
+ shard preallocates a channel of this size; oversizing here OOMs the process at \
+ boot)",
+ self.inbox_capacity, INBOX_CAPACITY_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let drain = self.shutdown_drain_timeout.get_duration();
+ if drain.is_zero() {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_drain_timeout must be > 0 (a zero \
+ budget force-tears the bus mid-WAL-fsync on every shutdown)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if drain > SHUTDOWN_DRAIN_TIMEOUT_MAX {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_drain_timeout {:?} exceeds the {:?} \
+ cap (an unbounded drain wedges process exit on bus stall)",
+ drain, SHUTDOWN_DRAIN_TIMEOUT_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
+ let poll = self.shutdown_poll_interval.get_duration();
+ if poll.is_zero() {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval must be > 0 (a zero \
+ cadence busy-loops every shard's watchdog and metadata-handoff poller)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if poll > SHUTDOWN_POLL_INTERVAL_MAX {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval {:?} exceeds the {:?} \
+ cap (a coarse cadence stalls Ctrl-C handling and metadata handoff abort)",
+ poll, SHUTDOWN_POLL_INTERVAL_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if poll > drain {
+ eprintln!(
+ "Invalid sharding configuration: shutdown_poll_interval {:?} must be <= \
+ shutdown_drain_timeout {:?} (a poll cadence coarser than the drain budget makes \
+ the shutdown flag effectively unobservable)",
+ poll, drain
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
let available_cpus = available_parallelism()
.map_err(|_| {
eprintln!("Failed to detect available CPU cores");
@@ -699,3 +762,66 @@ mod cluster_shards_count_determinism_tests {
assert!(host_dependent_cpu_allocation(&CpuAllocation::NumaAware(numa)).is_ok());
}
}
+
+#[cfg(test)]
+mod sharding_shutdown_knob_tests {
+ use super::*;
+ use crate::server_config::sharding::{SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX};
+ use iggy_common::IggyDuration;
+ use std::time::Duration;
+
+ #[test]
+ fn defaults_validate() {
+ assert!(ShardingConfig::default().validate().is_ok());
+ }
+
+ #[test]
+ fn zero_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout: IggyDuration::new(Duration::ZERO),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn over_cap_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout: IggyDuration::new(
+ SHUTDOWN_DRAIN_TIMEOUT_MAX + Duration::from_secs(1),
+ ),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn zero_poll_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_poll_interval: IggyDuration::new(Duration::ZERO),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn over_cap_poll_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_poll_interval: IggyDuration::new(
+ SHUTDOWN_POLL_INTERVAL_MAX + Duration::from_secs(1),
+ ),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn poll_greater_than_drain_is_rejected() {
+ let cfg = ShardingConfig {
+ shutdown_drain_timeout: IggyDuration::new(Duration::from_millis(20)),
+ shutdown_poll_interval: IggyDuration::new(Duration::from_millis(50)),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+}
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 653c6503b..e5c8f78bc 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -28,6 +28,7 @@ use iggy_binary_protocol::{
Command2, ConsensusHeader, DoViewChangeHeader, GenericHeader, Message, PrepareHeader,
PrepareOkHeader, ReplyHeader, RequestHeader, StartViewChangeHeader, StartViewHeader,
};
+use iggy_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
use message_bus::IggyMessageBus;
use message_bus::MessageBus;
use std::cell::{Cell, RefCell};
@@ -691,6 +692,16 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>> VsrConsensus<B, P> {
"replica index must be < replica_count"
);
assert!(replica_count >= 1, "need at least 1 replica");
+ // Consensus-control routing distinguishes metadata frames from
+ // partition frames by namespace value: metadata uses the sentinel,
+ // partitions use `IggyNamespace::inner()` which lives strictly
+ // inside the packed range. A namespace outside both ranges would
+ // route to neither and silently warn-drop on every receiving peer.
+ debug_assert!(
+ namespace == METADATA_CONSENSUS_NAMESPACE || IggyNamespace::is_packable(namespace),
+ "VsrConsensus namespace must be METADATA_CONSENSUS_NAMESPACE or a packable \
+ IggyNamespace; got {namespace:#x}"
+ );
// TODO: Verify that XOR-based seeding provides sufficient jitter diversity
// across groups. Consider using a proper hash (e.g., Murmur3) of
// (replica_id, namespace) for production.
diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs
index 66885911e..5904f1049 100644
--- a/core/consensus/src/metadata_helpers.rs
+++ b/core/consensus/src/metadata_helpers.rs
@@ -353,6 +353,10 @@ mod tests {
) -> Result<(), SendError> {
Ok(())
}
+
+ fn set_connection_lost_fn(&self, _f: message_bus::ConnectionLostFn) {}
+ fn set_replica_forward_fn(&self, _f: message_bus::ReplicaForwardFn) {}
+ fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
}
// Register-retry replay: cached IS the register reply; bytes replayed
diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs
index e2a500cae..7e0a89cf4 100644
--- a/core/consensus/src/plane_helpers.rs
+++ b/core/consensus/src/plane_helpers.rs
@@ -472,6 +472,10 @@ mod tests {
) -> Result<(), SendError> {
Ok(())
}
+
+ fn set_connection_lost_fn(&self, _f: message_bus::ConnectionLostFn) {}
+ fn set_replica_forward_fn(&self, _f: message_bus::ReplicaForwardFn) {}
+ fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
}
fn prepare_message(op: u64, parent: u128, checksum: u128) -> Message<PrepareHeader> {
@@ -675,6 +679,10 @@ mod tests {
self.sent.borrow_mut().push((replica, data));
Ok(())
}
+
+ fn set_connection_lost_fn(&self, _f: message_bus::ConnectionLostFn) {}
+ fn set_replica_forward_fn(&self, _f: message_bus::ReplicaForwardFn) {}
+ fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
}
#[test]
diff --git a/core/journal/src/file_storage.rs b/core/journal/src/file_storage.rs
index 6517f5b90..0de1ac673 100644
--- a/core/journal/src/file_storage.rs
+++ b/core/journal/src/file_storage.rs
@@ -31,7 +31,8 @@ pub struct FileStorage {
#[allow(clippy::future_not_send)]
impl FileStorage {
- /// Open or create the file at `path`, setting `write_offset` to current file length.
+ /// Open or create the file at `path` in read-write mode, setting
+ /// `write_offset` to current file length.
///
/// # Errors
/// Returns an I/O error if the file cannot be opened or created.
diff --git a/core/journal/src/prepare_journal.rs b/core/journal/src/prepare_journal.rs
index 2c5b564bc..ea8d2c9bc 100644
--- a/core/journal/src/prepare_journal.rs
+++ b/core/journal/src/prepare_journal.rs
@@ -21,11 +21,11 @@ use compio::io::AsyncWriteAtExt;
use iggy_binary_protocol::consensus::iobuf::Owned;
use iggy_binary_protocol::consensus::message::{MESSAGE_ALIGN, Message};
use iggy_binary_protocol::consensus::{Command2, PrepareHeader};
-use std::cell::{Cell, Ref, RefCell};
+use std::cell::{Cell, OnceCell, Ref, RefCell};
use std::fmt;
use std::io;
use std::ops::RangeInclusive;
-use std::path::Path;
+use std::path::{Path, PathBuf};
const HEADER_SIZE: usize = size_of::<PrepareHeader>();
@@ -44,7 +44,7 @@ const MAX_ENTRY_SIZE: u64 = 64 * 1024 * 1024;
/// still contains them, but they become unreachable for recovery).
///
/// **NOTE:** This number needs to be chosen in balance between number of
-/// entries in [`core::consensus::pipeline_prepare_queue_max`]. Because this number controls
+/// entries in [`consensus::PIPELINE_PREPARE_QUEUE_MAX`]. Because this number controls
/// how many committed but not yet snapshotted entries that the buffer can
/// hold. This may need to be tuned properly.
pub(crate) const SLOT_COUNT: usize = 1024;
@@ -54,12 +54,27 @@ pub(crate) const SLOT_COUNT: usize = 1024;
#[allow(clippy::module_name_repetitions)]
pub enum JournalError {
Io(io::Error),
+ /// Journal entered an irrecoverable state after a partial `drain()`
+ /// failure: the atomic `rename` succeeded but a subsequent step
+ /// (parent-dir fsync or storage reopen) failed, leaving in-memory
+ /// state desynced from on-disk state. Every IO entry point refuses
+ /// further work until the journal is rebuilt by re-opening the WAL.
+ ///
+ /// `stage` names which drain step failed (for diagnostics); `source`
+ /// is the underlying `io::Error` that caused it, kept so operators
+ /// see the original kernel error (`ENOSPC`, `EIO`, ...) rather than
+ /// just a generic poison marker.
+ Poisoned {
+ stage: &'static str,
+ source: io::Error,
+ },
}
impl fmt::Display for JournalError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Io(e) => write!(f, "journal I/O error: {e}"),
+ Self::Poisoned { stage, .. } => write!(f, "journal poisoned at {stage}"),
}
}
}
@@ -68,6 +83,7 @@ impl std::error::Error for JournalError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(e) => Some(e),
+ Self::Poisoned { source, .. } => Some(source),
}
}
}
@@ -101,6 +117,26 @@ pub struct PrepareJournal {
/// an un-snapshotted entry (op > `snapshot_op`) panics and the upper layer must
/// take a snapshot before the journal wraps.
snapshot_op: Cell<u64>,
+ /// Populated once `drain()` has progressed past the atomic `rename`
+ /// and a subsequent step has failed, meaning the on-disk WAL and the
+ /// in-memory index no longer agree. All IO entry points must
+ /// short-circuit with `JournalError::Poisoned` to prevent the next
+ /// `append` from writing into the orphaned old fd or
+ /// `entry`/`entry_at` from serving headers whose offsets reference
+ /// the pre-drain layout.
+ ///
+ /// `OnceCell` (not `Cell<Option<_>>`) because the underlying
+ /// `io::Error` is `!Clone` and the journal is dead after the first
+ /// set; first-write-wins matches the actual semantics and avoids
+ /// `RefCell` borrow-panic risk on the read fast path.
+ poisoned: OnceCell<PoisonState>,
+}
+
+/// Captured cause of journal poisoning. `stage` names the drain step
+/// that tripped; `source` is the original `io::Error` for forensics.
+struct PoisonState {
+ stage: &'static str,
+ source: io::Error,
}
impl fmt::Debug for PrepareJournal {
@@ -108,6 +144,7 @@ impl fmt::Debug for PrepareJournal {
f.debug_struct("PrepareJournal")
.field("write_offset", &self.storage.file_len())
.field("last_op", &self.last_op.get())
+ .field("poisoned", &self.poisoned.get().map(|p| p.stage))
.finish_non_exhaustive()
}
}
@@ -117,9 +154,83 @@ const fn slot_for_op(op: u64) -> usize {
op as usize % SLOT_COUNT
}
+/// Repair a damaged WAL tail by truncating to `pos`, or surface a loud
+/// error when truncation would be unsafe.
+///
+/// Truncation is only sound for a torn final append, which writes at most
+/// one entry's worth of bytes. If more than `MAX_ENTRY_SIZE` bytes follow
+/// `pos`, the damage is mid-file: truncating would silently discard every
+/// committed entry after it, so this hard-errors instead of repairing.
+#[allow(clippy::future_not_send)]
+async fn truncate_or_fail(
+ storage: &FileStorage,
+ pos: u64,
+ reason: &'static str,
+) -> Result<(), JournalError> {
+ let trailing = storage.file_len().saturating_sub(pos);
+ // Sound only because this WAL appends exactly one entry per `append`
+ // + fsync, so a torn tail is at most one entry wide. A batched-append
+ // WAL could leave a torn tail many entries wide, and this
+ // `> MAX_ENTRY_SIZE` check would misclassify it as mid-file damage.
+ if trailing > MAX_ENTRY_SIZE {
+ return Err(JournalError::Io(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "mid-file WAL corruption at pos {pos}: {reason}; {trailing} bytes \
+ follow (> MAX_ENTRY_SIZE {MAX_ENTRY_SIZE}), refusing to truncate \
+ and discard committed entries"
+ ),
+ )));
+ }
+ storage.truncate(pos).await?;
+ // The repair must be crash-durable. `FileStorage::truncate` is a
+ // bare `set_len`; without this fsync a power loss right after the
+ // repair re-presents the torn tail on the next boot. Mirrors the
+ // write-then-fsync the `append` path already does.
+ storage.fsync().await?;
+ Ok(())
+}
+
+/// Best-effort unlink of the drain temp file on any error path between
+/// `File::create(wal.tmp)` and the atomic `rename`. Without this, every
+/// failed drain leaks a `wal.tmp` next to the WAL; the next drain
+/// truncates it on re-create so safety holds, but operators see the tmp
+/// files accumulate across crashed/aborted drains. `defuse` is called
+/// after a successful rename so the now-renamed file is not removed.
+///
+/// `Drop` cannot be async, so the unlink is a blocking `std::fs::remove_file`.
+/// This only runs on the drain failure path (already returning an error),
+/// so a sync syscall here is acceptable. Errors are swallowed: the file
+/// may already be gone (e.g. rename succeeded but a later step failed
+/// and we defused too late) and there is no useful recovery from a
+/// failed cleanup unlink.
+struct TmpFileGuard {
+ path: PathBuf,
+ armed: bool,
+}
+
+impl TmpFileGuard {
+ const fn new(path: PathBuf) -> Self {
+ Self { path, armed: true }
+ }
+
+ fn defuse(mut self) {
+ self.armed = false;
+ }
+}
+
+impl Drop for TmpFileGuard {
+ fn drop(&mut self) {
+ if self.armed {
+ let _ = std::fs::remove_file(&self.path);
+ }
+ }
+}
+
#[allow(clippy::cast_possible_truncation)]
impl PrepareJournal {
- /// Open the WAL file, scanning forward to rebuild the in-memory index.
+ /// Open the WAL file in read-write mode, scanning forward to rebuild
+ /// the in-memory index.
///
/// `snapshot_op` is the highest op that has been durably snapshotted.
/// It must be provided so that `append()` can detect slot collisions
@@ -133,6 +244,11 @@ impl PrepareJournal {
#[allow(clippy::future_not_send)]
pub async fn open(path: &Path, snapshot_op: u64) -> Result<Self, JournalError> {
let storage = FileStorage::open(path).await?;
+ Self::scan(storage, snapshot_op).await
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn scan(storage: FileStorage, snapshot_op: u64) -> Result<Self, JournalError> {
let file_len = storage.file_len();
let mut headers: Vec<Option<PrepareHeader>> = vec![None; SLOT_COUNT];
let mut offsets: Vec<Option<u64>> = vec![None; SLOT_COUNT];
@@ -149,10 +265,13 @@ impl PrepareJournal {
aligned.as_mut_slice().copy_from_slice(&header_buf);
// `try_from_bytes`: corrupt discriminant on disk must NOT panic;
// route through the same truncate-here branch as command/size below.
+ // Copying into a 16-aligned scratch first keeps the
+ // `PrepareHeader` (u128 fields) load aligned for miri's
+ // strict-provenance / tree-borrows checks.
let Ok(header_ref) =
bytemuck::checked::try_from_bytes::<PrepareHeader>(aligned.as_slice())
else {
- storage.truncate(pos).await?;
+ truncate_or_fail(&storage, pos, "corrupt header (invalid bit pattern)").await?;
break;
};
let header: PrepareHeader = *header_ref;
@@ -162,19 +281,25 @@ impl PrepareJournal {
|| (header.size as usize) < HEADER_SIZE
|| u64::from(header.size) > MAX_ENTRY_SIZE
{
- // Corrupt or non-prepare entry, truncate here
- storage.truncate(pos).await?;
+ truncate_or_fail(&storage, pos, "corrupt or non-prepare entry").await?;
break;
}
let entry_size = u64::from(header.size);
+ // TODO(hubcio): verify `header.checksum` / `header.checksum_body`
+ // against the entry body during scan and route a mismatch
+ // through `truncate_or_fail`. Blocked on the writer side: the
+ // `PrepareHeader` projection in consensus builds prepares with
+ // `..Default::default()` so the integrity fields are always 0.
+ // Until a producer computes them, verification here would be
+ // trivially-passing noise. Without it, a body bit-flip that
+ // leaves the header valid is replayed silently as corrupt
+ // state.
+
// Check if the full entry fits
if pos + entry_size > file_len {
- // Truncated entry at tail
- // This handles the case where crash happened during write and
- // only header was written and body was not. so we truncate the file to the start of the entry.
- storage.truncate(pos).await?;
+ truncate_or_fail(&storage, pos, "truncated entry at tail").await?;
break;
}
@@ -196,7 +321,7 @@ impl PrepareJournal {
// If there are leftover bytes less than a header, truncate them
if pos < storage.file_len() {
- storage.truncate(pos).await?;
+ truncate_or_fail(&storage, pos, "leftover bytes shorter than a header").await?;
}
Ok(Self {
@@ -205,6 +330,7 @@ impl PrepareJournal {
offsets: RefCell::new(offsets),
last_op: Cell::new(last_op),
snapshot_op: Cell::new(snapshot_op),
+ poisoned: OnceCell::new(),
})
}
@@ -244,17 +370,63 @@ impl PrepareJournal {
&self.storage
}
+ /// Stage name at which the journal was poisoned, if any. `None`
+ /// means healthy. Returned as a `&'static str` so callers in
+ /// non-IO paths (diagnostics, `Debug`) can read it without an
+ /// `io::Error` clone.
+ pub fn poison_reason(&self) -> Option<&'static str> {
+ self.poisoned.get().map(|p| p.stage)
+ }
+
+ /// Build an `io::Error` representing the current poison state. The
+ /// caller is expected to have already established that the journal
+ /// is poisoned; the embedded `source.kind()` propagates the original
+ /// kernel error (`ENOSPC`, `EIO`, ...) so the operator sees more
+ /// than a generic poison marker.
+ fn poisoned_io_error(state: &PoisonState) -> io::Error {
+ io::Error::new(
+ state.source.kind(),
+ format!("journal poisoned at {}: {}", state.stage, state.source),
+ )
+ }
+
+ /// Record the first poison cause (subsequent calls are silently
+ /// ignored - the journal is already dead) and build the descriptive
+ /// `io::Error` callers return up the stack. Consumes `source` so the
+ /// original kernel error is preserved in the cell for forensics
+ /// rather than discarded after the immediate return.
+ fn poison(&self, stage: &'static str, source: io::Error) -> io::Error {
+ let kind = source.kind();
+ let msg = format!("journal poisoned at {stage}: {source}");
+ let _ = self.poisoned.set(PoisonState { stage, source });
+ io::Error::new(kind, msg)
+ }
+
+ #[cfg(test)]
+ pub(crate) fn force_poison(&self, reason: &'static str) {
+ let _ = self.poisoned.set(PoisonState {
+ stage: reason,
+ source: io::Error::other(reason),
+ });
+ }
+
/// Async entry read for recovery.
///
/// Returns `Ok(None)` if the op is not in the index.
///
/// # Errors
/// Returns an I/O error if the read fails or the entry is malformed.
+ /// Returns an `io::ErrorKind::Other` error if the journal is
+ /// poisoned; the in-memory index and the on-disk layout disagree
+ /// and the stored offsets cannot be trusted.
#[allow(clippy::future_not_send)]
pub async fn entry_at(
&self,
header: &PrepareHeader,
) -> io::Result<Option<Message<PrepareHeader>>> {
+ if let Some(state) = self.poisoned.get() {
+ return Err(Self::poisoned_io_error(state));
+ }
let (offset, size) = {
let headers = self.headers.borrow();
let offsets = self.offsets.borrow();
@@ -329,13 +501,10 @@ impl Journal<FileStorage> for PrepareJournal {
/// future appends treat drained slots as safe to overwrite. Rewrites
/// the WAL file keeping only entries outside the drained range.
async fn drain(&self, ops: RangeInclusive<u64>) -> io::Result<Vec<Self::Entry>> {
- let end_op = *ops.end();
-
- // Advance the snapshot watermark so future appends treat
- // drained ops as safe to overwrite.
- if end_op > self.snapshot_op.get() {
- self.snapshot_op.set(end_op);
+ if let Some(state) = self.poisoned.get() {
+ return Err(Self::poisoned_io_error(state));
}
+ let end_op = *ops.end();
// Partition slots into drained and live entries.
let mut to_drain: Vec<(PrepareHeader, u64)> = Vec::new();
@@ -372,6 +541,7 @@ impl Journal<FileStorage> for PrepareJournal {
// Write live entries to a temp file.
let wal_path = self.storage.path();
let tmp_path = wal_path.with_extension("wal.tmp");
+ let tmp_guard = TmpFileGuard::new(tmp_path.clone());
{
let mut tmp = compio::fs::File::create(&tmp_path).await?;
let mut write_pos: u64 = 0;
@@ -387,16 +557,55 @@ impl Journal<FileStorage> for PrepareJournal {
}
// Atomic replace.
+ //
+ // COMMIT POINT. From here on the on-disk WAL has been swapped
+ // and the in-memory index has not yet been rebuilt for the
+ // compacted layout. Every subsequent fallible step poisons the
+ // journal before returning so the next `append` cannot write at
+ // a stale `write_offset` into the orphaned old fd, and the next
+ // `entry`/`entry_at` cannot serve offsets from the pre-drain
+ // layout that no longer exist in the new file.
compio::fs::rename(&tmp_path, wal_path).await?;
-
- // Fsync parent directory to make the rename durable.
+ // Rename has consumed `tmp_path`; nothing left to unlink.
+ tmp_guard.defuse();
+
+ // Fsync parent directory to make the rename durable. Without
+ // this the rename can be lost across a power failure and the
+ // pre-drain WAL re-presents on recovery; with the journal
+ // poisoned the caller learns the drain is not durable instead
+ // of silently proceeding.
if let Some(parent) = wal_path.parent() {
- let dir = compio::fs::File::open(parent).await?;
- dir.sync_all().await?;
+ let dir = match compio::fs::File::open(parent).await {
+ Ok(d) => d,
+ Err(e) => {
+ return Err(self.poison("drain: open parent dir for fsync", e));
+ }
+ };
+ if let Err(e) = dir.sync_all().await {
+ return Err(self.poison("drain: parent dir fsync", e));
+ }
+ }
+
+ // Reopen the file descriptor at the same path. A failure here
+ // leaves the old fd (now pointing at the orphaned pre-rename
+ // inode) live inside `FileStorage` with a stale `write_offset`;
+ // poisoning prevents a follow-up `append` from writing bytes
+ // into the orphan that disappear on the next process restart.
+ if let Err(e) = self.storage.reopen().await {
+ return Err(self.poison("drain: storage reopen after rename", e));
}
- // Reopen the file descriptor at the same path.
- self.storage.reopen().await?;
+ // Advance the snapshot watermark only AFTER the WAL rewrite is
+ // durable (tmp create -> write -> fsync -> rename -> fsync parent
+ // -> reopen). Advancing earlier would leave `snapshot_op` past
+ // entries still present on disk on any `?` failure above, letting
+ // a future `append()` pass the slot collision check at
+ // `existing.op <= snapshot_op` and silently evict a live entry
+ // from the index. The entry would survive on disk but become
+ // unreachable, stalling `RetransmitPrepares` until view change.
+ if end_op > self.snapshot_op.get() {
+ self.snapshot_op.set(end_op);
+ }
// Rebuild offsets for the compacted layout and clear drained slots.
let mut headers = self.headers.borrow_mut();
@@ -420,27 +629,59 @@ impl Journal<FileStorage> for PrepareJournal {
}
async fn append(&self, entry: Self::Entry) -> io::Result<()> {
+ if let Some(state) = self.poisoned.get() {
+ return Err(Self::poisoned_io_error(state));
+ }
let header = *entry.header();
- let offset = self.storage.file_len();
+ let slot = slot_for_op(header.op);
+
+ // Slot collision must be detected BEFORE `write_append + fsync`:
+ // a post-fsync panic would leave bytes durably on disk, and the
+ // recovery scan on the next boot would re-hit the same collision,
+ // turning a single shard fault into a cluster-wide bootloop.
+ {
+ let headers = self.headers.borrow();
+ if let Some(existing) = &headers[slot]
+ && existing.op > self.snapshot_op.get()
+ {
+ return Err(io::Error::new(
+ io::ErrorKind::InvalidData,
+ format!(
+ "journal slot collision: appending op {} would evict op {} \
+ which has not been snapshotted (snapshot_op={})",
+ header.op,
+ existing.op,
+ self.snapshot_op.get(),
+ ),
+ ));
+ }
+ }
- self.storage.write_append(entry.as_slice().to_vec()).await?;
+ let offset = self.storage.file_len();
+ // `truncate_or_fail` classifies any post-`pos` trailing region
+ // larger than `MAX_ENTRY_SIZE` as mid-file damage rather than a
+ // torn tail. That classification is only sound because this WAL
+ // appends exactly one entry per `append` + fsync. A
+ // batched-append regression that wrote more than one entry's
+ // worth of bytes per fsync would let a crash leave a torn tail
+ // many entries wide, and `truncate_or_fail` would silently
+ // discard committed entries. Pin the invariant here so any such
+ // regression trips in debug/test builds before reaching prod.
+ debug_assert!(
+ u64::from(header.size) <= MAX_ENTRY_SIZE,
+ "WAL invariant: append must write at most MAX_ENTRY_SIZE bytes; \
+ truncate_or_fail relies on this (got {} bytes)",
+ header.size
+ );
+ // Hand the message's owned aligned buffer straight to compio: avoids a
+ // per-append heap alloc + memcpy of up to MAX_ENTRY_SIZE bytes on the
+ // consensus replicate hot path. `Owned` already implements `IoBuf`, so
+ // `write_append` consumes it without copying.
+ self.storage.write_append(entry.into_owned()).await?;
self.storage.fsync().await?;
- let slot = slot_for_op(header.op);
let mut headers = self.headers.borrow_mut();
let mut offsets = self.offsets.borrow_mut();
-
- if let Some(existing) = &headers[slot] {
- assert!(
- existing.op <= self.snapshot_op.get(),
- "journal slot collision: appending op {} would evict op {} \
- which has not been snapshotted (snapshot_op={})",
- header.op,
- existing.op,
- self.snapshot_op.get(),
- );
- }
-
headers[slot] = Some(header);
offsets[slot] = Some(offset);
@@ -454,6 +695,9 @@ impl Journal<FileStorage> for PrepareJournal {
}
async fn entry(&self, header: &Self::Header) -> Option<Self::Entry> {
+ if self.poisoned.get().is_some() {
+ return None;
+ }
let (size, offset) = {
let headers = self.headers.borrow();
let offsets = self.offsets.borrow();
@@ -631,6 +875,78 @@ mod tests {
assert_eq!(entry.header().op, 1);
}
+ #[compio::test]
+ async fn truncate_or_fail_durably_repairs_torn_tail() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+
+ let good_len = {
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ journal.append(make_prepare(1, 64)).await.unwrap();
+ journal.append(make_prepare(2, 64)).await.unwrap();
+ journal.storage.fsync().await.unwrap();
+ journal.storage.file_len()
+ };
+
+ // Simulate a writer crash mid-append: junk bytes past the last
+ // good entry, shorter than MAX_ENTRY_SIZE so it reads as a torn
+ // tail rather than mid-file corruption.
+ {
+ let storage = FileStorage::open(&path).await.unwrap();
+ storage.write_append(vec![0xAB_u8; 16]).await.unwrap();
+ storage.fsync().await.unwrap();
+ }
+
+ // The recovery repair path truncates back to the last good entry.
+ {
+ let storage = FileStorage::open(&path).await.unwrap();
+ truncate_or_fail(&storage, good_len, "torn tail test")
+ .await
+ .unwrap();
+ }
+
+ // The repair is durable: a fresh open sees the file ending
+ // exactly at the last good entry, with no torn tail to re-repair.
+ assert_eq!(std::fs::metadata(&path).unwrap().len(), good_len);
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ assert_eq!(journal.last_op(), Some(2));
+ assert!(journal.header(3).is_none());
+ }
+
+ #[compio::test]
+ async fn open_rejects_mid_file_corruption() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+
+ // A corrupt header followed by more than MAX_ENTRY_SIZE of
+ // trailing bytes is mid-file damage, not a torn final append. A
+ // read-write open must hard-error instead of silently truncating
+ // and discarding every committed entry after the corruption.
+ {
+ use std::io::Write;
+ let mut file = std::fs::File::create(&path).unwrap();
+ // All-0xFF is not a valid `Command2`/`Operation` bit pattern,
+ // so `try_from_bytes` rejects the header.
+ file.write_all(&[0xFF_u8; HEADER_SIZE]).unwrap();
+ // Sparse-extend past MAX_ENTRY_SIZE so the corruption at pos 0
+ // sits far from EOF without writing 64 MiB.
+ file.set_len(MAX_ENTRY_SIZE + HEADER_SIZE as u64).unwrap();
+ file.sync_all().unwrap();
+ }
+
+ let size_before = std::fs::metadata(&path).unwrap().len();
+ let err = PrepareJournal::open(&path, 0).await;
+ assert!(
+ err.is_err(),
+ "mid-file corruption must hard-error, not truncate"
+ );
+ let size_after = std::fs::metadata(&path).unwrap().len();
+ assert_eq!(
+ size_before, size_after,
+ "a rejected mid-file scan must not truncate the WAL"
+ );
+ }
+
#[compio::test]
async fn iter_headers_from() {
let dir = tempdir().unwrap();
@@ -750,18 +1066,144 @@ mod tests {
}
#[compio::test]
- #[should_panic(expected = "journal slot collision")]
- async fn append_panics_on_evicting_unsnapshotted_entry() {
+ async fn append_errors_on_evicting_unsnapshotted_entry() {
let dir = tempdir().unwrap();
let path = dir.path().join("journal.wal");
let journal = PrepareJournal::open(&path, 0).await.unwrap();
journal.append(make_prepare(3, 32)).await.unwrap();
- // No snapshot taken, evicting op 3 must panic
+ let size_after_first = journal.storage.file_len();
+
+ // No snapshot taken: evicting op 3 must return an error WITHOUT
+ // persisting any bytes; a post-fsync panic would otherwise wedge
+ // recovery into a bootloop on the next open.
let wraparound_op = 3 + SLOT_COUNT as u64;
- journal
+ let err = journal
.append(make_prepare(wraparound_op, 32))
.await
- .unwrap();
+ .expect_err("slot collision must surface as Err, not panic");
+ assert_eq!(err.kind(), io::ErrorKind::InvalidData);
+ assert!(
+ err.to_string().contains("journal slot collision"),
+ "unexpected error message: {err}"
+ );
+ assert_eq!(
+ journal.storage.file_len(),
+ size_after_first,
+ "failed append must not persist bytes"
+ );
+
+ // Reopen: prior op still recoverable, no collision residue.
+ drop(journal);
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ assert_eq!(journal.last_op(), Some(3));
+ }
+
+ const POISON_REASON: &str = "test: simulated post-rename failure";
+
+ #[compio::test]
+ async fn poisoned_journal_rejects_append() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ journal.append(make_prepare(1, 32)).await.unwrap();
+ let size_before = journal.storage.file_len();
+
+ journal.force_poison(POISON_REASON);
+ assert_eq!(journal.poison_reason(), Some(POISON_REASON));
+
+ let err = journal
+ .append(make_prepare(2, 32))
+ .await
+ .expect_err("poisoned journal must reject append");
+ assert_eq!(err.kind(), io::ErrorKind::Other);
+ assert!(
+ err.to_string().contains(POISON_REASON),
+ "missing poison reason: {err}"
+ );
+ assert_eq!(
+ journal.storage.file_len(),
+ size_before,
+ "rejected append must not touch storage"
+ );
+ }
+
+ #[compio::test]
+ async fn poisoned_journal_rejects_drain() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ for op in 1..=3 {
+ journal.append(make_prepare(op, 32)).await.unwrap();
+ }
+ let size_before = journal.storage.file_len();
+
+ journal.force_poison(POISON_REASON);
+
+ let err = journal
+ .drain(1..=2)
+ .await
+ .expect_err("poisoned journal must reject drain");
+ assert_eq!(err.kind(), io::ErrorKind::Other);
+ assert!(err.to_string().contains(POISON_REASON));
+ assert_eq!(
+ journal.storage.file_len(),
+ size_before,
+ "rejected drain must not rewrite WAL"
+ );
+ }
+
+ #[compio::test]
+ async fn poisoned_journal_returns_none_from_entry() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ let msg = make_prepare(1, 32);
+ journal.append(msg.deep_copy()).await.unwrap();
+
+ journal.force_poison(POISON_REASON);
+
+ let h = *msg.header();
+ assert!(
+ journal.entry(&h).await.is_none(),
+ "poisoned journal must not serve cached on-disk reads"
+ );
+ }
+
+ #[compio::test]
+ async fn poisoned_journal_rejects_entry_at() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ let msg = make_prepare(1, 32);
+ journal.append(msg.deep_copy()).await.unwrap();
+
+ journal.force_poison(POISON_REASON);
+
+ let err = journal
+ .entry_at(msg.header())
+ .await
+ .expect_err("poisoned entry_at must err");
+ assert_eq!(err.kind(), io::ErrorKind::Other);
+ assert!(err.to_string().contains(POISON_REASON));
+ }
+
+ #[compio::test]
+ async fn in_memory_accessors_still_work_when_poisoned() {
+ let dir = tempdir().unwrap();
+ let path = dir.path().join("journal.wal");
+ let journal = PrepareJournal::open(&path, 0).await.unwrap();
+ journal.append(make_prepare(1, 32)).await.unwrap();
+ journal.append(make_prepare(2, 32)).await.unwrap();
+
+ journal.force_poison(POISON_REASON);
+
+ // In-memory state remains the last-known-good snapshot. Useful
+ // for diagnostics/recovery; the IO paths are what would corrupt
+ // the WAL, not these lookups.
+ assert_eq!(journal.last_op(), Some(2));
+ assert!(journal.header(1).is_some());
+ assert!(journal.header(2).is_some());
+ assert_eq!(journal.iter_headers_from(1).len(), 2);
}
}
diff --git a/core/message_bus/src/client_listener/mod.rs b/core/message_bus/src/client_listener/mod.rs
index 2d8a9ea6c..2eea83bd9 100644
--- a/core/message_bus/src/client_listener/mod.rs
+++ b/core/message_bus/src/client_listener/mod.rs
@@ -25,7 +25,7 @@
//! peer cannot block the accept loop.
//! - The `on_accepted` callback is responsible for minting a
//! `client_id`. The plain TCP and pre-upgrade WS paths route to the
-//! owning shard via `ShardFramePayload::ClientConnectionSetup` and
+//! owning shard via `shard::LifecycleFrame::ClientConnectionSetup` and
//! `ClientWsConnectionSetup` respectively. TCP-TLS, WSS, and QUIC
//! stay shard-0 terminal: their connection state is not serialisable
//! so the callback installs locally on shard 0 instead of shipping a
diff --git a/core/message_bus/src/client_listener/tcp.rs b/core/message_bus/src/client_listener/tcp.rs
index e8b3ab8ce..eee380185 100644
--- a/core/message_bus/src/client_listener/tcp.rs
+++ b/core/message_bus/src/client_listener/tcp.rs
@@ -44,7 +44,7 @@ use tracing::{debug, error, info};
pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> {
// `SO_REUSEPORT` intentionally not set: only shard 0 binds the client
// listener. The shard-0 coordinator round-robins accepts to owning
- // shards via `ShardFramePayload::ClientConnectionSetup`.
+ // shards via `shard::LifecycleFrame::ClientConnectionSetup`.
let opts = SocketOpts::new().nodelay(true);
let listener = TcpListener::bind_with_options(addr, &opts)
.await
diff --git a/core/message_bus/src/client_listener/tcp_tls.rs b/core/message_bus/src/client_listener/tcp_tls.rs
index 7e7004424..756795e8c 100644
--- a/core/message_bus/src/client_listener/tcp_tls.rs
+++ b/core/message_bus/src/client_listener/tcp_tls.rs
@@ -59,7 +59,7 @@ use tracing::{debug, error, info};
/// time; the install path re-applies on every accepted stream because
/// Linux does not propagate listener options to accepted sockets.
/// `SO_KEEPALIVE` is intentionally NOT set (see
-/// [`crate::socket_opts`]): SDK clients manage their own keepalive
+/// `socket_opts`): SDK clients manage their own keepalive
/// policy at the application layer and replica<->replica liveness is
/// observed by VSR heartbeats, so the kernel timer would only race the
/// app-level signal.
diff --git a/core/message_bus/src/client_listener/ws.rs b/core/message_bus/src/client_listener/ws.rs
index 90a6aa266..20775cbb9 100644
--- a/core/message_bus/src/client_listener/ws.rs
+++ b/core/message_bus/src/client_listener/ws.rs
@@ -20,7 +20,7 @@
//! Runs only on shard 0. The accept loop performs no protocol work
//! beyond `TcpListener::accept`: every accepted stream is handed
//! verbatim to the supplied callback, which dups the fd and ships a
-//! `ShardFramePayload::ClientWsConnectionSetup` frame to the
+//! `shard::LifecycleFrame::ClientWsConnectionSetup` frame to the
//! round-robin-selected target shard. The HTTP-Upgrade handshake runs
//! on the owning shard inside
//! [`crate::ConnectionInstaller::install_client_ws_fd`]; no
@@ -30,7 +30,7 @@
//! well-defined. The WS state machine only materialises after the
//! upgrade, on the owning shard, where it can stay non-`Send`.
//!
-//! `ShardFramePayload::ClientWsConnectionSetup` is defined in
+//! `shard::LifecycleFrame::ClientWsConnectionSetup` is defined in
//! `core/shard/src/lib.rs`; the rustdoc cannot intra-link across crates
//! without pulling `shard` in as a doc-only dep.
@@ -47,7 +47,7 @@ use tracing::{debug, error, info};
///
/// Mirrors [`crate::client_listener::tcp::bind`] in shape: `TCP_NODELAY`
/// on by default; `SO_KEEPALIVE` intentionally NOT set (see
-/// [`crate::socket_opts`]). The receiving shard re-applies socket
+/// `socket_opts`). The receiving shard re-applies socket
/// options on the dup'd fd via the existing client-install path, so
/// kernel-level options propagate end-to-end.
///
@@ -58,7 +58,7 @@ use tracing::{debug, error, info};
pub async fn bind(addr: SocketAddr) -> Result<(TcpListener, SocketAddr), IggyError> {
// `SO_REUSEPORT` intentionally not set: only shard 0 binds the WS
// listener. The shard-0 coordinator round-robins accepts to owning
- // shards via `ShardFramePayload::ClientWsConnectionSetup`.
+ // shards via `shard::LifecycleFrame::ClientWsConnectionSetup`.
let opts = SocketOpts::new().nodelay(true);
let listener = TcpListener::bind_with_options(addr, &opts)
.await
diff --git a/core/message_bus/src/connector.rs b/core/message_bus/src/connector.rs
index 3f87d90fe..d2bd79751 100644
--- a/core/message_bus/src/connector.rs
+++ b/core/message_bus/src/connector.rs
@@ -100,11 +100,12 @@ async fn connect_all(
// connection lives on shard 0; `owning_shard` covers multi-shard
// deployments where the fd was delegated to a peer shard but the
// mapping broadcast reached shard 0. Either hit means a previous
- // sweep (or the inbound listener) already installed this peer.
- // Redialing wastes a TCP round-trip; the live entry stays intact
- // (the loser of the registry insert race in `install_replica_conn`
- // gets `RejectedRegistration` back and tears DOWN ITS OWN orphan
- // tasks via `drain_rejected_registration`, never the winner's).
+ // sweep (or the inbound listener) already installed this peer,
+ // and redialing wastes a TCP round-trip. The same `owning_shard`
+ // CAS is also taken inside `install_replica_conn` before any
+ // task is spawned, so a concurrent inbound install for the same
+ // peer on a different shard loses the race there and drops the
+ // fd; the live entry stays intact on the winner.
if bus.replicas().contains(peer_id) || bus.owning_shard(peer_id).is_some() {
debug!(
replica = peer_id,
diff --git a/core/message_bus/src/error.rs b/core/message_bus/src/error.rs
index 85ae33b7d..7d5fdd7df 100644
--- a/core/message_bus/src/error.rs
+++ b/core/message_bus/src/error.rs
@@ -26,12 +26,26 @@ use thiserror::Error;
/// The client- and replica-keyed variants separate three physically
/// distinct failure modes so operators can tell apart routing bugs from
/// transient disconnects:
-/// - `*NotFound` / `*NotConnected` — the key is unknown to this shard's
+/// - `*NotFound` / `*NotConnected` - the key is unknown to this shard's
/// local registry (genuine peer state, often recoverable by reconnect).
-/// - `*RouteMissing` — the bus cannot forward to the owning shard
+/// - `*RouteMissing` - the bus cannot forward to the owning shard
/// because no forward fn was installed (bootstrap ordering bug).
-/// - `*ForwardFailed` — the forward fn rejected the frame (inter-shard
+/// - `*ForwardFailed` - the forward fn rejected the frame (inter-shard
/// queue full, shutdown, etc.).
+///
+/// Recovery asymmetry between forward variants:
+/// - `ReplicaForwardFailed` is recoverable in-protocol. The dropped frame
+/// is a consensus message (`Prepare` / `PrepareOk` / view-change /
+/// `Commit`); VSR's WAL-driven retransmit re-sends it once the inbox
+/// drains.
+/// - `ClientForwardFailed` is terminal. Client Reply payloads live above
+/// the bus and have no retransmit path. A reply lost on full inbox is
+/// gone, the client times out, and request / response semantics break.
+/// Operators must size `inbox_capacity` for the worst-case cross-shard
+/// reply burst and alert on the drop-site `tracing` logs. The
+/// `frame_drops_total` `{variant="forward_client_send"}` counter is
+/// the structured complement, scrape-able once a per-shard exporter
+/// lands.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum SendError {
@@ -41,6 +55,11 @@ pub enum SendError {
#[error("client {0}: no inter-shard forward fn installed")]
ClientRouteMissing(u128),
+ /// The owning shard's inter-shard inbox rejected the forward frame
+ /// (full or disconnected). Unlike `ReplicaForwardFailed`, no VSR
+ /// retransmit covers this drop: the client Reply payload was moved
+ /// into the frame and is now gone. Callers must treat this as a
+ /// final failure for the affected request; the client will time out.
#[error("client {0}: inter-shard forward failed")]
ClientForwardFailed(u128),
diff --git a/core/message_bus/src/installer/mod.rs b/core/message_bus/src/installer/mod.rs
index 116f887ff..6af733657 100644
--- a/core/message_bus/src/installer/mod.rs
+++ b/core/message_bus/src/installer/mod.rs
@@ -55,7 +55,7 @@ use std::rc::Rc;
use tracing::warn;
/// Operations a shard needs to perform on its local bus when the router
-/// receives an inter-shard connection-setup or mapping frame.
+/// receives an inter-shard connection-setup frame.
///
/// The production implementation is on `Rc<IggyMessageBus>`. The simulator
/// does not exercise this path; if it ever does, add a no-op impl on
@@ -87,14 +87,6 @@ pub trait ConnectionInstaller {
/// subprotocol negotiation: the caller (server-ng) gates command
/// access via the LOGIN allowlist.
fn install_client_ws_fd(&self, fd: DupedFd, meta: ClientConnMeta, on_request: RequestHandler);
-
- /// Update the replica -> owning shard mapping used by the `send_to_replica`
- /// slow path on non-owning shards.
- fn set_shard_mapping(&self, replica: u8, owning_shard: u16);
-
- /// Forget the replica -> owning shard mapping (e.g. after a connection
- /// loss, before the next allocate).
- fn remove_shard_mapping(&self, replica: u8);
}
impl ConnectionInstaller for Rc<IggyMessageBus> {
@@ -140,12 +132,4 @@ impl ConnectionInstaller for Rc<IggyMessageBus> {
});
self.track_background(handle);
}
-
- fn set_shard_mapping(&self, replica: u8, owning_shard: u16) {
- IggyMessageBus::set_shard_mapping(self, replica, owning_shard);
- }
-
- fn remove_shard_mapping(&self, replica: u8) {
- IggyMessageBus::remove_shard_mapping(self, replica);
- }
}
diff --git a/core/message_bus/src/installer/replica.rs b/core/message_bus/src/installer/replica.rs
index 3c20fa0a3..9f33d3e81 100644
--- a/core/message_bus/src/installer/replica.rs
+++ b/core/message_bus/src/installer/replica.rs
@@ -77,24 +77,47 @@ pub fn install_replica_conn<C: TransportConn>(
return;
}
+ // Atomically claim cross-shard ownership before spawning any task.
+ // `mark_replica_owned` CAS-from-`OWNER_NONE` arbitrates parallel
+ // inbound installs that target different shards: the loser shard
+ // drops the fd here without ever touching the local registry, so
+ // there is no orphan post-loop and no need for
+ // `RejectedRegistration`-style drain on this path. The matching
+ // local-`replicas().contains` check above handles same-shard
+ // duplicates; same-shard reclaim during the previous install's
+ // post-loop clear window is accepted by `try_claim`'s
+ // `actual == shard_id` fallback (the live-registry guard in
+ // `notify_connection_lost` prevents the stale post-loop from
+ // clobbering the new entry).
+ if !bus.mark_replica_owned(peer_id) {
+ debug!(
+ replica = peer_id,
+ current_owner = ?bus.owning_shard(peer_id),
+ "replica already owned by a different shard, dropping delegated fd"
+ );
+ drop(conn);
+ return;
+ }
+
let (tx, rx) = async_channel::bounded(bus.peer_queue_capacity());
let (in_tx, in_rx) =
async_channel::bounded::<Message<GenericHeader>>(bus.peer_queue_capacity());
// Writer and reader both observe abnormal close and used to fire
- // `notify_connection_lost` twice per disconnect, causing shard 0 to
- // broadcast two `ReplicaMappingClear` rounds and churn the mapping.
+ // `notify_connection_lost` twice per disconnect, double-clearing the
+ // owner-table slot and double-firing any installed test callback.
// Shared one-shot guard: whichever post-loop runs first wins.
let notified = Rc::new(Cell::new(false));
// If the registry insert below races with a concurrent install for
// the same peer id and loses, both spawned halves must skip their
// post-loop cleanup: the loser's `replicas().remove` /
// `close_peer_if_token_matches` calls would no-op against the winner's
- // generation token (so they can't evict the live entry), but
- // `notify_connection_lost` has no token guard and would still broadcast
- // a spurious mapping-clear round. `compio::runtime::JoinHandle::drop`
- // does not cancel the spawned task, so we have to tell the tasks to
- // stand down in-band.
+ // generation token (so they can't evict the live entry), and
+ // `notify_connection_lost` stands down whenever a live registry entry
+ // exists - but the loser also drives `replica_dispatch_loop`, which
+ // must never hand the winner's replica id to `on_message`.
+ // `compio::runtime::JoinHandle::drop` does not cancel the spawned
+ // task, so we have to tell the tasks to stand down in-band.
let install_aborted = Rc::new(Cell::new(false));
// Generation token published by the registry on a successful insert.
@@ -265,12 +288,23 @@ pub fn install_replica_conn<C: TransportConn>(
conn_shutdown,
) {
Ok(token) => {
+ // Owner-table slot was already claimed at the top of this
+ // function; nothing more to publish here. The matching
+ // CAS-clear runs from `IggyMessageBus::notify_connection_lost`
+ // on either of the post-loop guards firing.
install_token.set(Some(token));
}
Err(rejected) => {
- // Tell both halves to stand down: the winner's entry is live and
- // must not be touched by this losing install.
+ // Defensive: on single-threaded compio there is no `.await`
+ // between the contains-check above and this insert, so a
+ // same-shard parallel install cannot interleave and this
+ // branch is unreachable. Kept as belt-and-suspenders in case
+ // a future refactor inserts an `.await` in the synchronous
+ // setup block: we own the owner-table slot from the CAS at
+ // the top, so unstamp it before draining the orphan to
+ // avoid an owner-table entry with no live registry.
install_aborted.set(true);
+ let _ = bus.clear_replica_owned(peer_id);
warn!(replica = peer_id, "replica registry insert raced");
// `drain_rejected_registration` triggers the per-connection
// shutdown (returned with `rejected`) to wake the transport
diff --git a/core/message_bus/src/installer/tcp_tls.rs b/core/message_bus/src/installer/tcp_tls.rs
index d9468f6d0..3497aac79 100644
--- a/core/message_bus/src/installer/tcp_tls.rs
+++ b/core/message_bus/src/installer/tcp_tls.rs
@@ -41,7 +41,7 @@ use tracing::warn;
/// handshake starts, matching [`super::tcp::install_client_tcp`]'s
/// plaintext behaviour. Linux does not propagate it from the listener
/// to accepted sockets, so toggling here is required. `SO_KEEPALIVE`
-/// is intentionally NOT set; see [`crate::socket_opts`].
+/// is intentionally NOT set; see `socket_opts`.
///
/// TCP-TLS is shard-0 terminal: the rustls connection state machine
/// is non-serialisable and tied to the local task; pre-handshake the
diff --git a/core/message_bus/src/installer/wss.rs b/core/message_bus/src/installer/wss.rs
index 3ba359c7f..32db26cb4 100644
--- a/core/message_bus/src/installer/wss.rs
+++ b/core/message_bus/src/installer/wss.rs
@@ -41,7 +41,7 @@ use tracing::warn;
///
/// `TCP_NODELAY` is applied pre-handshake for symmetry with
/// [`super::tcp_tls::install_client_tcp_tls`]. `SO_KEEPALIVE` is
-/// intentionally NOT set; see [`crate::socket_opts`].
+/// intentionally NOT set; see `socket_opts`.
///
/// WSS is shard-0 terminal for the same reasons as the TCP-TLS plane;
/// see [`super::tcp_tls::install_client_tcp_tls`] for the rustls
diff --git a/core/message_bus/src/lib.rs b/core/message_bus/src/lib.rs
index a927bf8c6..c93b133ef 100644
--- a/core/message_bus/src/lib.rs
+++ b/core/message_bus/src/lib.rs
@@ -59,7 +59,7 @@
//! - fd-delegation ([`fd_transfer`]) is TCP-only. TLS / QUIC
//! connections have no dupable plaintext fd, so shard 0 terminates
//! and forwards `Frozen<MESSAGE_ALIGN>` over the existing
-//! inter-shard flume.
+//! inter-shard crossfire channel.
//! - 0-RTT stays disabled by default on any future QUIC path. Per-
//! command opt-in requires a checked-in idempotence audit.
//!
@@ -84,8 +84,7 @@ pub mod framing;
pub mod installer;
pub mod lifecycle;
pub mod replica;
-#[doc(hidden)]
-pub mod socket_opts;
+pub(crate) mod socket_opts;
pub mod transports;
pub use config::{IOV_MAX_LIMIT, MessageBusConfig, QuicTuning, WebSocketConfig};
@@ -96,7 +95,7 @@ pub use installer::conn_info::{
};
pub use lifecycle::{
BusMessage, BusReceiver, BusSender, ConnectionRegistry, DrainOutcome, FusedShutdown,
- RejectedRegistration, ReplicaRegistry, Shutdown, ShutdownToken,
+ ReplicaRegistry, Shutdown, ShutdownToken,
};
pub use transports::tls::TlsServerCredentials;
@@ -105,26 +104,154 @@ use configs::server_ng::ServerNgConfig;
use iggy_binary_protocol::consensus::MESSAGE_ALIGN;
use iggy_binary_protocol::consensus::iobuf::Frozen;
use iggy_binary_protocol::{GenericHeader, Message};
-use std::cell::RefCell;
-use std::collections::HashMap;
+use std::array;
+use std::cell::{OnceCell, RefCell};
use std::net::SocketAddr;
use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU16, Ordering};
use std::time::Duration;
-/// Callback for forwarding a consensus message to a remote shard.
+/// Maximum number of replicas a single cluster supports. Replica ids are
+/// `u8`, so the address space is 0..=255.
+pub const MAX_REPLICAS: usize = 256;
+
+/// Sentinel stored in [`ReplicaOwnerTable`] slots that have no current owner.
+///
+/// `u16::MAX` is reserved by the server bootstrap so it can never be a real
+/// shard id: `bootstrap` rejects any server whose shard count does not fit
+/// in `u16` or is `>= OWNER_NONE`, returning `ShardsCountOverflow` rather
+/// than minting a shard id that collides with this sentinel.
+pub const OWNER_NONE: u16 = u16::MAX;
+
+/// Shared atomic owner table mapping `replica_id` to `owning_shard_id`.
+///
+/// One Arc-cloned instance is allocated per server-ng process at
+/// bootstrap and shared across every shard's [`IggyMessageBus`]. The owning shard
+/// stamps its id into a slot when an inbound replica connection passes
+/// the registry-insert race; the same shard CAS-clears the slot when
+/// its connection dies (`notify_connection_lost`). Non-owning shards
+/// only ever read.
+///
+/// Lock-free by construction: every operation is a single relaxed-or-
+/// stronger atomic on a fixed `[AtomicU16; MAX_REPLICAS]`. Sized for
+/// 256 slots upfront, so install / disconnect cost is `O(1)` and no
+/// hashing or allocation happens on the hot path.
+///
+/// The table is the sole authority for cross-shard replica routing:
+/// `send_to_replica`'s slow path and [`IggyMessageBus::owning_shard`]
+/// both read it directly. No separate broadcast or reconcile loop is
+/// involved.
+pub struct ReplicaOwnerTable {
+ slots: [AtomicU16; MAX_REPLICAS],
+}
+
+impl ReplicaOwnerTable {
+ /// Build a fresh table with every slot initialised to [`OWNER_NONE`].
+ #[must_use]
+ pub fn new() -> Self {
+ Self {
+ slots: array::from_fn(|_| AtomicU16::new(OWNER_NONE)),
+ }
+ }
+
+ /// Try to claim `shard_id` as the current owner of `replica_id`.
+ ///
+ /// Returns `true` when the caller now owns the slot. Two paths
+ /// resolve to `true`:
+ /// * `compare_exchange(OWNER_NONE -> shard_id)` wins. Common case.
+ /// * The CAS fails because the slot already stores `shard_id`. A
+ /// same-shard reclaim during the post-loop clear window
+ /// ([`IggyMessageBus::notify_connection_lost`]) is benign: the
+ /// slot already names us, and the stale post-loop will stand
+ /// down once it observes a live registry entry.
+ ///
+ /// Returns `false` only when a **different** shard owns the slot.
+ /// In that case the caller must drop the inbound fd without
+ /// spawning, mirroring the existing local-`replicas().contains`
+ /// early-return at [`installer::install_replica_conn`].
+ ///
+ /// Closes the multi-shard install race where two parallel inbound
+ /// shard-delegated installs for the same `replica_id` both pass
+ /// their local per-shard registry check and both stamp the
+ /// owner-table: the second stamp would clobber the first, leaving
+ /// the loser's registry as an orphan that no cross-shard route
+ /// reaches. The CAS gates the install before any task is spawned.
+ ///
+ /// The inverse op is [`Self::clear_if_owned_by`]; both must agree on
+ /// `shard_id` to keep the slot self-consistent.
+ pub fn try_claim(&self, replica_id: u8, shard_id: u16) -> bool {
+ match self.slots[usize::from(replica_id)].compare_exchange(
+ OWNER_NONE,
+ shard_id,
+ Ordering::AcqRel,
+ Ordering::Acquire,
+ ) {
+ Ok(_) => true,
+ Err(actual) => actual == shard_id,
+ }
+ }
+
+ /// Clear the slot iff its current owner is `shard_id`. Returns
+ /// `true` when the CAS succeeded.
+ ///
+ /// The CAS prevents a stale post-loop on shard A from clobbering a
+ /// slot that has since been re-registered on shard B.
+ pub fn clear_if_owned_by(&self, replica_id: u8, shard_id: u16) -> bool {
+ self.slots[usize::from(replica_id)]
+ .compare_exchange(shard_id, OWNER_NONE, Ordering::AcqRel, Ordering::Acquire)
+ .is_ok()
+ }
+
+ /// Read the current owner of `replica_id`. Returns `None` when the
+ /// slot is [`OWNER_NONE`].
+ #[must_use]
+ pub fn owner(&self, replica_id: u8) -> Option<u16> {
+ match self.slots[usize::from(replica_id)].load(Ordering::Acquire) {
+ OWNER_NONE => None,
+ owner => Some(owner),
+ }
+ }
+}
+
+impl Default for ReplicaOwnerTable {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+/// Callback for forwarding a consensus replica message to the owning shard.
///
/// Provided by the shard layer at bus construction and installed via
-/// [`IggyMessageBus::set_replica_forward_fn`] / [`IggyMessageBus::set_client_forward_fn`].
-/// Arguments: `(target_shard_id, message)`. Returns `Ok(())` on successful
-/// enqueue into the inter-shard channel, or [`SendError::RoutingFailed`] on
-/// `try_send` failure.
+/// [`IggyMessageBus::set_replica_forward_fn`]. Arguments:
+/// `(replica_id, owning_shard_id, message)`. Returns `Ok(())` on successful
+/// enqueue into the inter-shard channel.
+///
+/// `replica_id` is carried explicitly so the receiver can rebuild
+/// `ShardFrame::Lifecycle`'s `ForwardReplicaSend { replica_id, msg }`
+/// payload; the bus' slow path knows the id but the closure body, which lives
+/// in the shard layer, would otherwise have to re-derive it from the message.
///
/// Asymmetry vs the `Rc<dyn Fn>` siblings (`AcceptedReplicaFn`,
/// `AcceptedClientFn`, `ConnectionLostFn`): the forward fn has a single
/// owner (the bus) and is installed once at bootstrap; `Box` suffices.
/// The `Rc` siblings are shared with accept loops and connection tasks
/// that outlive the caller and need independent ownership.
-pub type ShardForwardFn = Box<dyn Fn(u16, Frozen<MESSAGE_ALIGN>) -> Result<(), SendError>>;
+pub type ReplicaForwardFn = Box<dyn Fn(u8, u16, Frozen<MESSAGE_ALIGN>) -> Result<(), SendError>>;
+
+/// Callback for forwarding a client-bound message to the owning shard.
+///
+/// Provided by the shard layer at bus construction and installed via
+/// [`IggyMessageBus::set_client_forward_fn`]. Arguments:
+/// `(client_id, owning_shard_id, message)`. Returns `Ok(())` on successful
+/// enqueue into the inter-shard channel.
+///
+/// `owning_shard_id` is the value returned by [`client_id_owning_shard`];
+/// the slow path passes it pre-resolved so the closure does not redo the
+/// shift. `client_id` is carried so the receiver can rebuild the
+/// `ShardFrame::Lifecycle`'s `ForwardClientSend { client_id, msg }`
+/// payload.
+pub type ClientForwardFn = Box<dyn Fn(u128, u16, Frozen<MESSAGE_ALIGN>) -> Result<(), SendError>>;
/// Callback invoked on every successful replica handshake.
///
@@ -161,8 +288,14 @@ pub struct AcceptedQuicConn {
impl AcceptedQuicConn {
/// Bundle a freshly-accepted QUIC connection and its first
/// bidirectional stream pair.
+ ///
+ /// `pub(crate)` by design: the constructor's signature mentions
+ /// `compio_quic::Connection`, `SendStream`, and `RecvStream`, all
+ /// kept off the bus's public `SemVer` surface for the same reason
+ /// [`Self::into_parts`] is crate-private. The QUIC listener in
+ /// [`crate::client_listener::quic`] is the only mint site.
#[must_use]
- pub const fn new(
+ pub(crate) const fn new(
connection: compio_quic::Connection,
streams: (compio_quic::SendStream, compio_quic::RecvStream),
) -> Self {
@@ -220,7 +353,7 @@ pub type AcceptedQuicClientFn = std::rc::Rc<dyn Fn(AcceptedQuicConn)>;
/// Fires after shard 0's WS listener accepts a raw TCP socket. The
/// HTTP-Upgrade handshake has NOT run yet; the callback hands the
/// raw stream off to the owning shard via inter-shard fd-shipping
-/// (`ShardFramePayload::ClientWsConnectionSetup`). The owning shard
+/// (`shard::LifecycleFrame::ClientWsConnectionSetup`). The owning shard
/// runs the upgrade locally; no subprotocol is negotiated. The
/// shipped fd is plain TCP at ship-time, so fd-delegation (which
/// requires a dupable plaintext fd) stays well-defined.
@@ -291,6 +424,34 @@ pub type ConnectionLostFn = std::rc::Rc<dyn Fn(u8)>;
/// on first poll. Future transports (QUIC, TLS) that would introduce real
/// yields must advertise that change in their own doc; consensus code
/// relies on the no-yield property to reason about reentrancy.
+///
+/// # Recovery asymmetry between `send_to_replica` and `send_to_client`
+///
+/// `send_to_replica` carries VSR-covered frames (`Prepare` / `PrepareOk`
+/// / view-change / `Commit`). A `ReplicaForwardFailed` on full
+/// inter-shard inbox is recovered by VSR retransmit and is
+/// informational.
+/// `send_to_client` carries the final Reply payload to the originating
+/// client; it has no in-protocol retransmit. A `ClientForwardFailed` is
+/// terminal: the client never receives the reply and times out.
+/// Operators must size `inbox_capacity` for the worst-case cross-shard
+/// reply burst. Each drop is logged at the drop site via `tracing`;
+/// alert on those logs today. The `frame_drops_total`
+/// `{variant="forward_client_send"}` counter is the structured
+/// complement and becomes scrape-able once a per-shard exporter lands.
+/// The symmetric `{variant="forward_replica_send"}` counter stays
+/// informational only.
+///
+/// # Setter install contract
+///
+/// The three setters below do NOT share one install rule:
+/// - [`Self::set_connection_lost_fn`] is re-installable (`RefCell`
+/// slot); it is a test observation hook some setups swap mid-test.
+/// - [`Self::set_replica_forward_fn`] and [`Self::set_client_forward_fn`]
+/// are one-shot (`OnceCell` slot) and panic on a second install;
+/// they are bootstrap-only wiring invariants.
+///
+/// A bus impl must preserve this divergence - see each method.
pub trait MessageBus {
fn send_to_client(
&self,
@@ -305,15 +466,50 @@ pub trait MessageBus {
) -> impl Future<Output = Result<(), SendError>>;
/// Install a notifier the bus will invoke when a delegated replica
- /// connection dies abnormally. Used by shard-0 bootstrap to push a
- /// `ShardFramePayload::ConnectionLost` into shard 0's inbox so the
- /// coordinator can forget the replica's mapping.
- ///
- /// Default impl is a no-op for buses that do not delegate real
- /// connections (simulator, test doubles); the production
- /// `IggyMessageBus` overrides it to wire into its internal
- /// `connection_lost_fn` hook.
- fn set_connection_lost_fn(&self, _f: ConnectionLostFn) {}
+ /// connection dies abnormally. Production wiring leaves this unset
+ /// because the shard-shared [`ReplicaOwnerTable`] is CAS-cleared
+ /// inside `IggyMessageBus::notify_connection_lost` before the
+ /// callback runs, so cross-shard routing recovers without any
+ /// follow-up message. Test code installs a callback to assert the
+ /// notifier path fires exactly once per disconnect.
+ ///
+ /// No default: every bus must explicitly opt in or stub. A silent
+ /// no-op default previously masked a wiring bug where shard 0 never
+ /// learned a delegated replica died.
+ ///
+ /// Re-installing is allowed by design - the backing slot is a
+ /// `RefCell`, so a second call overwrites the first. This is the
+ /// deliberate exception to the install rules of
+ /// [`Self::set_replica_forward_fn`] / [`Self::set_client_forward_fn`]:
+ /// the notifier is a test observation hook some setups swap
+ /// mid-test, not a one-shot bootstrap invariant.
+ fn set_connection_lost_fn(&self, f: ConnectionLostFn);
+
+ /// Install the replica-plane inter-shard forward closure.
+ ///
+ /// No default: every bus must explicitly opt in or stub. Mirrors the
+ /// reason `set_connection_lost_fn` is required - a silent no-op
+ /// default could mask a wiring bug where cross-shard replica sends
+ /// drop on the floor without anyone noticing.
+ ///
+ /// # Panics
+ ///
+ /// Panics on a second install on the same bus. Unlike
+ /// [`Self::set_connection_lost_fn`], the forward closure is a
+ /// one-shot bootstrap invariant backed by a `OnceCell`, so a
+ /// double-install is a wiring bug rather than a supported re-bind.
+ fn set_replica_forward_fn(&self, f: ReplicaForwardFn);
+
+ /// Install the client-plane inter-shard forward closure.
+ ///
+ /// No default: every bus must explicitly opt in or stub. Same
+ /// rationale as [`Self::set_replica_forward_fn`].
+ ///
+ /// # Panics
+ ///
+ /// Panics on a second install on the same bus, same one-shot
+ /// `OnceCell` invariant as [`Self::set_replica_forward_fn`].
+ fn set_client_forward_fn(&self, f: ClientForwardFn);
}
/// Production message bus backed by real TCP connections.
@@ -330,7 +526,10 @@ pub trait MessageBus {
/// - `send_to_*` clones the per-peer `Sender` out of the registry, calls
/// `try_send` on it, and returns. No `.await` happens in the body. Drops
/// on `Full` (returned as [`SendError::Backpressure`]) are recovered by
-/// VSR retransmission.
+/// VSR retransmission for replica-bound consensus traffic. Client Reply
+/// drops returned as [`SendError::ClientForwardFailed`] have no
+/// retransmit and are terminal; see the trait-level "Recovery
+/// asymmetry" section on [`MessageBus`].
/// - The per-connection writer task batches up to `config.max_batch` messages into
/// a single `writev` syscall on the plaintext TCP plane (see
/// [`transports::tcp`]).
@@ -345,25 +544,30 @@ pub struct IggyMessageBus {
replicas: ReplicaRegistry,
background_tasks: RefCell<Vec<JoinHandle<()>>>,
config: MessageBusConfig,
- /// For each replica, the shard that owns the TCP connection from this
- /// shard's perspective. Present on ALL shards (owning and non-owning).
- /// On the owning shard, `shard_mapping[replica] == self.shard_id`.
- shard_mapping: RefCell<HashMap<u8, u16>>,
/// Forwards a replica-bound message to the shard that owns the replica's
- /// TCP connection. `None` on single-shard setups and tests.
- replica_forward_fn: RefCell<Option<ShardForwardFn>>,
+ /// TCP connection. Empty on single-shard setups and tests. Installed
+ /// once at bootstrap via [`Self::set_replica_forward_fn`]; the slow
+ /// path reads through [`OnceCell::get`] without a runtime borrow.
+ replica_forward_fn: OnceCell<ReplicaForwardFn>,
/// Forwards a client-bound message to the shard that owns the client's
/// TCP connection (the owning shard is encoded in the top 16 bits of
- /// the client id). `None` on single-shard setups and tests.
- client_forward_fn: RefCell<Option<ShardForwardFn>>,
- /// Invoked by a delegated replica connection's writer / reader tasks
- /// when they exit abnormally. `None` when running without a shard-0
- /// coordinator (single-shard deployments and tests).
+ /// the client id). Empty on single-shard setups and tests. Installed
+ /// once at bootstrap via [`Self::set_client_forward_fn`]; the slow
+ /// path reads through [`OnceCell::get`] without a runtime borrow.
+ client_forward_fn: OnceCell<ClientForwardFn>,
+ /// Optional disconnect notifier invoked by
+ /// `Self::notify_connection_lost` after it CAS-clears the owner
+ /// table. Production leaves this unset; tests install a callback to
+ /// observe the path.
connection_lost_fn: RefCell<Option<ConnectionLostFn>>,
/// Per-connection metadata exposed to the caller via
/// [`Self::client_meta`]. Populated by the install path on
/// successful registry insert; removed on connection teardown.
client_meta: RefCell<ahash::AHashMap<u128, Rc<ClientConnMeta>>>,
+ /// Shard-shared atomic owner table. Stamped by the install path
+ /// after a successful registry insert; CAS-cleared by
+ /// `Self::notify_connection_lost`. See [`ReplicaOwnerTable`].
+ owner_table: Arc<ReplicaOwnerTable>,
}
impl IggyMessageBus {
@@ -408,6 +612,20 @@ impl IggyMessageBus {
Self::with_tunables(shard_id, MessageBusConfig::from(cfg))
}
+ /// Production constructor for multi-shard server-ng: same as
+ /// [`Self::with_config`] but takes a pre-allocated
+ /// [`ReplicaOwnerTable`] Arc. Bootstrap allocates one table per
+ /// server-ng process and clones the Arc into every shard so all
+ /// buses see the same atomic slots.
+ #[must_use]
+ pub fn with_config_and_owner_table(
+ shard_id: u16,
+ cfg: &ServerNgConfig,
+ owner_table: Arc<ReplicaOwnerTable>,
+ ) -> Self {
+ Self::with_tunables_and_owner_table(shard_id, MessageBusConfig::from(cfg), owner_table)
+ }
+
/// Construct a bus from already-derived runtime tunables.
///
/// Used by the public constructors above and by tests that need to
@@ -420,6 +638,24 @@ impl IggyMessageBus {
/// `config.max_batch > IOV_MAX_LIMIT`.
#[must_use]
pub fn with_tunables(shard_id: u16, config: MessageBusConfig) -> Self {
+ Self::with_tunables_and_owner_table(shard_id, config, Arc::new(ReplicaOwnerTable::new()))
+ }
+
+ /// Construct a bus with explicit runtime tunables and a pre-allocated
+ /// owner table. Server-ng bootstrap uses this so every shard's bus
+ /// shares the same atomic slots; tests use [`Self::with_tunables`]
+ /// which allocates a fresh table per bus.
+ ///
+ /// # Panics
+ ///
+ /// Panics if `config.max_batch == 0` or
+ /// `config.max_batch > IOV_MAX_LIMIT`.
+ #[must_use]
+ pub fn with_tunables_and_owner_table(
+ shard_id: u16,
+ config: MessageBusConfig,
+ owner_table: Arc<ReplicaOwnerTable>,
+ ) -> Self {
assert!(
config.max_batch > 0 && config.max_batch <= IOV_MAX_LIMIT,
"MessageBusConfig::max_batch must be in 1..={IOV_MAX_LIMIT} (writev IOV_MAX/2 cap); got {}",
@@ -434,14 +670,46 @@ impl IggyMessageBus {
replicas: ReplicaRegistry::new(),
background_tasks: RefCell::new(Vec::new()),
config,
- shard_mapping: RefCell::new(HashMap::new()),
- replica_forward_fn: RefCell::new(None),
- client_forward_fn: RefCell::new(None),
+ replica_forward_fn: OnceCell::new(),
+ client_forward_fn: OnceCell::new(),
connection_lost_fn: RefCell::new(None),
client_meta: RefCell::new(ahash::AHashMap::new()),
+ owner_table,
}
}
+ /// Clone of the shard-shared owner table Arc.
+ #[must_use]
+ pub fn owner_table(&self) -> Arc<ReplicaOwnerTable> {
+ Arc::clone(&self.owner_table)
+ }
+
+ /// Try to claim this bus' shard id as owner of `replica_id`.
+ ///
+ /// Returns `true` when the claim succeeded (the slot now points at
+ /// `self.shard_id`, either because it was `OWNER_NONE` or because
+ /// it already stored `self.shard_id`). Returns `false` when a
+ /// different shard owns the slot; the caller must drop the
+ /// inbound connection.
+ ///
+ /// Called by [`installer::install_replica_conn`] **before**
+ /// spawning tasks or inserting into the local registry. See
+ /// [`ReplicaOwnerTable::try_claim`] for the multi-shard race this
+ /// closes.
+ #[must_use]
+ pub fn mark_replica_owned(&self, replica_id: u8) -> bool {
+ self.owner_table.try_claim(replica_id, self.shard_id)
+ }
+
+ /// CAS-clear the owner-table slot for `replica_id` if its current
+ /// owner is this bus' shard id. The CAS prevents a stale post-loop
+ /// from clobbering a slot that has since been re-registered on a
+ /// different shard.
+ pub fn clear_replica_owned(&self, replica_id: u8) -> bool {
+ self.owner_table
+ .clear_if_owned_by(replica_id, self.shard_id)
+ }
+
/// Look up the per-connection metadata recorded for `client_id`.
///
/// Returns `None` if the client never connected on this bus or if
@@ -472,6 +740,31 @@ impl IggyMessageBus {
/// [`Self::set_connection_lost_fn`] (which takes a `borrow_mut`)
/// without tripping the runtime borrow check.
pub(crate) fn notify_connection_lost(&self, replica_id: u8) {
+ // Stand down if a live registry entry exists for this replica. A
+ // reconnect that round-robined back to this same shard installs a
+ // fresh entry before this stale post-loop runs; the owner-table
+ // CAS in `clear_replica_owned` only guards *different*-shard
+ // re-registration, so a same-shard reinstall leaves the slot
+ // == self.shard_id and the clear below would clobber the live
+ // connection's slot, stranding cross-shard `send_to_replica`
+ // forever. Every caller removes its own registry entry before
+ // invoking this, so a live entry here is always a newer install
+ // that now owns the slot. No token needed: the bus is
+ // single-threaded (compio), so this check, the clear, and a
+ // competing `install_replica_conn` cannot interleave.
+ if self.replicas.contains(replica_id) {
+ return;
+ }
+ // CAS-clear first so any reader racing the user callback already
+ // sees the slot vacated. The CAS no-ops if a different shard
+ // already owns the slot (post-loop arriving after a
+ // re-installation on another shard).
+ self.clear_replica_owned(replica_id);
+ tracing::warn!(
+ shard_id = self.shard_id,
+ replica_id,
+ "replica connection lost"
+ );
let cb = self
.connection_lost_fn
.borrow()
@@ -493,11 +786,14 @@ impl IggyMessageBus {
///
/// # Panics
///
- /// Panics on re-entrant `RefCell::borrow_mut` if called from inside
- /// an in-flight forward-closure invocation. All production call sites
- /// are bootstrap only (single-threaded compio, no re-entry).
- pub fn set_replica_forward_fn(&self, f: ShardForwardFn) {
- *self.replica_forward_fn.borrow_mut() = Some(f);
+ /// Panics if called more than once on the same bus. The bootstrap is
+ /// the sole caller and runs exactly once per bus; a second install
+ /// signals a wiring bug we want to surface loudly.
+ pub fn set_replica_forward_fn(&self, f: ReplicaForwardFn) {
+ self.replica_forward_fn
+ .set(f)
+ .ok()
+ .expect("replica_forward_fn installed twice (bootstrap invariant)");
}
/// Install the client-plane inter-shard forward closure.
@@ -510,41 +806,23 @@ impl IggyMessageBus {
///
/// # Panics
///
- /// Panics on re-entrant `RefCell::borrow_mut` if called from inside
- /// an in-flight forward-closure invocation.
- pub fn set_client_forward_fn(&self, f: ShardForwardFn) {
- *self.client_forward_fn.borrow_mut() = Some(f);
+ /// Panics if called more than once on the same bus. Same rationale as
+ /// [`Self::set_replica_forward_fn`].
+ pub fn set_client_forward_fn(&self, f: ClientForwardFn) {
+ self.client_forward_fn
+ .set(f)
+ .ok()
+ .expect("client_forward_fn installed twice (bootstrap invariant)");
}
- /// Update the shard mapping for a replica.
- ///
- /// Called when the allocation strategy assigns or reassigns connections.
- ///
- /// # Panics
- ///
- /// Panics on re-entrant `RefCell::borrow_mut` if called while
- /// [`IggyMessageBus::owning_shard`] (or any other read-side of
- /// `shard_mapping`) holds an outstanding read borrow on the same
- /// bus instance.
- pub fn set_shard_mapping(&self, replica: u8, owning_shard: u16) {
- self.shard_mapping
- .borrow_mut()
- .insert(replica, owning_shard);
- }
-
- /// Remove all shard mappings for a replica (on deallocation).
- ///
- /// # Panics
+ /// Get the owning shard for a replica, if any.
///
- /// Same re-entrant-`borrow_mut` constraint as
- /// [`IggyMessageBus::set_shard_mapping`].
- pub fn remove_shard_mapping(&self, replica: u8) {
- self.shard_mapping.borrow_mut().remove(&replica);
- }
-
- /// Get the owning shard for a replica, if mapped.
+ /// Reads the shard-shared [`ReplicaOwnerTable`], which is stamped
+ /// synchronously by the owning shard's installer and CAS-cleared on
+ /// disconnect inside `Self::notify_connection_lost`.
+ #[must_use]
pub fn owning_shard(&self, replica: u8) -> Option<u16> {
- self.shard_mapping.borrow().get(&replica).copied()
+ self.owner_table.owner(replica)
}
#[must_use]
@@ -713,6 +991,14 @@ impl<T: MessageBus + ?Sized> MessageBus for std::rc::Rc<T> {
fn set_connection_lost_fn(&self, f: ConnectionLostFn) {
(**self).set_connection_lost_fn(f);
}
+
+ fn set_replica_forward_fn(&self, f: ReplicaForwardFn) {
+ (**self).set_replica_forward_fn(f);
+ }
+
+ fn set_client_forward_fn(&self, f: ClientForwardFn) {
+ (**self).set_client_forward_fn(f);
+ }
}
#[allow(clippy::future_not_send)]
@@ -737,11 +1023,12 @@ impl MessageBus for IggyMessageBus {
Err(_msg) => Err(SendError::ClientNotFound(client_id)),
};
}
- let forward = self.client_forward_fn.borrow();
- let forward = forward
- .as_ref()
+ let forward = self
+ .client_forward_fn
+ .get()
.ok_or(SendError::ClientRouteMissing(client_id))?;
- forward(owning_shard, message).map_err(|_| SendError::ClientForwardFailed(client_id))
+ forward(client_id, owning_shard, message)
+ .map_err(|_| SendError::ClientForwardFailed(client_id))
}
async fn send_to_replica(
@@ -760,22 +1047,43 @@ impl MessageBus for IggyMessageBus {
Err(message) => message,
};
// Slow path: route via the inter-shard channel to the owning shard.
+ // The shard-shared `owner_table` is the authoritative routing
+ // view: `mark_replica_owned` stamps it on install and
+ // `clear_replica_owned` CAS-clears it on disconnect.
let owning_shard = self
- .shard_mapping
- .borrow()
- .get(&replica)
- .copied()
+ .owner_table
+ .owner(replica)
.ok_or(SendError::ReplicaNotConnected(replica))?;
- let forward = self.replica_forward_fn.borrow();
- let forward = forward
- .as_ref()
+ // The fast path already failed (no live registry slot), so a
+ // self-owner here means the disconnect is mid-flight:
+ // `close_peer_if_token_matches` took the slot but
+ // `clear_replica_owned` has not yet run. Forwarding now would
+ // `try_send` a `ForwardReplicaSend` onto our own inbox and the
+ // pump would re-enter this path - a self-inbox ping-pong on the
+ // consensus hot path. Treat as not connected, mirroring the
+ // `send_to_client` self-owner guard.
+ if owning_shard == self.shard_id {
+ return Err(SendError::ReplicaNotConnected(replica));
+ }
+ let forward = self
+ .replica_forward_fn
+ .get()
.ok_or(SendError::ReplicaRouteMissing(replica))?;
- forward(owning_shard, message).map_err(|_| SendError::ReplicaForwardFailed(replica))
+ forward(replica, owning_shard, message)
+ .map_err(|_| SendError::ReplicaForwardFailed(replica))
}
fn set_connection_lost_fn(&self, f: ConnectionLostFn) {
Self::set_connection_lost_fn(self, f);
}
+
+ fn set_replica_forward_fn(&self, f: ReplicaForwardFn) {
+ Self::set_replica_forward_fn(self, f);
+ }
+
+ fn set_client_forward_fn(&self, f: ClientForwardFn) {
+ Self::set_client_forward_fn(self, f);
+ }
}
/// Extract the owning shard from a client id.
@@ -823,16 +1131,20 @@ mod tests {
let bus = IggyMessageBus::new(5);
let captured: std::rc::Rc<RefCell<Option<u16>>> = std::rc::Rc::new(RefCell::new(None));
let captured_clone = captured.clone();
- bus.set_client_forward_fn(Box::new(move |target, _msg| {
+ let client_id = (7u128 << 112) | 0x2a;
+ let captured_id: std::rc::Rc<RefCell<Option<u128>>> = std::rc::Rc::new(RefCell::new(None));
+ let captured_id_clone = captured_id.clone();
+ bus.set_client_forward_fn(Box::new(move |id, target, _msg| {
+ *captured_id_clone.borrow_mut() = Some(id);
*captured_clone.borrow_mut() = Some(target);
Ok(())
}));
- let client_id = (7u128 << 112) | 0x2a;
bus.send_to_client(client_id, dummy_message().into_frozen())
.await
.expect("forward_fn should accept");
assert_eq!(*captured.borrow(), Some(7));
+ assert_eq!(*captured_id.borrow(), Some(client_id));
}
#[compio::test]
@@ -863,21 +1175,25 @@ mod tests {
#[compio::test]
#[allow(clippy::future_not_send)]
- async fn send_to_replica_slow_path_uses_shard_mapping_and_forwards() {
+ async fn send_to_replica_slow_path_uses_owner_table_and_forwards() {
let bus = IggyMessageBus::new(0);
let captured: std::rc::Rc<RefCell<Option<u16>>> = std::rc::Rc::new(RefCell::new(None));
let captured_clone = captured.clone();
- bus.set_replica_forward_fn(Box::new(move |target, _msg| {
+ let captured_id: std::rc::Rc<RefCell<Option<u8>>> = std::rc::Rc::new(RefCell::new(None));
+ let captured_id_clone = captured_id.clone();
+ bus.set_replica_forward_fn(Box::new(move |id, target, _msg| {
+ *captured_id_clone.borrow_mut() = Some(id);
*captured_clone.borrow_mut() = Some(target);
Ok(())
}));
- bus.set_shard_mapping(5, 3);
+ assert!(bus.owner_table().try_claim(5, 3));
bus.send_to_replica(5, dummy_message().into_frozen())
.await
.expect("forward ok");
assert_eq!(*captured.borrow(), Some(3));
+ assert_eq!(*captured_id.borrow(), Some(5));
}
#[compio::test]
@@ -891,6 +1207,24 @@ mod tests {
assert!(matches!(err, SendError::ReplicaNotConnected(9)));
}
+ /// Disconnect-race window: the registry slot is already gone (fast
+ /// path misses) but `owner_table` still points at this shard because
+ /// `clear_replica_owned` has not run yet. The slow path must treat a
+ /// self-owner as not connected rather than forwarding the frame onto
+ /// its own inbox - mirrors the `send_to_client` self-owner guard.
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn send_to_replica_slow_path_self_owner_returns_not_connected() {
+ let bus = IggyMessageBus::new(3);
+ assert!(bus.owner_table().try_claim(5, 3));
+
+ let err = bus
+ .send_to_replica(5, dummy_message().into_frozen())
+ .await
+ .unwrap_err();
+ assert!(matches!(err, SendError::ReplicaNotConnected(5)));
+ }
+
#[compio::test]
#[allow(clippy::future_not_send)]
async fn client_id_owning_shard_extracts_top_16_bits() {
@@ -990,6 +1324,43 @@ mod tests {
assert_eq!(counter.get(), 11);
}
+ /// Same-shard reconnect race: a stale post-loop from a dead connection
+ /// must not CAS-clear the owner-table slot that a live reinstall
+ /// (round-robined back to this shard) just stamped. The slot is only
+ /// released once no live registry entry remains.
+ #[compio::test]
+ #[allow(clippy::future_not_send)]
+ async fn notify_connection_lost_stands_down_when_replica_reregistered() {
+ let bus = IggyMessageBus::new(3);
+
+ // Live reinstall: registry entry present + owner table stamped.
+ let (tx, _rx) = async_channel::bounded(8);
+ let writer = compio::runtime::spawn(async {});
+ let reader = compio::runtime::spawn(async {});
+ let (conn_shutdown, _conn_token) = Shutdown::new();
+ bus.replicas()
+ .insert(7, tx, writer, reader, conn_shutdown)
+ .expect("insert ok");
+ assert!(bus.mark_replica_owned(7), "owner-table claim must succeed");
+
+ // Stale predecessor post-loop fires while the reinstall is live.
+ bus.notify_connection_lost(7);
+ assert_eq!(
+ bus.owning_shard(7),
+ Some(3),
+ "stale post-loop cleared a re-registered replica's owner slot",
+ );
+
+ // Genuine disconnect: no live entry, the slot is released.
+ bus.replicas().remove(7);
+ bus.notify_connection_lost(7);
+ assert_eq!(
+ bus.owning_shard(7),
+ None,
+ "owner slot must clear once no live connection remains",
+ );
+ }
+
#[compio::test]
#[allow(clippy::future_not_send)]
async fn shutdown_loop_drains_tasks_added_during_shutdown() {
@@ -1017,4 +1388,43 @@ mod tests {
"shutdown must leave background_tasks empty",
);
}
+
+ #[test]
+ fn try_claim_wins_from_none() {
+ let table = ReplicaOwnerTable::new();
+ assert!(table.try_claim(42, 3), "claim from OWNER_NONE must win");
+ assert_eq!(table.owner(42), Some(3));
+ }
+
+ #[test]
+ fn try_claim_loses_to_different_shard() {
+ let table = ReplicaOwnerTable::new();
+ assert!(table.try_claim(42, 5));
+ assert!(
+ !table.try_claim(42, 7),
+ "claim from different shard must lose"
+ );
+ assert_eq!(
+ table.owner(42),
+ Some(5),
+ "losing claim must not mutate the slot",
+ );
+ }
+
+ #[test]
+ fn try_claim_same_shard_returns_true() {
+ // Models the same-shard reclaim window: previous install on
+ // this shard stamped the slot, post-loop's
+ // `clear_replica_owned` has not run yet, a new install for the
+ // same replica id arrives. The CAS-from-NONE fails but
+ // `try_claim` treats `actual == shard_id` as success so the
+ // new install proceeds.
+ let table = ReplicaOwnerTable::new();
+ assert!(table.try_claim(42, 5));
+ assert!(
+ table.try_claim(42, 5),
+ "same-shard reclaim must be accepted",
+ );
+ assert_eq!(table.owner(42), Some(5));
+ }
}
diff --git a/core/message_bus/src/transports/mod.rs b/core/message_bus/src/transports/mod.rs
index fab08207c..a141ba04b 100644
--- a/core/message_bus/src/transports/mod.rs
+++ b/core/message_bus/src/transports/mod.rs
@@ -70,7 +70,7 @@
//! - fd-delegation (`F_DUPFD_CLOEXEC`) stays TCP-only and lives outside
//! this trait (see [`crate::fd_transfer`]). Other transports terminate
//! on shard 0 and forward `Frozen<MESSAGE_ALIGN>` over the inter-shard
-//! flume via [`crate::ShardForwardFn`].
+//! crossfire channel via [`crate::ReplicaForwardFn`] / [`crate::ClientForwardFn`].
//!
//! Per-transport impls live in
//! `transports/{tcp,tcp_tls,ws,wss,quic}.rs`.
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 3e5487b48..d4318889c 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -302,12 +302,45 @@ impl std::fmt::Display for RegisterSubmitError {
impl std::error::Error for RegisterSubmitError {}
+/// Log + surface `None` when a metadata callback runs on a peer shard
+/// (whose `consensus` / `journal` slot is `None`). The `Plane` trait
+/// callbacks are addressed to the shard 0 owner; if the routing layer
+/// ever dispatches one to a peer the only honest answer is "drop and
+/// alert" - panicking would crash the bus and mask the routing bug.
+fn require_shard_zero<'a, T>(
+ slot: Option<&'a T>,
+ callback: &'static str,
+ field: &'static str,
+) -> Option<&'a T> {
+ if slot.is_none() {
+ error!(
+ target: "iggy.metadata.diag",
+ plane = "metadata",
+ callback,
+ field,
+ "metadata callback fired on a peer shard (field is None); routing layer \
+ must direct metadata traffic to shard 0 - dropping the message"
+ );
+ }
+ slot
+}
+
pub struct IggyMetadata<C, J, S, M> {
- /// Some on shard0, None on other shards
+ /// `Some` on shard 0, `None` on other shards. Server-ng bootstrap
+ /// holds the invariant: only shard 0 owns the metadata consensus
+ /// replica; every other shard reconstructs `mux_stm` from the
+ /// `MetadataHandoff::Waiter` factory bundle broadcast by shard 0
+ /// (no consensus replica, no journal access).
pub consensus: Option<C>,
- /// Some on shard0, None on other shards
+ /// `Some` on shard 0, `None` on other shards. Shard 0 owns the WAL
+ /// writer (via `PrepareJournal::open`); non-owning shards never open
+ /// the WAL at all. They receive a `MetadataHandoff::Waiter` factory
+ /// bundle from shard 0 over the bootstrap broadcast channel and
+ /// reconstruct `mux_stm` from the in-memory snapshot it carries (see
+ /// `server-ng/src/bootstrap.rs` `await_metadata_bundle` /
+ /// `broadcast_metadata_bundle`).
pub journal: Option<J>,
- /// Some on shard0, None on other shards
+ /// `Some` on shard 0, `None` on other shards.
pub snapshot: Option<S>,
/// State machine - lives on all shards
pub mux_stm: M,
@@ -363,7 +396,11 @@ where
>,
{
async fn on_request(&self, message: <VsrConsensus<B> as Consensus>::Message<RequestHeader>) {
- let consensus = self.consensus.as_ref().unwrap();
+ let Some(consensus) =
+ require_shard_zero(self.consensus.as_ref(), "on_request", "consensus")
+ else {
+ return;
+ };
let client_id = message.header().client;
let session = message.header().session;
let request = message.header().request;
@@ -431,8 +468,15 @@ where
}
async fn on_replicate(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareHeader>) {
- let consensus = self.consensus.as_ref().unwrap();
- let journal = self.journal.as_ref().unwrap();
+ let Some(consensus) =
+ require_shard_zero(self.consensus.as_ref(), "on_replicate", "consensus")
+ else {
+ return;
+ };
+ let Some(journal) = require_shard_zero(self.journal.as_ref(), "on_replicate", "journal")
+ else {
+ return;
+ };
let header = *message.header();
@@ -517,6 +561,19 @@ where
// un-persisted prepare advertises an op the WAL doesn't hold,
// violates VSR tail-ahead-of-head, recoverable only via hash-chain
// fence + view change (burns a view).
+ //
+ // TODO(hubcio): the primary path violates the invariant in the
+ // comment above. `consensus::impls::push_prepare_entry` pre-advances
+ // `sequencer.set_sequence(header.op)` and
+ // `set_last_prepare_checksum(header.checksum)` BEFORE this append.
+ // If the append below returns `Err`, sequencer + checksum stay
+ // advanced while the WAL holds no matching entry: the next prepare
+ // chains off a phantom op, cluster state diverges, and the
+ // `MetadataHandoff::Waiter` factory bundle propagates the divergence
+ // to peers. Fix: rollback `sequencer.set_sequence` +
+ // `set_last_prepare_checksum` to their captured prior values on
+ // append failure (preferred per CLAUDE.md "no panics in libraries"),
+ // or abort the shard.
if let Err(e) = journal.handle().append(message.clone()).await {
error!(
target: "iggy.metadata.diag",
@@ -705,9 +762,13 @@ where
.send_to_client(prepare_header.client, reply_buffers)
.await
{
- warn!(
- "on_ack: failed to send reply to client={}: {e}",
- prepare_header.client
+ error!(
+ client = prepare_header.client,
+ op = prepare_header.op,
+ request_id = prepare_header.request,
+ operation = ?prepare_header.operation,
+ %e,
+ "client reply forward failed, no retransmit path; client will time out",
);
}
}
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 3d8d0d436..06987f062 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -98,6 +98,9 @@ pub struct RecoveredMetadata<M> {
/// 4. Replay journal entries from the first post-snapshot op through the state machine
/// 5. Return the assembled `RecoveredMetadata`
///
+/// Only the owning shard (shard 0) should call this. Peer shards receive
+/// a `ReadHandleFactory` bundle from shard 0 and skip WAL access entirely.
+///
/// # Errors
/// Returns `RecoveryError` if snapshot loading, journal opening, or replay fails.
#[allow(clippy::future_not_send)]
@@ -110,7 +113,6 @@ where
let metadata_dir = data_dir.join(super::METADATA_DIR);
std::fs::create_dir_all(&metadata_dir)?;
- // 1. Load snapshot if present.
let snapshot_path = metadata_dir.join("snapshot.bin");
let snapshot = if snapshot_path.exists() {
Some(IggySnapshot::load(&snapshot_path)?)
@@ -121,24 +123,17 @@ where
.as_ref()
.map_or(0, |snapshot| snapshot.sequence_number() + 1);
- // 2. Restore state machine from snapshot, or start from a fresh empty state.
let mux_stm = match snapshot.as_ref() {
Some(snapshot) => M::restore_snapshot(snapshot.snapshot())?,
None => M::default(),
};
- // 3. Open journal, scan the WAL and build index
let journal_path = metadata_dir.join("journal.wal");
- let journal = PrepareJournal::open(
- &journal_path,
- snapshot.as_ref().map_or(0, IggySnapshot::sequence_number),
- )
- .await?;
-
- // 4. Replay journal entries after snapshot.
- // Intentional fail-fast for now: if replay hits a bad entry, recovery
- // aborts and the operator must repair or truncate the WAL before the
- // node can boot again.
+ let watermark = snapshot.as_ref().map_or(0, IggySnapshot::sequence_number);
+ let journal = PrepareJournal::open(&journal_path, watermark).await?;
+
+ // Intentional fail-fast: a bad entry aborts recovery and the operator
+ // must repair or truncate the WAL before the node can boot again.
let headers_to_replay = journal.iter_headers_from(replay_from);
let mut last_applied_op: Option<u64> = None;
diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs
index bde07b02e..90f19c903 100644
--- a/core/metadata/src/stm/mod.rs
+++ b/core/metadata/src/stm/mod.rs
@@ -23,11 +23,13 @@ pub mod user;
use bytes::Bytes;
use iggy_common::Either;
-use left_right::{Absorb, ReadHandle, WriteHandle};
+use left_right::{Absorb, ReadHandle, ReadHandleFactory, WriteHandle};
use std::cell::Cell;
use std::cell::UnsafeCell;
use std::sync::Arc;
+pub use left_right::ReadHandleFactory as LeftRightFactory;
+
pub struct WriteCell<T, O>
where
T: Absorb<O>,
@@ -113,8 +115,63 @@ where
let guard = self.read.enter().expect("read handle should be accessible");
f(&*guard)
}
+
+ /// Produce a `Send + Sync` factory that can mint additional [`ReadHandle`]s
+ /// on other threads. Use this to share the read side across compio shard
+ /// runtimes without sharing the (`!Sync`) [`ReadHandle`] itself.
+ #[must_use]
+ pub fn factory(&self) -> ReadHandleFactory<T> {
+ self.read.factory()
+ }
+
+ /// Build a reader-only [`LeftRight`] from a factory minted on the owning
+ /// thread. The new [`ReadHandle`] is materialised here (the caller's
+ /// thread), keeping the `!Sync` discipline local. `write` stays `None`,
+ /// so [`Self::try_apply`] returns
+ /// [`ApplyOnReaderError::ReaderOnly`] instead of mutating state if the
+ /// routing layer ever dispatches a write to this instance.
+ ///
+ /// Takes the factory by value so the constructor reads as ownership
+ /// transfer from the owning shard to the peer-shard reader.
+ #[must_use]
+ #[allow(clippy::needless_pass_by_value)]
+ pub fn from_factory(factory: ReadHandleFactory<T>) -> Self {
+ Self {
+ write: None,
+ read: Arc::new(factory.handle()),
+ }
+ }
+}
+
+/// Error returned by [`LeftRight::try_apply`] when invoked on a
+/// reader-only instance built via [`LeftRight::from_factory`].
+///
+/// A reader-only [`LeftRight`] holds no [`WriteCell`] (`write` is `None`)
+/// because peer shards rebuild their state machine from a
+/// [`ReadHandleFactory`] minted on the owning shard. Hitting this error
+/// means the routing layer dispatched a write to a non-owner shard,
+/// which is a programming error - the namespace-to-shard mapping owns
+/// the invariant.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum ApplyOnReaderError {
+ /// `try_apply` was called on a reader-only [`LeftRight`] (no write
+ /// handle). The command was not applied; state is unchanged.
+ ReaderOnly,
+}
+
+impl std::fmt::Display for ApplyOnReaderError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::ReaderOnly => f.write_str(
+ "apply invoked on reader-only LeftRight (no write handle on this shard)",
+ ),
+ }
+ }
}
+impl std::error::Error for ApplyOnReaderError {}
+
impl<T> From<T> for LeftRight<T, <T as Command>::Cmd>
where
T: Absorb<<T as Command>::Cmd> + Clone + Command,
@@ -132,13 +189,19 @@ impl<T> LeftRight<T, <T as Command>::Cmd>
where
T: Absorb<<T as Command>::Cmd> + Clone + Command,
{
- /// # Panics
- /// Panics if this is not the owner shard (no write handle available).
- pub fn do_apply(&self, cmd: <T as Command>::Cmd) {
- self.write
- .as_ref()
- .expect("no write handle - not the owner shard")
- .apply(cmd);
+ /// Apply `cmd` to the underlying writer.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`ApplyOnReaderError::ReaderOnly`] when invoked on an
+ /// instance built via [`Self::from_factory`] (peer-shard reader). The
+ /// command is dropped without mutating state - the caller is the
+ /// routing layer and the invariant violation must be surfaced, not
+ /// hidden behind a panic.
+ pub fn try_apply(&self, cmd: <T as Command>::Cmd) -> Result<(), ApplyOnReaderError> {
+ let cell = self.write.as_ref().ok_or(ApplyOnReaderError::ReaderOnly)?;
+ cell.apply(cmd);
+ Ok(())
}
}
@@ -256,6 +319,41 @@ macro_rules! define_state {
[<$state Inner>]::new().into()
}
}
+
+ impl $state {
+ /// Mint a `Send + Sync` [`ReadHandleFactory`] for this state.
+ /// Allows the read side to be carried across shard threads
+ /// without sharing the underlying `!Sync` [`ReadHandle`].
+ #[must_use]
+ pub fn factory(&self) -> $crate::stm::LeftRightFactory<[<$state Inner>]> {
+ self.inner.factory()
+ }
+
+ /// Construct a reader-only state wrapper from a factory minted
+ /// by the writer-side shard. The thread that calls this owns
+ /// the resulting [`ReadHandle`]; calling `apply` on the
+ /// returned wrapper panics because `write` is `None`.
+ #[must_use]
+ pub fn from_factory(
+ factory: $crate::stm::LeftRightFactory<[<$state Inner>]>,
+ ) -> Self {
+ Self {
+ inner: $crate::stm::LeftRight::from_factory(factory),
+ }
+ }
+ }
+
+ impl $crate::stm::mux::WithFactory for $state {
+ type Bundle = $crate::stm::LeftRightFactory<[<$state Inner>]>;
+
+ fn factory_bundle(&self) -> Self::Bundle {
+ self.factory()
+ }
+
+ fn from_factory_bundle(bundle: Self::Bundle) -> Self {
+ Self::from_factory(bundle)
+ }
+ }
}
};
}
@@ -336,7 +434,15 @@ macro_rules! collect_handlers {
match <[<$state Inner>] as $crate::stm::Command>::parse(input)? {
Either::Left(cmd) => {
- self.inner.do_apply(cmd);
+ self.inner.try_apply(cmd).map_err(|err| {
+ ::tracing::error!(
+ state = stringify!($state),
+ "metadata write dispatched to reader-only \
+ state machine: {err}; the routing layer must \
+ own namespace-to-shard mapping for this state"
+ );
+ ::iggy_common::IggyError::InvalidConfiguration
+ })?;
let result = self.inner.read(|state| {
state.last_result.clone().unwrap_or_default()
});
@@ -365,3 +471,68 @@ macro_rules! collect_handlers {
}
};
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use left_right::Absorb;
+
+ #[derive(Debug, Default, Clone)]
+ struct Counter(u64);
+
+ #[derive(Debug, Clone, Copy)]
+ enum CounterCmd {
+ Inc,
+ }
+
+ impl Absorb<CounterCmd> for Counter {
+ fn absorb_first(&mut self, _cmd: &mut CounterCmd, _other: &Self) {
+ self.0 += 1;
+ }
+
+ fn absorb_second(&mut self, _cmd: CounterCmd, _other: &Self) {
+ self.0 += 1;
+ }
+
+ fn sync_with(&mut self, first: &Self) {
+ *self = first.clone();
+ }
+ }
+
+ impl Command for Counter {
+ type Cmd = CounterCmd;
+ type Input = ();
+ type Error = ();
+
+ fn parse(_input: Self::Input) -> Result<Either<Self::Cmd, Self::Input>, Self::Error> {
+ Ok(Either::Left(CounterCmd::Inc))
+ }
+ }
+
+ #[test]
+ fn try_apply_on_writer_mutates_state() {
+ let lr: LeftRight<Counter, CounterCmd> = Counter::default().into();
+ lr.try_apply(CounterCmd::Inc)
+ .expect("writer must accept apply");
+ assert_eq!(lr.read(|c| c.0), 1);
+ }
+
+ #[test]
+ fn try_apply_on_reader_returns_reader_only_error() {
+ // Reader-only LeftRight (built from a factory minted on the writer)
+ // must surface `ApplyOnReaderError::ReaderOnly` instead of mutating
+ // state or panicking - the routing layer is the only caller and the
+ // invariant violation has to be visible to it.
+ let writer: LeftRight<Counter, CounterCmd> = Counter::default().into();
+ let reader = LeftRight::<Counter, CounterCmd>::from_factory(writer.factory());
+ let err = reader
+ .try_apply(CounterCmd::Inc)
+ .expect_err("reader-only instance must reject try_apply");
+ assert_eq!(err, ApplyOnReaderError::ReaderOnly);
+ assert_eq!(
+ reader.read(|c| c.0),
+ 0,
+ "reader state must be unchanged after a rejected try_apply"
+ );
+ }
+}
diff --git a/core/metadata/src/stm/mux.rs b/core/metadata/src/stm/mux.rs
index 8097cc750..c9a0d1e6a 100644
--- a/core/metadata/src/stm/mux.rs
+++ b/core/metadata/src/stm/mux.rs
@@ -22,6 +22,49 @@ use iggy_common::variadic;
use crate::stm::{State, StateMachine};
+/// Cross-thread read-side handoff for a state machine.
+///
+/// The owning shard (typically shard 0) materialises a `Bundle` after
+/// recovering the WAL and hands one clone to every peer shard. Peer
+/// shards then call [`WithFactory::from_factory_bundle`] on their own
+/// runtime to construct a reader-mode state machine — no WAL replay,
+/// no shared `!Sync` `ReadHandle`, just `Send + Sync` factories per
+/// state.
+///
+/// The variadic tuple impl below recurses over the state list, so a
+/// `MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>` ends up
+/// with `Bundle = (Users::Bundle, (Streams::Bundle, (ConsumerGroups::Bundle, ())))`.
+pub trait WithFactory: Sized {
+ type Bundle: Clone + Send + Sync;
+ fn factory_bundle(&self) -> Self::Bundle;
+ fn from_factory_bundle(bundle: Self::Bundle) -> Self;
+}
+
+impl WithFactory for () {
+ type Bundle = ();
+ fn factory_bundle(&self) -> Self::Bundle {}
+ fn from_factory_bundle((): Self::Bundle) -> Self {}
+}
+
+impl<S, Rest> WithFactory for variadic!(S, ...Rest)
+where
+ S: WithFactory,
+ Rest: WithFactory,
+{
+ type Bundle = (S::Bundle, Rest::Bundle);
+
+ fn factory_bundle(&self) -> Self::Bundle {
+ (self.0.factory_bundle(), self.1.factory_bundle())
+ }
+
+ fn from_factory_bundle(bundle: Self::Bundle) -> Self {
+ (
+ S::from_factory_bundle(bundle.0),
+ Rest::from_factory_bundle(bundle.1),
+ )
+ }
+}
+
// MuxStateMachine that proxies to an tuple of variadic state machines
#[derive(Debug)]
pub struct MuxStateMachine<T>
@@ -55,6 +98,26 @@ where
}
}
+impl<T> MuxStateMachine<T>
+where
+ T: StateMachine + WithFactory,
+{
+ /// Mint a factory bundle from the owning state machine so peer shards
+ /// can rebuild a reader-mode mirror on their own runtimes.
+ #[must_use]
+ pub fn factory_bundle(&self) -> T::Bundle {
+ self.inner.factory_bundle()
+ }
+
+ /// Reconstruct a reader-mode `MuxStateMachine` from a bundle minted
+ /// on the writer-side shard. Must be invoked on the destination
+ /// runtime so the inner `!Sync` `ReadHandle`s are created in place.
+ #[must_use]
+ pub fn from_factory_bundle(bundle: T::Bundle) -> Self {
+ Self::new(T::from_factory_bundle(bundle))
+ }
+}
+
impl<T> StateMachine for MuxStateMachine<T>
where
T: StateMachine,
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 4a82aaec8..f57bd790b 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -1416,14 +1416,14 @@ where
.send_to_client(prepare_header.client, reply_buffers)
.await
{
- warn!(
+ tracing::error!(
target: "iggy.partitions.diag",
plane = "partitions",
client = prepare_header.client,
op = prepare_header.op,
namespace_raw,
%error,
- "failed to send reply to client"
+ "client reply forward failed, no retransmit path; client will time out",
);
}
}
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index d11700a19..3b130fdb4 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -43,7 +43,6 @@ ignored = [
"err_trail",
"error_set",
"figlet-rs",
- "flume",
"fs2",
"hash32",
"human-repr",
@@ -105,6 +104,7 @@ clap = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
consensus = { workspace = true }
+crossfire = { workspace = true }
ctrlc = { workspace = true }
cyper = { workspace = true }
cyper-axum = { workspace = true }
@@ -113,7 +113,6 @@ dotenvy = { workspace = true }
err_trail = { workspace = true }
error_set = { workspace = true }
figlet-rs = { workspace = true }
-flume = { workspace = true }
fs2 = { workspace = true }
hash32 = { workspace = true }
human-repr = { workspace = true }
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 7d8b51123..4aa14cc72 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -589,7 +589,22 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket = 8093, tcp_replica =
# - numa settings:
# + "numa:auto": Use all available numa node, cores
# + "numa:nodes=0,1;cores=4;no_ht=true": Use NUMA node 0 and 1, each nodes use 4 cores, and no hyperthreads
-cpu_allocation = "numa:auto"
+# TODO(hubcio): revert to "numa:auto" once multi-shard server-ng is stable.
+cpu_allocation = 1
+
+# Wall-clock budget for a single shard's bus drain on shutdown. Drives
+# the per-shard watchdog and the parallel-join survivor path; sized
+# larger than typical TCP RTT times in-flight write-batch so writers
+# receive their full last `write_vectored_all` budget before the
+# connection registry force-tears the bus. Slow-fsync hosts may need
+# to extend this past the default.
+shutdown_drain_timeout = "10 s"
+
+# Poll cadence for the cross-thread shutdown flag and for the
+# metadata-handoff loops. Trades off Ctrl-C latency against idle wakeup
+# cost; the default keeps shutdown observably prompt without measurable
+# scheduler overhead. Must be less than or equal to shutdown_drain_timeout.
+shutdown_poll_interval = "50 ms"
[websocket]
enabled = true
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 816a464f5..79ecf1748 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -18,9 +18,16 @@
*/
use crate::config_writer::write_current_config;
-use crate::server_error::ServerNgError;
+use crate::server_error::{ServerNgError, ShardJoinFailure, ShardJoinFailureKind};
use configs::server_ng::ServerNgConfig;
+use configs::sharding::{
+ INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
+};
use consensus::{LocalPipeline, PartitionsHandle, Sequencer, VsrConsensus};
+// `try_send` / `try_recv` resolve through these traits on `MAsyncTx` /
+// `MAsyncRx`; the metadata-handoff loops below depend on the
+// non-blocking variants for cancel-safe shutdown polling.
+use crossfire::{AsyncRxTrait, AsyncTxTrait};
use iggy_binary_protocol::RequestHeader;
use iggy_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
use iggy_common::{
@@ -30,9 +37,7 @@ use iggy_common::{
use journal::Journal;
use journal::prepare_journal::PrepareJournal;
use message_bus::client_listener::{self, RequestHandler};
-use message_bus::fd_transfer;
use message_bus::installer;
-use message_bus::installer::ConnectionInstaller;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
use message_bus::replica::io as replica_io;
use message_bus::replica::listener::{self as replica_listener, MessageHandler};
@@ -42,13 +47,14 @@ use message_bus::transports::tls::{
};
use message_bus::{
AcceptedClientFn, AcceptedQuicClientFn, AcceptedReplicaFn, AcceptedTlsClientFn,
- AcceptedWsClientFn, IggyMessageBus, connector,
+ AcceptedWsClientFn, IggyMessageBus, ReplicaOwnerTable, connector,
};
use metadata::IggyMetadata;
use metadata::MuxStateMachine;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
use metadata::impls::recovery::recover;
use metadata::stm::consumer_group::ConsumerGroups;
+use metadata::stm::mux::WithFactory;
use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
use metadata::stm::user::Users;
@@ -56,38 +62,46 @@ use partitions::{
IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
};
// TODO: decouple bootstrap/storage helpers and logging from the `server` crate.
-use server::bootstrap::create_directories;
+use server::bootstrap::{create_directories, create_shard_executor};
use server::log::logger::Logging;
+use server::shard_allocator::{ShardAllocator, ShardInfo};
use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets};
use server::streaming::segments::storage::create_segment_storage;
use shard::builder::IggyShardBuilder;
-use shard::shards_table::PapayaShardsTable;
+use shard::metrics::ShardMetrics;
+use shard::shards_table::{PapayaShardsTable, calculate_shard_assignment};
use shard::{
- CoordinatorConfig, IggyShard, PartitionConsensusConfig, ShardIdentity, channel, shard_channel,
+ CoordinatorConfig, IggyShard, PartitionConsensusConfig, Receiver as ShardReceiver, ShardFrame,
+ ShardIdentity, TaggedSender, channel, shard_mesh_channels,
};
-use std::cell::{Cell, RefCell};
-use std::future::Future;
+use std::cell::RefCell;
use std::net::{IpAddr, SocketAddr};
use std::path::{Path, PathBuf};
use std::rc::{Rc, Weak};
use std::sync::Arc;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::thread;
+use std::time::Duration;
use tracing::{error, info, warn};
const CLUSTER_ID: u128 = 1;
-const SHARD_ID: u16 = 0;
const SHARD_REPLICA_ID: u8 = 0;
-const SHARD_NAME: &str = "server-ng-shard-0";
-const SHARD_INBOX_CAPACITY: usize = 1024;
type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+
+/// Cross-thread bundle carrying one `ReadHandleFactory` per metadata
+/// state. Shard 0 mints one after `recover()` and broadcasts a clone to
+/// every peer shard; each peer rebuilds a reader-mode
+/// [`ServerNgMuxStateMachine`] on its own runtime, skipping the WAL.
+type ServerNgMetadataBundle = <variadic!(Users, Streams, ConsumerGroups) as WithFactory>::Bundle;
+
type ServerNgMetadata = IggyMetadata<
VsrConsensus<Rc<IggyMessageBus>>,
PrepareJournal,
IggySnapshot,
ServerNgMuxStateMachine,
>;
-type ServerNgShard = IggyShard<
+pub type ServerNgShard = IggyShard<
Rc<IggyMessageBus>,
PrepareJournal,
IggySnapshot,
@@ -97,6 +111,228 @@ type ServerNgShard = IggyShard<
type ServerNgShardHandle = Rc<RefCell<Option<Weak<ServerNgShard>>>>;
+/// Result of a multi-shard bootstrap.
+///
+/// Carries the cross-thread shutdown flag and one OS-thread `JoinHandle`
+/// per shard. The caller flips the flag via [`Self::install_ctrlc_handler`]
+/// and then drains every shard via [`Self::join_all`].
+pub struct ShardHandles {
+ shutdown_flag: Arc<AtomicBool>,
+ shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerNgError>>)>,
+}
+
+impl ShardHandles {
+ /// Install a SIGINT/Ctrl-C handler that flips the shutdown flag on
+ /// the first signal. A second signal is logged but otherwise
+ /// ignored so an in-flight WAL fsync or replica drain runs to
+ /// completion.
+ ///
+ /// # Errors
+ ///
+ /// Returns the underlying `ctrlc::Error` if the handler cannot be
+ /// installed (typically because another handler already owns the
+ /// signal).
+ pub fn install_ctrlc_handler(&self) -> Result<(), ctrlc::Error> {
+ let flag = Arc::clone(&self.shutdown_flag);
+ ctrlc::set_handler(move || {
+ if flag.swap(true, Ordering::Relaxed) {
+ // Second Ctrl-C: leave the shutdown machinery to drain.
+ // Refusing to abort here keeps the WAL fsync / replica
+ // drain from being interrupted mid-frame.
+ warn!("second Ctrl-C ignored; server is already shutting down");
+ } else {
+ info!("Ctrl-C received; signalling server shutdown");
+ }
+ })
+ }
+
+ /// Drain every shard thread. Each shard's outcome is logged
+ /// (`info` on clean exit, `error` on Err or panic). If any shard
+ /// failed, returns every failure together as
+ /// [`ServerNgError::ShardJoinFailures`] so the operator sees the
+ /// full set rather than just the first.
+ ///
+ /// # Errors
+ ///
+ /// Returns [`ServerNgError::ShardJoinFailures`] if any shard
+ /// returned a `Result::Err` or panicked. The variant carries every
+ /// per-shard failure (`ShardJoinFailureKind::Error` or
+ /// `ShardJoinFailureKind::Panic`) in shard-id order so the caller
+ /// does not need to read the trace log to discover late-failing shards.
+ pub fn join_all(self) -> Result<(), ServerNgError> {
+ let mut failures: Vec<ShardJoinFailure> = Vec::new();
+ for (shard_id, handle) in self.shard_threads {
+ match handle.join() {
+ Ok(Ok(())) => {
+ info!(shard_id, "shard thread exited cleanly");
+ }
+ Ok(Err(error)) => {
+ error!(shard_id, error = %error, "shard thread returned error");
+ failures.push(ShardJoinFailure {
+ shard_id,
+ kind: ShardJoinFailureKind::Error(Box::new(error)),
+ });
+ }
+ Err(panic_payload) => {
+ let message = panic_payload_to_string(&*panic_payload);
+ error!(shard_id, message = %message, "shard thread panicked");
+ failures.push(ShardJoinFailure {
+ shard_id,
+ kind: ShardJoinFailureKind::Panic { message },
+ });
+ }
+ }
+ }
+ if failures.is_empty() {
+ Ok(())
+ } else {
+ Err(ServerNgError::ShardJoinFailures { failures })
+ }
+ }
+}
+
+/// Best-effort extraction of the panic message from a
+/// `Box<dyn Any + Send>` returned by `JoinHandle::join`. Tries the two
+/// payload shapes the standard library guarantees (`&'static str` and
+/// `String`) and falls back to a placeholder so the panic still surfaces
+/// in the error chain.
+fn panic_payload_to_string(payload: &(dyn std::any::Any + Send)) -> String {
+ if let Some(s) = payload.downcast_ref::<&'static str>() {
+ return (*s).to_string();
+ }
+ if let Some(s) = payload.downcast_ref::<String>() {
+ return s.clone();
+ }
+ "<panic payload not String/&str>".to_string()
+}
+
+/// Joins survivor shard threads after a partial-spawn failure without
+/// panicking the bootstrap thread on `pthread_create` EAGAIN.
+///
+/// Bare `thread::spawn` panics on EAGAIN, which is the most likely OS
+/// state on this path since the parent `Builder::spawn` already failed
+/// for the same reason. A panic would unwind `bootstrap()` while
+/// survivor shard threads keep driving their compio runtimes and
+/// `io_uring` rings, orphaning them across process exit.
+///
+/// Uses `thread::Builder::spawn` and hands each survivor over via a
+/// one-shot `sync_channel(1)` so an `Err` drops the rx (not the
+/// survivor `JoinHandle`), letting us fall back to a sequential
+/// `survivor.join()` instead. Once one cleanup spawn fails, treats the
+/// OS as exhausted and routes every remaining survivor straight to the
+/// sequential pool to avoid re-trying spawn.
+///
+/// This routine bounds CPU/IO via the survivor's own
+/// `shutdown_drain_timeout` (driven by each shard's watchdog after
+/// `shutdown_flag` is set by the caller), not via a wall-clock
+/// deadline here. If a survivor's `shard_main` blocks past the drain
+/// window without observing the flag, this join hangs - that scenario
+/// is the same surface as the deferred watchdog-detach gap and is not
+/// addressed by this helper.
+///
+/// TODO(hubcio): no hard time limit on shard shutdown here. If a
+/// survivor's `shard_main` never returns, `survivor.join()` blocks
+/// forever.
+fn join_partial_shard_survivors(
+ shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerNgError>>)>,
+) {
+ let mut joiners = Vec::with_capacity(shard_threads.len());
+ let mut sequential_join: Vec<thread::JoinHandle<Result<(), ServerNgError>>> = Vec::new();
+ let mut spawn_exhausted = false;
+ for (sid, survivor) in shard_threads {
+ if spawn_exhausted {
+ sequential_join.push(survivor);
+ continue;
+ }
+ let (handover_tx, handover_rx) =
+ std::sync::mpsc::sync_channel::<thread::JoinHandle<Result<(), ServerNgError>>>(1);
+ match thread::Builder::new()
+ .name(format!("shard-{sid}-cleanup"))
+ .spawn(move || {
+ if let Ok(survivor) = handover_rx.recv() {
+ let _ = survivor.join();
+ }
+ }) {
+ Ok(joiner) => {
+ let _ = handover_tx.send(survivor);
+ joiners.push(joiner);
+ }
+ Err(spawn_err) => {
+ warn!(
+ error = %spawn_err,
+ shard_id = sid,
+ "cleanup helper thread spawn failed; falling back to sequential survivor join"
+ );
+ spawn_exhausted = true;
+ sequential_join.push(survivor);
+ }
+ }
+ }
+ for joiner in joiners {
+ let _ = joiner.join();
+ }
+ for survivor in sequential_join {
+ let _ = survivor.join();
+ }
+}
+
+/// Flips the cross-thread shutdown flag on `Drop` unless disarmed.
+///
+/// A shard thread that exits via an error `?` or a panic unwind would
+/// otherwise leave sibling shards parked forever on `bus.token().wait()`:
+/// their watchdogs never observe the flag and the bus has no
+/// `Drop`-triggered shutdown. Arming this for the whole thread body makes
+/// every non-clean exit drive sibling-shard teardown. Disarmed only on a
+/// clean `Ok(())`.
+struct ShutdownOnDrop {
+ flag: Arc<AtomicBool>,
+ armed: bool,
+}
+
+impl ShutdownOnDrop {
+ const fn new(flag: Arc<AtomicBool>) -> Self {
+ Self { flag, armed: true }
+ }
+
+ const fn disarm(&mut self) {
+ self.armed = false;
+ }
+}
+
+impl Drop for ShutdownOnDrop {
+ fn drop(&mut self) {
+ if self.armed {
+ self.flag.store(true, Ordering::Relaxed);
+ }
+ }
+}
+
+/// Shard-local end of the metadata bundle handoff.
+///
+/// Shard 0 owns the WAL writer and runs `recover()` to build the only
+/// `WriteHandle`-bearing [`ServerNgMuxStateMachine`]. It then mints a
+/// [`ServerNgMetadataBundle`] (a tuple of `Send + Sync`
+/// `ReadHandleFactory`s) and pushes one clone per peer onto `bundle_tx`.
+/// Every other shard receives the bundle and rebuilds a reader-mode
+/// `MuxStateMachine` on its own runtime - no WAL access, no replay, no
+/// `RecoverySync` two-phase fence. Phase 2 of the old handshake was
+/// only there to keep peer scans away from shard 0's torn-tail repair;
+/// with no peer scan that race is structurally gone.
+///
+/// The channel is bounded to the peer count so shard 0's `send` never
+/// blocks beyond a peer drain. A peer that dies before recv drops its
+/// `bundle_rx`, so shard 0's `send` eventually sees a disconnected
+/// channel; the cross-thread shutdown flag drives every waiter out of
+/// its `recv` loop if shard 0 panics before broadcasting.
+enum MetadataHandoff {
+ Owner {
+ bundle_tx: crossfire::MAsyncTx<crossfire::mpmc::Array<ServerNgMetadataBundle>>,
+ },
+ Waiter {
+ bundle_rx: crossfire::MAsyncRx<crossfire::mpmc::Array<ServerNgMetadataBundle>>,
+ },
+}
+
struct TcpTopology {
self_replica_id: u8,
replica_count: u8,
@@ -123,58 +359,6 @@ struct BoundClientListeners {
quic: Option<SocketAddr>,
}
-pub trait RunServerNg {
- fn run(
- &self,
- config: &ServerNgConfig,
- current_replica_id: Option<u8>,
- ) -> impl Future<Output = Result<(), ServerNgError>>;
-}
-
-impl RunServerNg for Rc<ServerNgShard> {
- /// Run the fully bootstrapped `server-ng` shard.
- ///
- /// # Errors
- ///
- /// Returns an error if TCP listener bootstrap fails or cluster TCP
- /// addresses cannot be resolved from config.
- async fn run(
- &self,
- config: &ServerNgConfig,
- current_replica_id: Option<u8>,
- ) -> Result<(), ServerNgError> {
- let topology = resolve_tcp_topology(config, current_replica_id)?;
- let (stop_tx, stop_rx) = channel(1);
- let message_pump_shard = Self::clone(self);
- let message_pump_handle = compio::runtime::spawn(async move {
- message_pump_shard.run_message_pump(stop_rx).await;
- });
- self.bus.track_background(message_pump_handle);
-
- let on_replica_message = make_replica_message_handler(self);
- let on_client_request = make_client_request_handler(self);
- let accepted_replica = make_local_replica_accept_fn(&self.bus, on_replica_message);
- let accepted_client = make_local_client_accept_fns(&self.bus, on_client_request);
-
- info!(
- shard = self.id,
- partitions = self.plane.partitions().len(),
- "server-ng shard initialized"
- );
-
- if let Err(error) =
- start_tcp_runtime(self, config, &topology, accepted_replica, accepted_client).await
- {
- let _ = stop_tx.try_send(());
- return Err(error);
- }
-
- self.bus.token().wait().await;
- let _ = stop_tx.try_send(());
- Ok(())
- }
-}
-
/// Load config, prepare directories, and complete late logging init.
///
/// # Errors
@@ -205,92 +389,592 @@ pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, Server
Ok(config)
}
-/// Bootstraps `server-ng` from config and on-disk metadata/partition state.
+/// Re-validate the runtime sharding knobs that the per-shard runtime
+/// consumes directly. Mirrors `ShardingConfig::validate` so a caller
+/// that built the config without running it (e.g. tests, embedded
+/// usage) cannot OOM at boot or wedge process exit with an out-of-range
+/// value.
+fn validate_sharding_runtime_knobs(
+ sharding: &configs::sharding::ShardingConfig,
+) -> Result<(), ServerNgError> {
+ let inbox_capacity = sharding.inbox_capacity;
+ if inbox_capacity == 0 || inbox_capacity > INBOX_CAPACITY_MAX {
+ return Err(ServerNgError::InvalidInboxCapacity {
+ value: inbox_capacity,
+ max: INBOX_CAPACITY_MAX,
+ });
+ }
+ let drain_timeout = sharding.shutdown_drain_timeout.get_duration();
+ if drain_timeout.is_zero() || drain_timeout > SHUTDOWN_DRAIN_TIMEOUT_MAX {
+ return Err(ServerNgError::InvalidShutdownDrainTimeout {
+ value: drain_timeout,
+ max: SHUTDOWN_DRAIN_TIMEOUT_MAX,
+ });
+ }
+ let poll_interval = sharding.shutdown_poll_interval.get_duration();
+ if poll_interval.is_zero() || poll_interval > SHUTDOWN_POLL_INTERVAL_MAX {
+ return Err(ServerNgError::InvalidShutdownPollInterval {
+ value: poll_interval,
+ max: SHUTDOWN_POLL_INTERVAL_MAX,
+ });
+ }
+ // Ordering: a poll cadence coarser than the drain budget makes the
+ // cross-thread shutdown flag effectively unobservable during teardown.
+ if poll_interval > drain_timeout {
+ return Err(ServerNgError::ShutdownPollExceedsDrain {
+ poll: poll_interval,
+ drain: drain_timeout,
+ });
+ }
+ Ok(())
+}
+
+/// Spawn the multi-shard `server-ng` runtime.
+///
+/// Resolves shard count + CPU affinities from
+/// `system.sharding.cpu_allocation`, builds canonical-ordered
+/// `(senders, inboxes)` channels, and spawns one OS thread per shard.
+///
+/// Each thread pins itself (`nix::sched::sched_setaffinity` on Linux via
+/// [`ShardInfo::bind_cpu`]), binds memory to its NUMA node when
+/// configured, builds a fresh `compio::runtime::Runtime` (one
+/// `io_uring` instance per shard), and runs `shard_main` inside it.
+///
+/// Returns [`ShardHandles`] containing the cross-thread shutdown flag
+/// and the per-shard `JoinHandle`s. The caller (`main.rs`) installs a
+/// `ctrlc` handler that flips the flag, then `.join()`s every handle.
///
/// # Errors
///
-/// Returns an error if metadata recovery, consensus restoration, or
-/// partition hydration fails.
-pub async fn bootstrap(
- config: &ServerNgConfig,
+/// Returns an error if shard allocation fails, the inbox capacity is
+/// invalid, or any OS thread fails to spawn. Per-shard recovery /
+/// listener / consensus failures surface through the per-thread `Result`
+/// the caller observes on `.join()`.
+///
+/// # Panics
+///
+/// Panics if [`shard_mesh_channels`] returns an inbox slot already
+/// consumed - a bootstrap programming error that would only fire if this
+/// function were called twice with the same inboxes.
+pub fn bootstrap(
+ config: ServerNgConfig,
current_replica_id: Option<u8>,
-) -> Result<Rc<ServerNgShard>, ServerNgError> {
- let topology = resolve_tcp_topology(config, current_replica_id)?;
- let bus = Rc::new(IggyMessageBus::with_config(SHARD_ID, config));
- let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
+) -> Result<ShardHandles, ServerNgError> {
+ let allocator = ShardAllocator::new(&config.system.sharding.cpu_allocation)
+ .map_err(ServerNgError::ShardAllocator)?;
+ let assignments = allocator
+ .to_shard_assignments()
+ .map_err(ServerNgError::ShardAllocator)?;
+ let shards_count = assignments.len();
+ if shards_count == 0 {
+ return Err(ServerNgError::ShardsCountZero);
+ }
+ // Shard ids index `ReplicaOwnerTable` slots as `u16`. `OWNER_NONE`
+ // (`u16::MAX`) is reserved as the empty-slot sentinel, so a server
+ // configured with `u16::MAX` shards would mint a shard id that
+ // collides with the sentinel and an owner-table lookup could never
+ // tell that shard apart from an unowned slot. Reject at boot so the
+ // invariant is held by the type system above this line, not by hoping
+ // the operator never configures 65535 cores worth of shards.
+ let total_shards = match u16::try_from(shards_count) {
+ Ok(count) if count < message_bus::OWNER_NONE => count,
+ _ => {
+ return Err(ServerNgError::ShardsCountOverflow {
+ count: shards_count,
+ });
+ }
+ };
+
+ // Re-check the full valid range, not just the zero floor: a caller
+ // that built the config without running `ShardingConfig::validate`
+ // would otherwise OOM at boot allocating an oversized inbox channel,
+ // busy-loop every shutdown watchdog on a zero poll cadence, or wedge
+ // process exit on an unbounded drain budget.
+ let inbox_capacity = config.system.sharding.inbox_capacity;
+ validate_sharding_runtime_knobs(&config.system.sharding)?;
+
+ let (senders, mut inboxes) = shard_mesh_channels(total_shards, inbox_capacity);
+ let shutdown_flag = Arc::new(AtomicBool::new(false));
+ let config = Arc::new(config);
+ // One owner table per server process, Arc-cloned into every shard's bus so
+ // any shard's bus reads the same atomic slots that the owning
+ // shard's installer / disconnect path writes.
+ let owner_table = Arc::new(ReplicaOwnerTable::new());
+
+ // Single-shot bundle handoff (see `MetadataHandoff`): shard 0 sends
+ // one cloned `ServerNgMetadataBundle` per peer; each peer drains
+ // exactly one. Bounded to the peer count so shard 0's broadcast
+ // never blocks past a peer drain. A single-shard deployment (zero
+ // peers) still needs a non-zero capacity, so clamp up explicitly
+ // rather than relying on crossfire's internal cap=0 -> 1 promotion.
+ // If a peer dies before recv, shard 0's `send` eventually sees a
+ // disconnected channel; the cross-thread shutdown flag drives every
+ // waiter out of its recv loop if shard 0 panics before broadcasting.
+ let metadata_peers = shards_count.saturating_sub(1).max(1);
+ let (metadata_bundle_tx, metadata_bundle_rx) =
+ crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(metadata_peers);
+
+ let mut shard_threads: Vec<(u16, thread::JoinHandle<Result<(), ServerNgError>>)> =
+ Vec::with_capacity(shards_count);
+ for (idx, assignment) in assignments.into_iter().enumerate() {
+ #[allow(clippy::cast_possible_truncation)]
+ let shard_id = idx as u16;
+ let inbox = inboxes[idx]
+ .take()
+ .expect("shard_mesh_channels populates every inbox slot exactly once");
+ let senders_for_shard = senders.clone();
+ let config_for_shard = Arc::clone(&config);
+ let shutdown_flag_for_shard = Arc::clone(&shutdown_flag);
+ let owner_table_for_shard = Arc::clone(&owner_table);
+ let metadata_handoff_for_shard = if shard_id == 0 {
+ MetadataHandoff::Owner {
+ bundle_tx: metadata_bundle_tx.clone(),
+ }
+ } else {
+ MetadataHandoff::Waiter {
+ bundle_rx: metadata_bundle_rx.clone(),
+ }
+ };
+
+ let handle = match thread::Builder::new()
+ .name(format!("shard-{shard_id}"))
+ .spawn(move || -> Result<(), ServerNgError> {
+ run_shard_thread(
+ shard_id,
+ total_shards,
+ current_replica_id,
+ assignment,
+ senders_for_shard,
+ inbox,
+ config_for_shard,
+ shutdown_flag_for_shard,
+ metadata_handoff_for_shard,
+ owner_table_for_shard,
+ )
+ }) {
+ Ok(handle) => handle,
+ Err(source) => {
+ // Signal every shard already spawned before propagating, so
+ // their watchdog loops drive `bus.shutdown(...)` and the
+ // process can exit instead of hanging on stuck OS threads.
+ shutdown_flag.store(true, Ordering::Relaxed);
+ // Drop bootstrap's own channel clones before joining
+ // survivors. Otherwise a peer waiting on `bundle_rx.recv`
+ // would never observe the sender side disconnecting and
+ // would hang until the shutdown watchdog kicks the bus.
+ drop(metadata_bundle_tx);
+ drop(metadata_bundle_rx);
+ join_partial_shard_survivors(shard_threads);
+ return Err(ServerNgError::ShardSpawnFailed { shard_id, source });
+ }
+ };
+ shard_threads.push((shard_id, handle));
+ }
+
+ // Drop bootstrap's own channel clones now that every shard owns its
+ // half. Keeping them on bootstrap's stack would deadlock a peer
+ // whose `bundle_rx.recv` only completes once every sender
+ // disconnects.
+ drop(metadata_bundle_tx);
+ drop(metadata_bundle_rx);
+
+ info!(
+ shards_count,
+ "server-ng bootstrap dispatched; awaiting shard runtimes"
+ );
+
+ Ok(ShardHandles {
+ shutdown_flag,
+ shard_threads,
+ })
+}
+
+/// Per-shard OS thread entry. Pins CPU + memory, builds the compio
+/// runtime, and `block_on`s `shard_main`.
+#[allow(clippy::needless_pass_by_value, clippy::too_many_arguments)]
+fn run_shard_thread(
+ shard_id: u16,
+ total_shards: u16,
+ replica_id: Option<u8>,
+ assignment: ShardInfo,
+ senders: Vec<TaggedSender>,
+ inbox: ShardReceiver<ShardFrame>,
+ config: Arc<ServerNgConfig>,
+ shutdown_flag: Arc<AtomicBool>,
+ metadata_handoff: MetadataHandoff,
+ owner_table: Arc<ReplicaOwnerTable>,
+) -> Result<(), ServerNgError> {
+ // Armed for the whole thread body: a post-spawn error `?` or a panic
+ // unwind here must flip `shutdown_flag` so sibling watchdogs drive
+ // their bus shutdown instead of parking forever on `bus.token().wait()`.
+ let mut shutdown_guard = ShutdownOnDrop::new(Arc::clone(&shutdown_flag));
+
+ assignment
+ .bind_cpu()
+ .map_err(|source| ServerNgError::CpuAffinityFailed { shard_id, source })?;
+ assignment
+ .bind_memory()
+ .map_err(|source| ServerNgError::MemoryAffinityFailed { shard_id, source })?;
+
+ // TODO(hubcio): decouple runtime creation from the `server` crate
+ // (mirrors the identical TODO in `main.rs`). Reusing legacy here so
+ // server-ng and the legacy server share one io_uring tuning surface.
+ let runtime = create_shard_executor()
+ .map_err(|source| ServerNgError::ShardRuntimeCreateFailed { shard_id, source })?;
+
+ let result = runtime.block_on(async move {
+ shard_main(
+ shard_id,
+ total_shards,
+ replica_id,
+ senders,
+ inbox,
+ &config,
+ shutdown_flag,
+ metadata_handoff,
+ owner_table,
+ )
.await
- .map_err(ServerNgError::MetadataRecovery)?;
- let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
- recovered
- .snapshot
- .as_ref()
- .map_or(0, IggySnapshot::sequence_number)
});
+ if result.is_ok() {
+ shutdown_guard.disarm();
+ }
+ result
+}
+
+/// Per-shard async lifecycle. Builds the bus, recovers metadata,
+/// constructs the `IggyShard` for this shard's slice of partitions,
+/// wires listeners on shard 0, and runs the message pump until
+/// shutdown.
+#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
+async fn shard_main(
+ shard_id: u16,
+ total_shards: u16,
+ replica_id: Option<u8>,
+ senders: Vec<TaggedSender>,
+ inbox: ShardReceiver<ShardFrame>,
+ config: &ServerNgConfig,
+ shutdown_flag: Arc<AtomicBool>,
+ metadata_handoff: MetadataHandoff,
+ owner_table: Arc<ReplicaOwnerTable>,
+) -> Result<(), ServerNgError> {
+ let topology = resolve_tcp_topology(config, replica_id)?;
+ let bus = Rc::new(IggyMessageBus::with_config_and_owner_table(
+ shard_id,
+ config,
+ owner_table,
+ ));
+
+ let drain_timeout = config.system.sharding.shutdown_drain_timeout.get_duration();
+ let poll_interval = config.system.sharding.shutdown_poll_interval.get_duration();
+
+ let shutdown_flag_for_handoff = Arc::clone(&shutdown_flag);
+ spawn_shutdown_watchdog(Rc::clone(&bus), shutdown_flag, drain_timeout, poll_interval);
+
+ // Metadata bootstrap is single-writer: shard 0 owns the WAL and the
+ // only `WriteHandle`-bearing `MuxStateMachine`. Peer shards receive
+ // a `ReadHandleFactory` bundle on the inter-thread channel and
+ // rebuild a reader-mode `MuxStateMachine` on their own runtime - no
+ // WAL access, no replay. Writes still funnel through shard 0's
+ // metadata VSR; per-commit `publish()` (in `WriteCell::apply`)
+ // bounds reader staleness to one op.
+ let data_dir = Path::new(&config.system.path);
+ let (mux_stm, owner_state) = match metadata_handoff {
+ MetadataHandoff::Owner { bundle_tx } => {
+ let recovered = recover::<ServerNgMuxStateMachine>(data_dir)
+ .await
+ .map_err(ServerNgError::MetadataRecovery)?;
+ // The factory bundle hands every peer a read handle over the
+ // same `Inner`, so `Arc<TopicStats>` (and the parent
+ // `Arc<StreamStats>`) is shared across all shards. Zero the
+ // snapshot totals here, once, before any peer can observe the
+ // bundle. Per-shard `load_partition` deltas in
+ // `build_shard_for_thread` then race only against other
+ // atomic adds, never against a concurrent `swap(0)` that
+ // would mistake an in-flight delta for the snapshot total
+ // and decrement the parent `StreamStats` by it.
+ let () = recovered.mux_stm.streams().read(|inner| {
+ for (_, stream) in &inner.items {
+ for (_, topic) in &stream.topics {
+ topic.stats.zero_out_all();
+ }
+ }
+ });
+ broadcast_metadata_bundle(
+ shard_id,
+ &bundle_tx,
+ recovered.mux_stm.factory_bundle(),
+ total_shards.saturating_sub(1),
+ &shutdown_flag_for_handoff,
+ poll_interval,
+ )
+ .await?;
+ (
+ recovered.mux_stm,
+ Some((
+ recovered.journal,
+ recovered.snapshot,
+ recovered.last_applied_op,
+ )),
+ )
+ }
+ MetadataHandoff::Waiter { bundle_rx } => {
+ let bundle = await_metadata_bundle(
+ shard_id,
+ &bundle_rx,
+ &shutdown_flag_for_handoff,
+ poll_interval,
+ )
+ .await?;
+ (ServerNgMuxStateMachine::from_factory_bundle(bundle), None)
+ }
+ };
+
+ // Metadata consensus + journal + snapshot live only on shard 0.
+ // `IggyShard::tick_metadata` short-circuits when `consensus.is_none()`,
+ // so peer shards have no caller that reads `journal` or `snapshot`.
+ let (metadata_consensus, journal_for_metadata, snapshot_for_metadata) =
+ if let Some((journal, snapshot, last_applied_op)) = owner_state {
+ let restored_op = last_applied_op
+ .unwrap_or_else(|| snapshot.as_ref().map_or(0, IggySnapshot::sequence_number));
+ let consensus = restore_metadata_consensus(
+ &journal,
+ restored_op,
+ topology.self_replica_id,
+ topology.replica_count,
+ Rc::clone(&bus),
+ );
+ (Some(consensus), Some(journal), snapshot)
+ } else {
+ (None, None, None)
+ };
let metadata = ServerNgMetadata::new(
- Some(restore_metadata_consensus(
- &recovered.journal,
- restored_op,
- topology.self_replica_id,
- topology.replica_count,
- Rc::clone(&bus),
- )),
- Some(recovered.journal),
- recovered.snapshot,
- recovered.mux_stm,
+ metadata_consensus,
+ journal_for_metadata,
+ snapshot_for_metadata,
+ mux_stm,
Some(PathBuf::from(&config.system.path)),
);
- let shard = build_single_shard(config, &topology, metadata, bus).await?;
- info!(shard = shard.id, "server-ng bootstrap complete");
- Ok(shard)
-}
+ let shard_metrics = ShardMetrics::for_shard();
+ let shard = build_shard_for_thread(
+ shard_id,
+ total_shards,
+ config,
+ &topology,
+ metadata,
+ Rc::clone(&bus),
+ senders,
+ inbox,
+ shard_metrics,
+ )
+ .await?;
-fn restore_metadata_consensus(
- journal: &PrepareJournal,
- restored_op: u64,
- self_replica_id: u8,
- replica_count: u8,
- bus: Rc<IggyMessageBus>,
-) -> VsrConsensus<Rc<IggyMessageBus>> {
- let mut consensus = VsrConsensus::new(
- CLUSTER_ID,
- self_replica_id,
- replica_count,
- 0,
- bus,
- LocalPipeline::new(),
+ info!(
+ shard = shard_id,
+ partitions = shard.plane.partitions().len(),
+ "server-ng shard initialized"
);
- let last_header = journal
- .last_op()
- .and_then(|op| usize::try_from(op).ok())
- .and_then(|op| journal.header(op).map(|header| *header));
- if let Some(header) = last_header {
- consensus.set_view(header.view);
+ // Re-check the cross-thread shutdown flag here, *before* spawning the
+ // message pump. A sibling shard may have failed in the window between
+ // the metadata broadcast and this point; gating before spawn keeps the
+ // bus' `background_tasks` vec empty on the shutdown path. Spawn-then-
+ // check would leave `bus.track_background(pump_handle)` registering a
+ // `JoinHandle` that only `bus.shutdown()` drains, but the watchdog
+ // driving `bus.shutdown()` is `.detach()`'d (see TODO at
+ // `spawn_shutdown_watchdog`) and may not be scheduled before this
+ // function returns `Ok(())` and the compio runtime drops, cancelling
+ // the pump mid-`write_vectored_all`.
+ //
+ // Without this gate shard 0 would also still open TCP/QUIC/WS
+ // listeners for a server that is already tearing down, briefly
+ // accepting connections that immediately get torn by the watchdog.
+ if shutdown_flag_for_handoff.load(Ordering::Relaxed) {
+ return Ok(());
}
- consensus.init();
- consensus.sequencer().set_sequence(restored_op);
- // Known gap: clustered bootstrap does not yet persist a durable commit
- // watermark. Until that exists, recovery assumes the not-yet-supported
- // case where replicas are not rejoining with divergent commit state.
- consensus.restore_commit_state(restored_op, restored_op);
- if let Some(header) = last_header {
- consensus.set_last_prepare_checksum(header.checksum);
+ let (stop_tx, stop_rx) = channel(1);
+ let pump_shard = Rc::clone(&shard);
+ let pump_handle = compio::runtime::spawn(async move {
+ pump_shard.run_message_pump(stop_rx).await;
+ });
+ bus.track_background(pump_handle);
+
+ // Listeners (replica + every client transport) bind on shard 0 only.
+ // Shard 0's coordinator round-robins inbound TCP/WS connections to
+ // peer shards via fd-transfer. QUIC and TCP-TLS clients terminate
+ // locally on shard 0 (their per-connection state is non-portable -
+ // see `LifecycleFrame::ClientWsConnectionSetup` rustdoc).
+ //
+ // No phase-2 listener fence is needed: peer shards no longer scan
+ // the WAL, so a shard-0 append accepted mid-boot cannot race a
+ // peer's `truncate_or_fail`. The factory-bundle handoff has already
+ // installed reader-mode `MuxStateMachine`s on every peer by the
+ // time shard 0 returns from `broadcast_metadata_bundle`.
+ if shard_id == 0 {
+ let coord = shard
+ .coordinator()
+ .expect("shard 0 always has a coordinator attached by the builder");
+ let on_client_request = make_client_request_handler(&shard);
+ let accepted_replica = make_delegating_replica_accept_fn(Rc::clone(&coord));
+ let accepted_client = make_shard_zero_client_accept_fns(coord, &bus, on_client_request);
+
+ if let Err(error) =
+ start_tcp_runtime(&shard, config, &topology, accepted_replica, accepted_client).await
+ {
+ let _ = stop_tx.try_send(());
+ return Err(error);
+ }
}
- consensus
+ bus.token().wait().await;
+ let _ = stop_tx.try_send(());
+
+ info!(shard = shard_id, "server-ng shard exited cleanly");
+ Ok(())
+}
+
+/// Block until shard 0 broadcasts the metadata factory bundle, or the
+/// cross-thread shutdown flag flips. Polled in a `poll_interval` loop
+/// so a shard 0 that panics before it broadcasts cannot strand peer
+/// shards: the shutdown path flips the flag, every waiter observes it
+/// on the next tick, and the server tears down instead of hanging.
+///
+/// Uses `try_recv` + sleep rather than `timeout(recv())`. Crossfire 3.x
+/// documents `recv()` as cancellation-safe (no leak/deadlock) but does
+/// not guarantee atomicity for the dropped future's result; `try_recv`
+/// keeps each tick fully synchronous and side-effect-free, so the
+/// shutdown poll cadence cannot ambiguously consume a bundle.
+async fn await_metadata_bundle(
+ shard_id: u16,
+ bundle_rx: &crossfire::MAsyncRx<crossfire::mpmc::Array<ServerNgMetadataBundle>>,
+ shutdown_flag: &Arc<AtomicBool>,
+ poll_interval: Duration,
+) -> Result<ServerNgMetadataBundle, ServerNgError> {
+ loop {
+ match bundle_rx.try_recv() {
+ Ok(bundle) => return Ok(bundle),
+ Err(crossfire::TryRecvError::Disconnected) => {
+ return Err(ServerNgError::MetadataHandoffAborted { shard_id });
+ }
+ Err(crossfire::TryRecvError::Empty) => {
+ if shutdown_flag.load(Ordering::Relaxed) {
+ return Err(ServerNgError::MetadataHandoffAborted { shard_id });
+ }
+ compio::time::sleep(poll_interval).await;
+ }
+ }
+ }
}
-async fn build_single_shard(
+/// Push `peers` cloned bundles onto `bundle_tx`, polling each send in a
+/// `poll_interval` loop so the cross-thread shutdown flag can interrupt
+/// a stalled handoff. Symmetric to [`await_metadata_bundle`]: shutdown
+/// observed mid-handshake aborts cleanly rather than stalling on a
+/// `send` future that can no longer make progress.
+///
+/// Uses `try_send` + sleep rather than `timeout(send())`. Crossfire 3.x
+/// documents `send()` as cancellation-safe in the leak/deadlock sense
+/// but explicitly warns the true result is unknown when `SendFuture` is
+/// dropped on cancellation. For a retry loop that re-clones on every
+/// tick that would risk publishing the same bundle twice, stuffing the
+/// bounded channel past `peers` and stranding a follow-up `send`.
+/// `try_send` returns the bundle back inside `TrySendError::Full`, so
+/// the loop reuses it instead of re-cloning when the channel is full.
+async fn broadcast_metadata_bundle(
+ shard_id: u16,
+ bundle_tx: &crossfire::MAsyncTx<crossfire::mpmc::Array<ServerNgMetadataBundle>>,
+ bundle: ServerNgMetadataBundle,
+ peers: u16,
+ shutdown_flag: &Arc<AtomicBool>,
+ poll_interval: Duration,
+) -> Result<(), ServerNgError> {
+ for _ in 0..peers {
+ let mut pending = bundle.clone();
+ loop {
+ match bundle_tx.try_send(pending) {
+ Ok(()) => break,
+ Err(crossfire::TrySendError::Disconnected(_)) => {
+ // Every peer dropped its `bundle_rx` before recv. Shard
+ // 0 must not silently continue past handoff: it would
+ // bind listeners and commit consensus state for a
+ // cluster whose peers are gone. Propagate the abort so
+ // `shard_main` short-circuits before further side
+ // effects; `shutdown_flag` will flip via the normal
+ // teardown path.
+ return Err(ServerNgError::MetadataHandoffAborted { shard_id });
+ }
+ Err(crossfire::TrySendError::Full(returned)) => {
+ if shutdown_flag.load(Ordering::Relaxed) {
+ return Err(ServerNgError::MetadataHandoffAborted { shard_id });
+ }
+ pending = returned;
+ compio::time::sleep(poll_interval).await;
+ }
+ }
+ }
+ }
+ Ok(())
+}
+
+/// Spawn a per-shard polling task that watches the cross-thread shutdown
+/// flag and triggers this shard's bus shutdown on transition. The flag
+/// is the only Send signal we have; the bus' shutdown machinery is
+/// `!Send` (`Rc<Cell<bool>>` + per-shard `async_channel`), so it must be
+/// triggered from within the runtime that owns the bus.
+#[allow(clippy::needless_pass_by_value)]
+fn spawn_shutdown_watchdog(
+ bus: Rc<IggyMessageBus>,
+ shutdown_flag: Arc<AtomicBool>,
+ drain_timeout: Duration,
+ poll_interval: Duration,
+) {
+ let bus_for_task = Rc::clone(&bus);
+ let bus_token = bus.token();
+ let watchdog = compio::runtime::spawn(async move {
+ loop {
+ if shutdown_flag.load(Ordering::Relaxed) {
+ break;
+ }
+ if bus_token.is_triggered() {
+ // Bus shutdown was driven from elsewhere (e.g. internal
+ // failure path). Watchdog has nothing left to do.
+ return;
+ }
+ compio::time::sleep(poll_interval).await;
+ }
+ let _ = bus_for_task.shutdown(drain_timeout).await;
+ });
+ // TODO(hubcio): `.detach()` races bus shutdown: when `bus.token()` is
+ // triggered, `shard_main` returns and the runtime drops the watchdog
+ // mid-`bus.shutdown()`, truncating in-flight `ClientForwardFailed`
+ // replies (terminal per `SendError` docs). Cannot use
+ // `bus.track_background(watchdog)` here because the watchdog itself
+ // drives `bus.shutdown()`, and the bg-drain loop in `shutdown()`
+ // would re-enter awaiting the watchdog's own pending shutdown call
+ // (self-deadlock). Fix: extract a `core/task_registry` crate mirroring
+ // `core/server`'s task-tracking mechanism, share it between the bus
+ // and server-ng so background tasks can be reaped without coupling
+ // to the bus shutdown order.
+ watchdog.detach();
+}
+
+#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
+async fn build_shard_for_thread(
+ shard_id: u16,
+ total_shards: u16,
config: &ServerNgConfig,
topology: &TcpTopology,
metadata: ServerNgMetadata,
bus: Rc<IggyMessageBus>,
+ senders: Vec<TaggedSender>,
+ inbox: ShardReceiver<ShardFrame>,
+ metrics: ShardMetrics,
) -> Result<Rc<ServerNgShard>, ServerNgError> {
- let shard_id = ShardId::new(SHARD_ID);
- let partition_count = metadata.mux_stm.streams().read(|inner| {
+ let shard_local_id = ShardId::new(shard_id);
+ let total_partitions = metadata.mux_stm.streams().read(|inner| {
inner
.items
.iter()
@@ -301,10 +985,20 @@ async fn build_single_shard(
.map(|(_, topic)| topic.partitions.len())
.sum::<usize>()
})
- .sum()
+ .sum::<usize>()
});
+
+ // IggyPartitions holds only the partitions owned by this shard
+ // (see the filter below at insert time), so the server-wide total
+ // is an N-fold overshoot. `ceil(total / shards) * 2` is a coarse
+ // upper bound that absorbs hash skew without paying the full
+ // multiplier. PapayaShardsTable below stays sized to the server-wide
+ // total because every shard routes every namespace.
+ let owned_partitions_capacity = total_partitions
+ .div_ceil(usize::from(total_shards).max(1))
+ .saturating_mul(2);
let mut partitions = IggyPartitions::with_capacity(
- shard_id,
+ shard_local_id,
PartitionsConfig {
messages_required_to_save: config.system.partition.messages_required_to_save,
size_of_messages_required_to_save: config
@@ -314,29 +1008,45 @@ async fn build_single_shard(
enforce_fsync: config.system.partition.enforce_fsync,
segment_size: config.system.segment.size,
},
- partition_count,
+ owned_partitions_capacity,
);
- let shards_table = PapayaShardsTable::with_capacity(partition_count);
-
- let (topic_stats, namespaces) = metadata.mux_stm.streams().read(|inner| {
- let mut topic_stats = Vec::new();
- let mut namespaces = Vec::with_capacity(partition_count);
+ let shards_table = PapayaShardsTable::with_capacity(total_partitions);
+
+ // Stream-filter inside the `read()` closure: only partitions owned by
+ // this shard need the heavy (`Arc<TopicStats>` + `Partition`) clones
+ // for the async `load_partition` below. Non-owning entries are pushed
+ // straight into `shards_table` here, so no Vec scales with the
+ // server-wide partition count.
+ let owned = metadata.mux_stm.streams().read(|inner| {
+ let mut owned = Vec::with_capacity(owned_partitions_capacity);
for (_, stream) in &inner.items {
for (topic_id, topic) in &stream.topics {
- topic_stats.push(topic.stats.clone());
for partition in &topic.partitions {
- namespaces.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ let namespace = IggyNamespace::new(stream.id, topic_id, partition.id);
+ let owning_shard =
+ calculate_shard_assignment(&namespace, u32::from(total_shards));
+ if owning_shard == shard_id {
+ owned.push((stream.id, topic_id, topic.stats.clone(), partition.clone()));
+ } else {
+ // Non-owning entry: only `shard_id` is consulted
+ // by the router; `local_idx` is meaningful only on
+ // the owning shard, so a sentinel `0` is safe.
+ shards_table.insert(
+ namespace,
+ PartitionLocation::new(ShardId::new(owning_shard), LocalIdx::new(0)),
+ );
+ }
}
}
}
- (topic_stats, namespaces)
+ owned
});
- for topic_stats in topic_stats {
- topic_stats.zero_out_all();
- }
-
- for (stream_id, topic_id, topic_stats, partition_metadata) in namespaces {
+ // Snapshot totals were zeroed once on shard 0 before the factory
+ // bundle was broadcast (see `MetadataHandoff::Owner`). All shards
+ // here only add their per-partition deltas, so the shared
+ // `Arc<TopicStats>` atomics race only against other atomic adds.
+ for (stream_id, topic_id, topic_stats, partition_metadata) in owned {
validate_recovered_namespace(config, stream_id, topic_id, partition_metadata.id)?;
let namespace = IggyNamespace::new(stream_id, topic_id, partition_metadata.id);
let partition = load_partition(
@@ -352,17 +1062,16 @@ async fn build_single_shard(
let local_idx = partitions.insert(namespace, partition);
shards_table.insert(
namespace,
- PartitionLocation::new(shard_id, LocalIdx::new(*local_idx)),
+ PartitionLocation::new(ShardId::new(shard_id), LocalIdx::new(*local_idx)),
);
}
- let (sender, inbox) = shard_channel::<()>(SHARD_ID, SHARD_INBOX_CAPACITY);
- let senders = vec![sender];
let shard_handle = Rc::new(RefCell::new(None));
let on_replica_message = make_deferred_replica_message_handler(&shard_handle);
let on_client_request = make_deferred_client_request_handler(&shard_handle);
+ let shard_name = format!("server-ng-shard-{shard_id}");
let built = IggyShardBuilder::new(
- ShardIdentity::new(SHARD_ID, SHARD_NAME.to_string()),
+ ShardIdentity::new(shard_id, shard_name),
Rc::clone(&bus),
on_replica_message,
on_client_request,
@@ -373,18 +1082,75 @@ async fn build_single_shard(
shards_table,
PartitionConsensusConfig::new(CLUSTER_ID, topology.replica_count, Rc::clone(&bus)),
CoordinatorConfig::default(),
- bus.token(),
+ metrics,
)
- .build();
- if let Some(refresh_task) = built.refresh_task {
- bus.track_background(refresh_task);
- }
+ .build()
+ .map_err(ServerNgError::ShardConstruction)?;
let shard = Rc::new(built.shard);
*shard_handle.borrow_mut() = Some(Rc::downgrade(&shard));
Ok(shard)
}
+fn restore_metadata_consensus(
+ journal: &PrepareJournal,
+ restored_op: u64,
+ self_replica_id: u8,
+ replica_count: u8,
+ bus: Rc<IggyMessageBus>,
+) -> VsrConsensus<Rc<IggyMessageBus>> {
+ let mut consensus = VsrConsensus::new(
+ CLUSTER_ID,
+ self_replica_id,
+ replica_count,
+ iggy_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ bus,
+ LocalPipeline::new(),
+ );
+
+ let last_header = journal
+ .last_op()
+ .and_then(|op| usize::try_from(op).ok())
+ .and_then(|op| journal.header(op).map(|header| *header));
+ if let Some(header) = last_header {
+ consensus.set_view(header.view);
+ }
+
+ consensus.init();
+ consensus.sequencer().set_sequence(restored_op);
+ // TODO(hubcio): clustered bootstrap does not persist a durable
+ // (view, commit_op) watermark, so we collapse commit_min/commit_max to
+ // `restored_op` (= last journaled op). VSR consequence: on a view
+ // change after partial recovery, a replica that came back with a
+ // commit_max below the cluster's true commit_min will accept stale
+ // prepares as new and overwrite already-committed log entries
+ // (split-brain on the committed prefix).
+ //
+ // Fix direction: persist (view, commit_op) on the journal-header path
+ // (`core/journal/src/prepare_journal.rs` PrepareHeader already carries
+ // `view`; extend with `commit_op` or add a sidecar watermark file
+ // updated on every commit), seed `restore_commit_state(min, max)`
+ // from durable state on recovery, and refuse boot if the gap exceeds
+ // a configurable threshold. Tracked under the "durable
+ // PartitionJournal + durable (view, commit_op) watermark" milestone
+ // named in the multi-shard wiring commit body.
+ //
+ // Reproducible in `core/simulator` once it grows a restart-from-disk
+ // path: today `SimNetwork::enable_process` only un-disables links
+ // without replaying `SimJournal` + `SimSnapshot` through
+ // `restore_metadata_consensus`, so the flatten cannot trip. Add a
+ // crash-restart-replay primitive in the sim, then write a scenario
+ // that commits an op on the primary, crashes the primary mid-
+ // replicate-ack, triggers a view change, and asserts the recovered
+ // replica refuses to re-accept the committed prepare.
+ consensus.restore_commit_state(restored_op, restored_op);
+ if let Some(header) = last_header {
+ consensus.set_last_prepare_checksum(header.checksum);
+ }
+
+ consensus
+}
+
const fn validate_recovered_namespace(
config: &ServerNgConfig,
stream_id: usize,
@@ -861,16 +1627,20 @@ fn resolve_tcp_topology(
let default_quic_addr =
resolve_optional_listener_addr(config.quic.enabled, "quic.address", &config.quic.address)?;
if !config.cluster.enabled {
- if let Some(replica_id) = current_replica_id {
- warn!(
- replica_id,
- "cluster is disabled, ignoring --replica-id for single-node server-ng startup"
- );
+ if let Some(replica_id) = current_replica_id
+ && replica_id != SHARD_REPLICA_ID
+ {
+ return Err(ServerNgError::ReplicaIdRequiresCluster {
+ supplied: replica_id,
+ default: SHARD_REPLICA_ID,
+ });
}
return Ok(TcpTopology {
// Keep parity with the current server binary and the integration
- // harness: `--replica-id` may be passed unconditionally, but in
- // single-node mode there is only replica 0.
+ // harness: `--replica-id 0` may be passed unconditionally in
+ // single-node mode; any other id is rejected above so the WAL
+ // cannot commit under an identity that will later disagree with
+ // a cluster.nodes[] entry.
self_replica_id: SHARD_REPLICA_ID,
replica_count: 1,
client_listen_addr: default_client_addr,
@@ -1209,13 +1979,6 @@ async fn start_manual_runtime(
Ok(())
}
-fn make_replica_message_handler(shard: &Rc<ServerNgShard>) -> MessageHandler {
- let shard = Rc::clone(shard);
- Rc::new(move |_replica_id, message| {
- shard.dispatch(message);
- })
-}
-
fn make_client_request_handler(shard: &Rc<ServerNgShard>) -> RequestHandler {
let shard = Rc::clone(shard);
Rc::new(move |client_id, message| {
@@ -1271,87 +2034,73 @@ fn upgrade_shard_handle(shard_handle: &ServerNgShardHandle) -> Option<Rc<ServerN
.and_then(std::rc::Weak::upgrade)
}
-fn make_local_replica_accept_fn(
- bus: &Rc<IggyMessageBus>,
- on_message: MessageHandler,
+/// Replica accept callback that ships every inbound connection through
+/// the shard-0 coordinator's round-robin fd-delegation. The fd lands on
+/// the target shard's inbox as a [`shard::LifecycleFrame::ReplicaConnectionSetup`]
+/// frame and is installed on that shard's bus.
+fn make_delegating_replica_accept_fn(
+ coord: Rc<shard::coordinator::ShardZeroCoordinator>,
) -> AcceptedReplicaFn {
- let bus = Rc::clone(bus);
- Rc::new(move |stream, peer_id| {
- installer::install_replica_tcp(&bus, peer_id, stream, on_message.clone());
- })
+ Rc::new(
+ move |stream, peer_id| match coord.delegate_replica(stream, peer_id) {
+ Ok(target) => {
+ info!(peer_id, target, "replica connection delegated");
+ }
+ Err(error) => {
+ warn!(
+ peer_id,
+ error = ?error,
+ "delegate_replica failed; dropping inbound replica connection"
+ );
+ }
+ },
+ )
}
-fn make_local_client_accept_fns(
+/// Shard-0 client accept callbacks. TCP and WS clients are delegated via
+/// the coordinator (round-robin to peer shards); QUIC and TCP-TLS install
+/// locally on shard 0 because their per-connection state is not portable
+/// across shards (`compio_quic` endpoint binds one UDP socket; rustls TLS
+/// state ties to the post-handshake reactor).
+fn make_shard_zero_client_accept_fns(
+ coord: Rc<shard::coordinator::ShardZeroCoordinator>,
bus: &Rc<IggyMessageBus>,
on_request: RequestHandler,
) -> LocalClientAcceptFns {
- let tcp_bus = Rc::clone(bus);
- let ws_bus = Rc::clone(bus);
let quic_bus = Rc::clone(bus);
let tcp_tls_bus = Rc::clone(bus);
- let tcp_request = on_request.clone();
- let ws_request = on_request.clone();
let quic_request = on_request.clone();
let tcp_tls_request = on_request;
- let counter = Rc::new(Cell::new(1_u128));
- let shard_id = u128::from(bus.shard_id());
-
- let tcp_counter = Rc::clone(&counter);
- let tcp = Rc::new(move |stream| {
- let Some(meta) = client_meta_from_stream(
- &stream,
- tcp_counter.as_ref(),
- shard_id,
- ClientTransportKind::Tcp,
- ) else {
- return;
- };
- installer::install_client_tcp(&tcp_bus, meta, stream, tcp_request.clone());
+
+ let tcp_coord = Rc::clone(&coord);
+ let tcp = Rc::new(move |stream| match tcp_coord.delegate_client(stream) {
+ Ok(client_id) => info!(client_id, "TCP client delegated"),
+ Err(error) => warn!(error = ?error, "delegate_client failed; dropping TCP client"),
});
- let ws_counter = Rc::clone(&counter);
- let ws = Rc::new(move |stream| {
- let Some(meta) = client_meta_from_stream(
- &stream,
- ws_counter.as_ref(),
- shard_id,
- ClientTransportKind::Ws,
- ) else {
- return;
- };
- let fd = match fd_transfer::dup_fd(&stream) {
- Ok(fd) => fd,
- Err(error) => {
- warn!(
- client_id = meta.client_id,
- error = %error,
- "dropping accepted websocket client after fd duplication failure"
- );
- return;
- }
- };
- ws_bus.install_client_ws_fd(fd, meta, ws_request.clone());
+ let ws_coord = Rc::clone(&coord);
+ let ws = Rc::new(move |stream| match ws_coord.delegate_ws_client(stream) {
+ Ok(client_id) => info!(client_id, "WS client delegated"),
+ Err(error) => warn!(error = ?error, "delegate_ws_client failed; dropping WS client"),
});
- let quic_counter = Rc::clone(&counter);
+ // QUIC and TCP-TLS terminate locally on shard 0 but mint their client
+ // ids through the coordinator's `client_seq`, the same counter the
+ // delegated TCP/WS path uses. A separate counter here would let a
+ // shard-0-local id collide with a delegated id that round-robined to
+ // shard 0 (both encode target shard 0) in shard 0's connection
+ // registry.
+ let quic_coord = Rc::clone(&coord);
let quic = Rc::new(move |accepted: message_bus::AcceptedQuicConn| {
- let meta = mint_client_meta(
- quic_counter.as_ref(),
- shard_id,
- accepted.peer_addr(),
- ClientTransportKind::Quic,
- );
+ let meta = mint_client_meta(&quic_coord, accepted.peer_addr(), ClientTransportKind::Quic);
installer::install_client_quic(&quic_bus, meta, accepted, quic_request.clone());
});
- let tcp_tls_counter = Rc::clone(&counter);
+ let tcp_tls_coord = coord;
let tcp_tls = Rc::new(move |stream, tls_config| {
- let Some(meta) = client_meta_from_stream(
- &stream,
- tcp_tls_counter.as_ref(),
- shard_id,
- ClientTransportKind::TcpTls,
- ) else {
+ let Some(meta) =
+ client_meta_from_stream(&stream, &tcp_tls_coord, ClientTransportKind::TcpTls)
+ else {
return;
};
installer::install_client_tcp_tls(
@@ -1373,34 +2122,25 @@ fn make_local_client_accept_fns(
fn client_meta_from_stream(
stream: &compio::net::TcpStream,
- counter: &Cell<u128>,
- shard_id: u128,
+ coord: &shard::coordinator::ShardZeroCoordinator,
transport: ClientTransportKind,
) -> Option<ClientConnMeta> {
let peer_addr = match stream.peer_addr() {
Ok(peer_addr) => peer_addr,
Err(error) => {
- let client_id = preview_client_id(counter, shard_id);
- warn!(client_id, error = %error, "dropping accepted client with unknown peer address");
+ warn!(error = %error, "dropping accepted client with unknown peer address");
return None;
}
};
- Some(mint_client_meta(counter, shard_id, peer_addr, transport))
-}
-
-const fn preview_client_id(counter: &Cell<u128>, shard_id: u128) -> u128 {
- (shard_id << 112) | counter.get()
+ Some(mint_client_meta(coord, peer_addr, transport))
}
fn mint_client_meta(
- counter: &Cell<u128>,
- shard_id: u128,
+ coord: &shard::coordinator::ShardZeroCoordinator,
peer_addr: SocketAddr,
transport: ClientTransportKind,
) -> ClientConnMeta {
- let seq = counter.get();
- counter.set(seq.wrapping_add(1));
- ClientConnMeta::new((shard_id << 112) | seq, peer_addr, transport)
+ ClientConnMeta::new(coord.mint_shard_zero_client_id(), peer_addr, transport)
}
async fn start_client_listeners(
@@ -1571,3 +2311,145 @@ fn socket_addr_from_parts(
})?;
Ok(SocketAddr::new(ip, port))
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn shutdown_on_drop_armed_flips_flag() {
+ let flag = Arc::new(AtomicBool::new(false));
+ drop(ShutdownOnDrop::new(Arc::clone(&flag)));
+ assert!(
+ flag.load(Ordering::Relaxed),
+ "an armed guard must flip the flag on drop (covers the error `?` \
+ and panic-unwind exit paths of run_shard_thread)"
+ );
+ }
+
+ #[test]
+ fn shutdown_on_drop_disarmed_leaves_flag() {
+ let flag = Arc::new(AtomicBool::new(false));
+ let mut guard = ShutdownOnDrop::new(Arc::clone(&flag));
+ guard.disarm();
+ drop(guard);
+ assert!(
+ !flag.load(Ordering::Relaxed),
+ "a disarmed guard must not flip the flag (clean `Ok(())` exit)"
+ );
+ }
+
+ const TEST_POLL_INTERVAL: Duration = Duration::from_millis(50);
+
+ #[compio::test]
+ async fn broadcast_metadata_bundle_returns_immediately_with_no_peers() {
+ // Single-shard deployment: shard 0 has no peers to fan out to,
+ // so the handoff must complete without ever calling `send`.
+ let (bundle_tx, _bundle_rx) = crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(0);
+ let flag = Arc::new(AtomicBool::new(false));
+ let mux = ServerNgMuxStateMachine::default();
+ broadcast_metadata_bundle(
+ 0,
+ &bundle_tx,
+ mux.factory_bundle(),
+ 0,
+ &flag,
+ TEST_POLL_INTERVAL,
+ )
+ .await
+ .expect("zero peers must not block shard 0");
+ }
+
+ #[compio::test]
+ async fn metadata_bundle_round_trips_through_channel() {
+ // End-to-end: shard 0 mints a bundle, a peer receives it on
+ // another runtime, and `from_factory_bundle` constructs a
+ // reader-mode mux that observes shard 0's writes via the same
+ // LeftRight pair.
+ let peers = 1u16;
+ let (bundle_tx, bundle_rx) =
+ crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(usize::from(peers));
+ let flag = Arc::new(AtomicBool::new(false));
+
+ let owner = ServerNgMuxStateMachine::default();
+ let bundle = owner.factory_bundle();
+ broadcast_metadata_bundle(0, &bundle_tx, bundle, peers, &flag, TEST_POLL_INTERVAL)
+ .await
+ .expect("broadcast must succeed with one peer drained");
+
+ let received = await_metadata_bundle(1, &bundle_rx, &flag, TEST_POLL_INTERVAL)
+ .await
+ .expect("peer must receive the broadcast bundle");
+ let _peer_mux = ServerNgMuxStateMachine::from_factory_bundle(received);
+ }
+
+ #[compio::test]
+ async fn broadcast_metadata_bundle_aborts_when_peers_drop_rx() {
+ // Shard 0 drives handoff but every peer's `bundle_rx` was dropped
+ // before recv. Silently returning Ok would commit listener binds
+ // and consensus init for a cluster whose peers are gone; the
+ // broadcast must surface the disconnect so `shard_main` aborts.
+ let (bundle_tx, bundle_rx) = crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(0);
+ drop(bundle_rx);
+ let flag = Arc::new(AtomicBool::new(false));
+ let mux = ServerNgMuxStateMachine::default();
+
+ let err = broadcast_metadata_bundle(
+ 0,
+ &bundle_tx,
+ mux.factory_bundle(),
+ 3,
+ &flag,
+ TEST_POLL_INTERVAL,
+ )
+ .await
+ .expect_err("dropped rx must surface as MetadataHandoffAborted");
+ assert!(
+ matches!(err, ServerNgError::MetadataHandoffAborted { shard_id: 0 }),
+ "expected MetadataHandoffAborted, got {err:?}"
+ );
+ }
+
+ #[compio::test]
+ async fn await_metadata_bundle_aborts_when_owner_drops_without_sending() {
+ let (bundle_tx, bundle_rx) = crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(1);
+ let flag = Arc::new(AtomicBool::new(false));
+
+ // Shard 0 dies before broadcasting; the peer must observe the
+ // disconnect and abort instead of hanging forever.
+ drop(bundle_tx);
+
+ let err = await_metadata_bundle(1, &bundle_rx, &flag, TEST_POLL_INTERVAL)
+ .await
+ .expect_err("a peer whose owner never sends must abort");
+ assert!(
+ matches!(err, ServerNgError::MetadataHandoffAborted { shard_id: 1 }),
+ "expected MetadataHandoffAborted, got {err:?}"
+ );
+ }
+
+ #[compio::test]
+ async fn await_metadata_bundle_aborts_on_shutdown_flag() {
+ let (_bundle_tx, bundle_rx) = crossfire::mpmc::bounded_async::<ServerNgMetadataBundle>(1);
+ let flag = Arc::new(AtomicBool::new(false));
+
+ let waiter = compio::runtime::spawn({
+ let flag = Arc::clone(&flag);
+ async move { await_metadata_bundle(1, &bundle_rx, &flag, TEST_POLL_INTERVAL).await }
+ });
+
+ // Owner has not sent yet, but shutdown was requested; the peer
+ // must exit via the flag poll instead of hanging.
+ compio::time::sleep(TEST_POLL_INTERVAL / 2).await;
+ flag.store(true, Ordering::Relaxed);
+
+ let err = waiter
+ .await
+ .unwrap_or_else(|e| std::panic::resume_unwind(e))
+ .expect_err("shutdown flag must abort the bundle wait");
+ assert!(
+ matches!(err, ServerNgError::MetadataHandoffAborted { shard_id: 1 }),
+ "expected MetadataHandoffAborted on shutdown, got {err:?}"
+ );
+ }
+}
diff --git a/core/server-ng/src/main.rs b/core/server-ng/src/main.rs
index 169f32f65..0561b3001 100644
--- a/core/server-ng/src/main.rs
+++ b/core/server-ng/src/main.rs
@@ -23,50 +23,71 @@ mod args;
use args::Args;
use clap::Parser;
-use server_ng::bootstrap::RunServerNg;
+use server_ng::bootstrap::{bootstrap, load_config};
use server_ng::server_error::ServerNgError;
+use tracing::{error, info};
fn main() -> Result<(), ServerNgError> {
- // TODO: decouple runtime creation from the `server` crate and move the shared
- // compio executor setup into a lower-level crate/module used by both binaries.
- let runtime = match server::bootstrap::create_shard_executor() {
+ // TODO(hubcio): decouple runtime creation from the `server` crate and
+ // move the shared compio executor setup into a lower-level crate/module
+ // used by both binaries.
+ let bootstrap_runtime = match server::bootstrap::create_shard_executor() {
Ok(rt) => rt,
Err(e) => {
match e.kind() {
std::io::ErrorKind::InvalidInput => {
- // TODO: decouple io_uring diagnostics from the `server` crate.
server::diagnostics::print_invalid_io_uring_args_info();
}
std::io::ErrorKind::OutOfMemory => {
- // TODO: decouple io_uring diagnostics from the `server` crate.
server::diagnostics::print_locked_memory_limit_info();
}
std::io::ErrorKind::PermissionDenied => {
- // TODO: decouple io_uring diagnostics from the `server` crate.
server::diagnostics::print_io_uring_permission_info();
}
_ => {}
}
- panic!("Cannot create server-ng executor: {e}");
+ panic!("Cannot create server-ng bootstrap executor: {e}");
}
};
- runtime.block_on(async {
- let args = Args::parse();
- if let Ok(env_path) = std::env::var("IGGY_ENV_PATH") {
- let _ = dotenvy::from_path(&env_path);
- } else {
- let _ = dotenvy::dotenv();
- }
- // TODO: decouple logging from the `server` crate and move the shared
- // logging bootstrap into a lower-level crate/module.
- let mut logging = server::log::logger::Logging::new();
- logging.early_init();
+ // Bootstrap on a temporary runtime: parse args, init logging, load
+ // config, init the memory pool. Then drop the runtime and spawn the
+ // per-shard runtimes - each shard thread builds its OWN
+ // `compio::runtime::Runtime` via `create_shard_executor`, pinned to
+ // its CPU.
+ let bootstrap_result: Result<(configs::server_ng::ServerNgConfig, Option<u8>), ServerNgError> =
+ bootstrap_runtime.block_on(async {
+ let args = Args::parse();
+ if let Ok(env_path) = std::env::var("IGGY_ENV_PATH") {
+ let _ = dotenvy::from_path(&env_path);
+ } else {
+ let _ = dotenvy::dotenv();
+ }
+
+ // TODO: decouple logging from the `server` crate.
+ let mut logging = server::log::logger::Logging::new();
+ logging.early_init();
+
+ let config = load_config(&mut logging).await?;
+ iggy_common::MemoryPool::init_pool(&config.system.memory_pool.into_other());
+
+ Ok((config, args.replica_id))
+ });
+ let (config, replica_id) = bootstrap_result?;
+ drop(bootstrap_runtime);
- let config = server_ng::bootstrap::load_config(&mut logging).await?;
- iggy_common::MemoryPool::init_pool(&config.system.memory_pool.into_other());
+ let shards = bootstrap(config, replica_id)?;
+ if let Err(error) = shards.install_ctrlc_handler() {
+ // Without a working SIGINT handler the server has no way to
+ // observe an operator Ctrl-C and the shutdown flag would never
+ // flip, leaving shard threads parked indefinitely. Fail fast
+ // rather than boot into an un-killable state.
+ error!(error = %error, "failed to install Ctrl-C handler; aborting boot");
+ std::process::exit(1);
+ }
- let shard = server_ng::bootstrap::bootstrap(&config, args.replica_id).await?;
- shard.run(&config, args.replica_id).await
- })
+ info!("server-ng running; waiting on shard threads");
+ shards.join_all()?;
+ info!("server-ng shutdown complete");
+ Ok(())
}
diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs
index 77e17ee1c..6f24cb66a 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -20,14 +20,75 @@
use metadata::impls::recovery::RecoveryError;
// TODO: decouple logging errors from the `server` crate.
use server::server_error::LogError;
+use server::shard_allocator::ShardingError;
+use shard::ShardCtorError;
use thiserror::Error;
#[derive(Debug, Error)]
+#[non_exhaustive]
pub enum ServerNgError {
#[error(transparent)]
Iggy(Box<iggy_common::IggyError>),
#[error("failed to load server-ng config")]
Config(#[source] configs::ConfigurationError),
+ #[error("failed to allocate shards from sharding.cpu_allocation")]
+ ShardAllocator(#[source] ShardingError),
+ #[error("failed to bind shard {shard_id} to its CPU set")]
+ CpuAffinityFailed {
+ shard_id: u16,
+ #[source]
+ source: ShardingError,
+ },
+ #[error("failed to bind shard {shard_id} memory to its NUMA node")]
+ MemoryAffinityFailed {
+ shard_id: u16,
+ #[source]
+ source: ShardingError,
+ },
+ #[error("failed to spawn OS thread for shard {shard_id}")]
+ ShardSpawnFailed {
+ shard_id: u16,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to create io_uring runtime for shard {shard_id}")]
+ ShardRuntimeCreateFailed {
+ shard_id: u16,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error(
+ "shard allocator produced zero shards; server must run at least one \
+ shard (check [system.sharding] cpu_allocation)"
+ )]
+ ShardsCountZero,
+ #[error(
+ "computed shards_count = {count} exceeds the maximum of {} shards per \
+ server; shard ids must fit in u16 and stay below the OWNER_NONE \
+ sentinel",
+ message_bus::OWNER_NONE - 1
+ )]
+ ShardsCountOverflow { count: usize },
+ #[error("system.sharding.inbox_capacity must be in 1..={max}; got {value}")]
+ InvalidInboxCapacity { value: usize, max: usize },
+ #[error("system.sharding.shutdown_drain_timeout must be in (0, {max:?}]; got {value:?}")]
+ InvalidShutdownDrainTimeout {
+ value: std::time::Duration,
+ max: std::time::Duration,
+ },
+ #[error("system.sharding.shutdown_poll_interval must be in (0, {max:?}]; got {value:?}")]
+ InvalidShutdownPollInterval {
+ value: std::time::Duration,
+ max: std::time::Duration,
+ },
+ #[error(
+ "system.sharding.shutdown_poll_interval ({poll:?}) must be <= \
+ shutdown_drain_timeout ({drain:?})"
+ )]
+ ShutdownPollExceedsDrain {
+ poll: std::time::Duration,
+ drain: std::time::Duration,
+ },
#[error("failed to serialize current server-ng config")]
CurrentConfigSerialize(#[source] toml::ser::Error),
#[error("failed to write current server-ng config at {path}")]
@@ -40,6 +101,11 @@ pub enum ServerNgError {
Logging(#[source] LogError),
#[error("failed to recover metadata snapshot and journal")]
MetadataRecovery(#[source] RecoveryError),
+ #[error(
+ "shard {shard_id} aborted while waiting for shard-0 to broadcast the metadata \
+ factory bundle; shard 0 dropped its sender (most likely it failed to recover)"
+ )]
+ MetadataHandoffAborted { shard_id: u16 },
#[error("failed to parse {context} socket address '{address}'")]
SocketAddressParse {
context: &'static str,
@@ -53,6 +119,12 @@ pub enum ServerNgError {
ClusterReplicaCountTooLarge { count: usize },
#[error("cluster mode requires --replica-id to identify the current node")]
MissingReplicaId,
+ #[error(
+ "--replica-id {supplied} was passed with cluster.enabled=false; the WAL would commit \
+ under replica {default} which permanently fixes this node's identity. Either set \
+ cluster.enabled=true with a matching nodes[] entry, or drop --replica-id"
+ )]
+ ReplicaIdRequiresCluster { supplied: u8, default: u8 },
#[error("cluster node for replica {replica_id} is missing tcp_replica port")]
ClusterReplicaPortMissing { replica_id: u8 },
#[error(
@@ -96,6 +168,47 @@ pub enum ServerNgError {
#[source]
source: std::io::Error,
},
+ #[error("failed to construct IggyShard from bootstrap inputs")]
+ ShardConstruction(#[source] ShardCtorError),
+ #[error("{} shard thread(s) failed: {}", failures.len(), format_shard_failures(failures))]
+ ShardJoinFailures { failures: Vec<ShardJoinFailure> },
+}
+
+/// Per-shard outcome captured by [`crate::bootstrap::ShardHandles::join_all`]
+/// when a shard either returned `Err` or panicked.
+///
+/// Bundled into [`ServerNgError::ShardJoinFailures`] so the operator sees
+/// every failing shard rather than only the first one, which previously
+/// lived in the trace log alone.
+#[derive(Debug)]
+pub struct ShardJoinFailure {
+ pub shard_id: u16,
+ pub kind: ShardJoinFailureKind,
+}
+
+#[derive(Debug)]
+pub enum ShardJoinFailureKind {
+ Error(Box<ServerNgError>),
+ Panic { message: String },
+}
+
+fn format_shard_failures(failures: &[ShardJoinFailure]) -> String {
+ use std::fmt::Write as _;
+ let mut out = String::new();
+ for (idx, failure) in failures.iter().enumerate() {
+ if idx > 0 {
+ out.push_str("; ");
+ }
+ match &failure.kind {
+ ShardJoinFailureKind::Error(err) => {
+ let _ = write!(out, "shard {} -> {err}", failure.shard_id);
+ }
+ ShardJoinFailureKind::Panic { message } => {
+ let _ = write!(out, "shard {} panicked: {message}", failure.shard_id);
+ }
+ }
+ }
+ out
}
impl From<iggy_common::IggyError> for ServerNgError {
@@ -103,3 +216,37 @@ impl From<iggy_common::IggyError> for ServerNgError {
Self::Iggy(Box::new(source))
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn shard_join_failures_display_aggregates_all_entries() {
+ let failures = vec![
+ ShardJoinFailure {
+ shard_id: 0,
+ kind: ShardJoinFailureKind::Error(Box::new(ServerNgError::MissingReplicaId)),
+ },
+ ShardJoinFailure {
+ shard_id: 2,
+ kind: ShardJoinFailureKind::Panic {
+ message: "boom".to_string(),
+ },
+ },
+ ];
+ let rendered = ServerNgError::ShardJoinFailures { failures }.to_string();
+ assert!(
+ rendered.starts_with("2 shard thread(s) failed:"),
+ "expected count prefix, got {rendered}"
+ );
+ assert!(
+ rendered.contains("shard 0 ->"),
+ "shard 0 entry missing: {rendered}"
+ );
+ assert!(
+ rendered.contains("shard 2 panicked: boom"),
+ "shard 2 panic entry missing: {rendered}"
+ );
+ }
+}
diff --git a/core/shard/Cargo.toml b/core/shard/Cargo.toml
index 82bdb69f8..9a9a8b9f9 100644
--- a/core/shard/Cargo.toml
+++ b/core/shard/Cargo.toml
@@ -36,6 +36,8 @@ message_bus = { path = "../message_bus" }
metadata = { path = "../metadata" }
papaya = { workspace = true }
partitions = { path = "../partitions" }
+prometheus-client = { workspace = true }
+thiserror = { workspace = true }
tracing = { workspace = true }
[lints.clippy]
diff --git a/core/shard/src/builder.rs b/core/shard/src/builder.rs
index b365b9913..00cb92a9a 100644
--- a/core/shard/src/builder.rs
+++ b/core/shard/src/builder.rs
@@ -16,65 +16,49 @@
// under the License.
//! One-stop construction for an [`IggyShard`] plus, on shard 0 only, its
-//! associated [`ShardZeroCoordinator`] and periodic mapping-refresh task.
+//! associated [`ShardZeroCoordinator`].
//!
-//! The coordinator owns the authoritative `replica_id -> owning_shard`
-//! snapshot used by the router's `ConnectionLost` fast path and by the
-//! refresh task that re-broadcasts the snapshot to recover shards whose
-//! inbox was full when an earlier `ReplicaMappingUpdate` was sent.
-//! Both the `set_coordinator` call and the refresh-task spawn must happen
-//! together: wiring one without the other breaks the drop-recovery story
-//! the coordinator's rustdoc promises.
+//! The coordinator owns shard-0's round-robin state for replica and
+//! client placement. Cluster-wide `replica_id -> owning_shard` routing
+//! flows through the cluster-shared [`message_bus::ReplicaOwnerTable`],
+//! stamped by the owning shard's installer and CAS-cleared on disconnect.
//!
-//! Bootstrap code constructs an [`IggyShardBuilder`] per shard and calls
-//! [`IggyShardBuilder::build`]. On shard 0 the returned [`BuiltShard`]
-//! carries the refresh task's [`JoinHandle`]; bootstrap is responsible for
-//! tracking it on the bus' background-task set so graceful shutdown awaits
-//! it. On non-zero shards the handle is `None` and the coordinator field
-//! on the shard stays unset.
+//! The coordinator is constructed before the shard and passed to
+//! [`IggyShard::new`] as a ctor argument, so an `IggyShard` is fully
+//! wired the instant it becomes observable to any reader.
-use crate::coordinator::ShardZeroCoordinator;
+use crate::coordinator::{ShardZeroCoordinator, classify_try_send_err};
+use crate::metrics::{ShardMetrics, frame_drop_variant};
use crate::{
- CoordinatorConfig, IggyShard, PartitionConsensusConfig, Receiver, ShardFrame,
- ShardFramePayload, ShardIdentity, TaggedSender,
+ CoordinatorConfig, IggyShard, LifecycleFrame, PartitionConsensusConfig, Receiver,
+ ShardCtorError, ShardFrame, ShardIdentity, TaggedSender,
};
-use compio::runtime::JoinHandle;
use consensus::VsrConsensus;
use journal::JournalHandle;
-use message_bus::MessageBus;
use message_bus::client_listener::RequestHandler;
-use message_bus::lifecycle::ShutdownToken;
use message_bus::replica::listener::MessageHandler;
+use message_bus::{MessageBus, SendError};
use metadata::IggyMetadata;
use metadata::stm::StateMachine;
use partitions::IggyPartitions;
use std::rc::Rc;
-use tracing::warn;
use crate::shards_table::ShardsTable;
-/// A freshly constructed [`IggyShard`], paired with the refresh-task
-/// [`JoinHandle`] when this is shard 0. Bootstrap tracks the handle on
-/// the bus' background tasks so graceful shutdown awaits it.
-pub struct BuiltShard<B, MJ, S, M, T, R = ()>
+/// A freshly constructed [`IggyShard`].
+pub struct BuiltShard<B, MJ, S, M, T>
where
B: MessageBus,
- R: Send + 'static,
{
- pub shard: IggyShard<B, MJ, S, M, T, R>,
- /// `Some` on shard 0, `None` otherwise. Pass to
- /// `IggyMessageBus::track_background` (or an equivalent tracker) so
- /// the task is awaited on shutdown.
- pub refresh_task: Option<JoinHandle<()>>,
+ pub shard: IggyShard<B, MJ, S, M, T>,
}
/// Builder that pairs [`IggyShard`] construction with coordinator wiring
/// on shard 0. Non-zero shards skip the coordinator entirely; the
-/// `coord_config` and `shutdown_token` fields are then ignored.
-pub struct IggyShardBuilder<B, MJ, S, M, T, R = ()>
+/// `coord_config` field is then ignored.
+pub struct IggyShardBuilder<B, MJ, S, M, T>
where
B: MessageBus,
- R: Send + 'static,
{
identity: ShardIdentity,
bus: B,
@@ -82,26 +66,24 @@ where
on_client_request: RequestHandler,
metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>,
partitions: IggyPartitions<B>,
- senders: Vec<TaggedSender<R>>,
- inbox: Receiver<ShardFrame<R>>,
+ senders: Vec<TaggedSender>,
+ inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coord_config: CoordinatorConfig,
- shutdown_token: ShutdownToken,
+ metrics: ShardMetrics,
}
-impl<B, MJ, S, M, T, R> IggyShardBuilder<B, MJ, S, M, T, R>
+impl<B, MJ, S, M, T> IggyShardBuilder<B, MJ, S, M, T>
where
- B: MessageBus,
- R: Send + 'static,
+ B: MessageBus + Clone + 'static,
T: ShardsTable,
MJ: JournalHandle,
S: Send + 'static,
M: StateMachine,
{
/// Create a builder carrying every input needed by both
- /// [`IggyShard::new`] and (for shard 0) `ShardZeroCoordinator::new`
- /// plus `spawn_refresh_task`.
+ /// [`IggyShard::new`] and (for shard 0) `ShardZeroCoordinator::new`.
#[allow(clippy::too_many_arguments)]
pub fn new(
identity: ShardIdentity,
@@ -110,12 +92,12 @@ where
on_client_request: RequestHandler,
metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>,
partitions: IggyPartitions<B>,
- senders: Vec<TaggedSender<R>>,
- inbox: Receiver<ShardFrame<R>>,
+ senders: Vec<TaggedSender>,
+ inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coord_config: CoordinatorConfig,
- shutdown_token: ShutdownToken,
+ metrics: ShardMetrics,
) -> Self {
Self {
identity,
@@ -129,65 +111,99 @@ where
shards_table,
partition_consensus,
coord_config,
- shutdown_token,
+ metrics,
}
}
/// Consume the builder and produce a fully wired [`BuiltShard`]. On
- /// shard 0 this constructs the coordinator, attaches it to the shard,
- /// and spawns the periodic refresh task. On any other shard the
- /// coordinator-specific fields are dropped and `refresh_task` is
- /// `None`.
+ /// shard 0 this constructs the coordinator and passes it to the shard
+ /// ctor. On any other shard the coordinator-specific fields are
+ /// dropped.
///
- /// # Panics
+ /// # Errors
///
- /// Panics if `senders` is not in canonical order
- /// (`senders[i].shard_id() != i`) or if `senders.len()` does not
+ /// Returns [`ShardCtorError::SenderOrderingInvalid`] if `senders` is
+ /// not in canonical order (`senders[i].shard_id() != i`) and
+ /// [`ShardCtorError::ShardCountOverflow`] if `senders.len()` does not
/// fit in `u16`. Both are bootstrap programming errors and the
/// `u16` overflow check fires on every shard, not only shard 0.
- pub fn build(self) -> BuiltShard<B, MJ, S, M, T, R> {
+ pub fn build(self) -> Result<BuiltShard<B, MJ, S, M, T>, ShardCtorError> {
let is_shard_zero = self.identity.id == 0;
- let total_shards = u16::try_from(self.senders.len())
- .expect("cluster shard count must fit in u16 (bootstrap invariant)");
+ // Fail fast before installing forward closures: a misordered
+ // senders vec would silently misroute every cross-shard frame.
+ crate::validate_sender_ordering(&self.senders)?;
+ let total_shards =
+ u16::try_from(self.senders.len()).map_err(|_| ShardCtorError::ShardCountOverflow {
+ count: self.senders.len(),
+ })?;
- // Wire the bus' connection-lost notifier to push a ConnectionLost
- // frame into shard 0's inbox. Every shard's bus needs this so that
- // a dead replica connection anywhere surfaces at the coordinator.
- // Shard 0's sender is cloned once here; the closure captures that
- // clone and runs on the owning shard's reader/writer tasks.
- let shard_zero_sender = self.senders[0].sender().clone();
- let connection_lost_fn: message_bus::ConnectionLostFn = Rc::new(move |replica_id: u8| {
- let frame = ShardFrame::<R> {
- payload: ShardFramePayload::ConnectionLost { replica_id },
- response_sender: None,
- };
- if let Err(e) = shard_zero_sender.try_send(frame) {
- // Inbox full: the coordinator's periodic refresh task
- // re-broadcasts the authoritative snapshot and the bus
- // retries on next dial, so a single drop is recoverable.
- // Warn so operators see repeated drops if they occur.
- warn!(
- replica_id,
- "shard-0 inbox rejected ConnectionLost frame: {e}"
- );
- }
- });
- self.bus.set_connection_lost_fn(connection_lost_fn);
+ // Install the cross-shard forward closures. The bus' slow path
+ // (`send_to_replica` / `send_to_client`) calls these when the
+ // destination connection lives on a different shard; the closure
+ // wraps the message into a `LifecycleFrame::Forward{Replica,
+ // Client}Send` and `try_send`s it to the owning shard's inbox.
+ // The receiving shard's router unwraps the frame and re-enters
+ // the bus' fast path locally. The `senders` Vec is canonically
+ // ordered (`senders[i].shard_id() == i`).
+ let senders_for_replica: Rc<Vec<TaggedSender>> = Rc::new(self.senders.clone());
+ let senders_for_client = Rc::clone(&senders_for_replica);
+ let metrics_for_replica = self.metrics.clone();
+ let metrics_for_client = self.metrics.clone();
+ self.bus
+ .set_replica_forward_fn(Box::new(move |replica_id, owning_shard, msg| {
+ senders_for_replica.get(usize::from(owning_shard)).map_or(
+ Err(SendError::ReplicaForwardFailed(replica_id)),
+ |sender| {
+ let frame = ShardFrame::lifecycle(LifecycleFrame::ForwardReplicaSend {
+ replica_id,
+ msg,
+ });
+ sender.try_send(frame).map_err(|err| {
+ metrics_for_replica.record_frame_drop(
+ frame_drop_variant::FORWARD_REPLICA_SEND,
+ classify_try_send_err(&err),
+ );
+ SendError::ReplicaForwardFailed(replica_id)
+ })
+ },
+ )
+ }));
+ self.bus
+ .set_client_forward_fn(Box::new(move |client_id, owning_shard, msg| {
+ senders_for_client.get(usize::from(owning_shard)).map_or(
+ Err(SendError::ClientForwardFailed(client_id)),
+ |sender| {
+ let frame = ShardFrame::lifecycle(LifecycleFrame::ForwardClientSend {
+ client_id,
+ msg,
+ });
+ sender.try_send(frame).map_err(|err| {
+ metrics_for_client.record_frame_drop(
+ frame_drop_variant::FORWARD_CLIENT_SEND,
+ classify_try_send_err(&err),
+ );
+ SendError::ClientForwardFailed(client_id)
+ })
+ },
+ )
+ }));
- let coord_inputs = if is_shard_zero {
- let coord_senders: Vec<TaggedSender<R>> = self.senders.clone();
- Some((
+ // On shard 0: build the coordinator BEFORE the shard, so the
+ // coordinator can be passed to `IggyShard::new` as a ctor arg.
+ let coordinator = if is_shard_zero {
+ let coord_senders: Vec<TaggedSender> = self.senders.clone();
+ Some(Rc::new(ShardZeroCoordinator::new(
Rc::new(coord_senders),
total_shards,
- self.coord_config,
- self.shutdown_token,
- ))
+ self.coord_config.clone(),
+ self.metrics.clone(),
+ )?))
} else {
None
};
- let mut shard = IggyShard::new(
+ let shard = IggyShard::new(
self.identity,
self.bus,
self.on_replica_message,
@@ -198,20 +214,10 @@ where
self.inbox,
self.shards_table,
self.partition_consensus,
- );
-
- let refresh_task = if let Some((senders_rc, total_shards, cfg, token)) = coord_inputs {
- let refresh_period = cfg.refresh_period;
- let coord = Rc::new(ShardZeroCoordinator::new(senders_rc, total_shards, cfg));
- shard.set_coordinator(Rc::clone(&coord));
- Some(coord.spawn_refresh_task(token, refresh_period))
- } else {
- None
- };
+ coordinator,
+ self.metrics,
+ )?;
- BuiltShard {
- shard,
- refresh_task,
- }
+ Ok(BuiltShard { shard })
}
}
diff --git a/core/shard/src/config.rs b/core/shard/src/config.rs
index 516582834..27535c2e9 100644
--- a/core/shard/src/config.rs
+++ b/core/shard/src/config.rs
@@ -23,19 +23,9 @@
//! from `ServerConfig` lands. Kept in-crate for now to avoid churning
//! the configs crate ahead of that wiring.
-use std::time::Duration;
-
/// Tunables for [`crate::coordinator::ShardZeroCoordinator`].
#[derive(Debug, Clone)]
pub struct CoordinatorConfig {
- /// Cadence at which shard 0 re-broadcasts its authoritative
- /// `replica_id -> owning_shard` snapshot. A shard whose inbox was
- /// full when the original `ReplicaMappingUpdate` was sent recovers
- /// its mapping on the next tick rather than staying silently stale.
- /// Must be comfortably shorter than the VSR view-change timeout so
- /// a missed mapping cannot trigger a spurious view change.
- pub refresh_period: Duration,
-
/// When `total_shards > 1`, exclude shard 0 from replica placement.
/// Shard 0 already hosts the coordinator, the metadata writer, and
/// both listeners; replicas are long-lived steady flows, so offload
@@ -52,7 +42,6 @@ pub struct CoordinatorConfig {
impl Default for CoordinatorConfig {
fn default() -> Self {
Self {
- refresh_period: Duration::from_secs(10),
skip_shard_zero_for_replicas: true,
skip_shard_zero_for_clients: false,
}
diff --git a/core/shard/src/coordinator.rs b/core/shard/src/coordinator.rs
index bcf69f6a2..3815640cd 100644
--- a/core/shard/src/coordinator.rs
+++ b/core/shard/src/coordinator.rs
@@ -23,12 +23,16 @@
//!
//! 1. picks the next target shard via round-robin,
//! 2. duplicates the TCP fd,
-//! 3. sends a `ShardFramePayload::{Replica,Client}ConnectionSetup` frame to
+//! 3. sends a `LifecycleFrame::{Replica,Client}ConnectionSetup` frame to
//! the target shard's inbox,
//! 4. drops its own `TcpStream` so only the target shard's wrapped fd
-//! keeps the socket alive,
-//! 5. broadcasts the resulting replica -> owning shard mapping so every
-//! bus' `send_to_replica` slow path can route to the owner.
+//! keeps the socket alive.
+//!
+//! The shard-shared [`message_bus::ReplicaOwnerTable`] is the
+//! authoritative `replica_id -> owning_shard` view; the owning shard
+//! stamps and CAS-clears its slot synchronously through
+//! `mark_replica_owned` / `clear_replica_owned`, so the coordinator does
+//! not need a separate mapping cache or broadcast subsystem.
//!
//! Client ids encode the owning shard in their top 16 bits, so clients do
//! not need a mapping table; any shard can route a client reply from the
@@ -40,102 +44,69 @@
//! dropped connection.
use crate::config::CoordinatorConfig;
-use crate::{ShardFrame, ShardFramePayload, TaggedSender, assert_sender_ordering};
+use crate::metrics::{frame_drop_reason, frame_drop_variant};
+use crate::{LifecycleFrame, ShardCtorError, ShardFrame, TaggedSender, validate_sender_ordering};
use compio::net::TcpStream;
-use compio::runtime::JoinHandle;
use message_bus::installer::conn_info::{ClientConnMeta, ClientTransportKind};
-use message_bus::lifecycle::ShutdownToken;
use message_bus::{SendError, fd_transfer};
-use std::cell::{Cell, RefCell};
+use std::cell::Cell;
use std::rc::Rc;
-use std::time::Duration;
-use tracing::{debug, warn};
+use tracing::warn;
/// Coordinator owned by shard 0 only.
///
/// Wrapped in `Rc` by the bootstrap and shared with the replica listener,
/// the connector, and the client listener so each of those paths can
/// delegate immediately.
-pub struct ShardZeroCoordinator<R: Send + 'static = ()> {
+pub struct ShardZeroCoordinator {
/// Inter-shard channel senders, indexed by shard id.
///
/// Each [`TaggedSender`] carries the id of the shard whose paired
- /// receiver drains it, and the ctor asserts `senders[i].shard_id() == i`.
- /// The previous plain-`Sender` form was a silent-misroute hazard: a Vec
- /// with correct length but permuted ordering had no way to be caught.
- senders: Rc<Vec<TaggedSender<R>>>,
+ /// receiver drains it; the ctor asserts `senders[i].shard_id() == i`
+ /// so a permuted Vec is caught at boot rather than misrouting frames.
+ senders: Rc<Vec<TaggedSender>>,
+ /// Per-shard observability handle. Cloned at construction; bumps the
+ /// `frame_drops_total{variant=fd_transfer}` counter when a `try_send`
+ /// rejection occurs in the lifecycle path.
+ metrics: crate::metrics::ShardMetrics,
total_shards: u16,
cfg: CoordinatorConfig,
replica_rr: Cell<u16>,
client_rr: Cell<u16>,
client_seq: Cell<u128>,
- /// Authoritative `replica_id -> owning_shard` view as understood by the
- /// coordinator. Populated on successful `delegate_replica`, transitioned
- /// to [`MappingSlot::Cleared`] by [`forget_mapping`](Self::forget_mapping)
- /// when a replica dies. The periodic refresh task re-broadcasts this
- /// snapshot so shards whose inbox was full when the original
- /// `ReplicaMappingUpdate` or `ReplicaMappingClear` was sent recover on
- /// the next tick rather than staying silently stale until the replica
- /// reconnects.
- ///
- /// Three-state slot is what makes refresh complete: `Untouched` slots
- /// emit nothing (the cluster never saw that replica id), `Active` slots
- /// re-emit `ReplicaMappingUpdate`, and `Cleared` slots re-emit
- /// `ReplicaMappingClear`. A two-state `Option<u16>` would silently lose
- /// clears against shards whose inbox was full at clear-time.
- ///
- /// Flat `[MappingSlot; 256]` to mirror `ReplicaRegistry` and avoid a
- /// hash lookup on the refresh hot path.
- mappings: RefCell<[MappingSlot; 256]>,
}
-/// Per-replica mapping state tracked by the coordinator.
-///
-/// `Cleared` is structurally distinct from `Untouched` so the refresh task
-/// can re-broadcast `ReplicaMappingClear` for replicas the cluster did once
-/// see but is no longer routing to.
-#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
-enum MappingSlot {
- /// Coordinator has never delegated this replica id.
- #[default]
- Untouched,
- /// Replica is connected and routed via `owning_shard`.
- Active(u16),
- /// Replica was once active and is now disconnected. Refresh re-broadcasts
- /// a `ReplicaMappingClear` for this slot until a fresh `delegate_replica`
- /// flips it back to `Active`.
- Cleared,
-}
-
-impl<R: Send + 'static> ShardZeroCoordinator<R> {
- /// # Panics
+impl ShardZeroCoordinator {
+ /// # Errors
///
- /// Panics if `senders.len() != total_shards`, if `total_shards < 1`, or
- /// if any `senders[i].shard_id() != i`. Permuted senders are a
- /// bootstrap programming error; `TaggedSender` lifts the ordering
- /// invariant from a doc comment to a ctor assertion.
- #[must_use]
+ /// Returns [`ShardCtorError::CoordinatorSendersMismatch`] if
+ /// `senders.len() != total_shards` or `total_shards == 0`, and
+ /// [`ShardCtorError::SenderOrderingInvalid`] if any
+ /// `senders[i].shard_id() != i`. Permuted senders are a bootstrap
+ /// programming error; `TaggedSender` lifts the ordering invariant
+ /// from a doc comment to a ctor check.
pub fn new(
- senders: Rc<Vec<TaggedSender<R>>>,
+ senders: Rc<Vec<TaggedSender>>,
total_shards: u16,
cfg: CoordinatorConfig,
- ) -> Self {
- assert_eq!(
- senders.len(),
- total_shards as usize,
- "senders must have one entry per shard",
- );
- assert!(total_shards >= 1, "total_shards must be at least 1");
- assert_sender_ordering(&senders);
- Self {
+ metrics: crate::metrics::ShardMetrics,
+ ) -> Result<Self, ShardCtorError> {
+ if total_shards == 0 || senders.len() != total_shards as usize {
+ return Err(ShardCtorError::CoordinatorSendersMismatch {
+ senders: senders.len(),
+ total_shards,
+ });
+ }
+ validate_sender_ordering(&senders)?;
+ Ok(Self {
senders,
+ metrics,
total_shards,
cfg,
replica_rr: Cell::new(0),
client_rr: Cell::new(0),
client_seq: Cell::new(1),
- mappings: RefCell::new([MappingSlot::Untouched; 256]),
- }
+ })
}
/// Pick the next target shard for a replica connection.
@@ -170,11 +141,28 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
(u128::from(target_shard) << 112) | seq
}
+ /// Mint a client id for a connection that terminates locally on shard 0
+ /// (QUIC, TCP-TLS) instead of being round-robin delegated.
+ ///
+ /// Shares the coordinator's `client_seq` counter with the delegated
+ /// TCP/WS path. Both a shard-0-local connection and a delegated
+ /// connection that round-robined to shard 0 encode `target_shard = 0`
+ /// in the top 16 bits; drawing from separate counters would let them
+ /// mint the same id and collide in shard 0's single connection
+ /// registry. One counter keeps every shard-0 id distinct.
+ #[must_use]
+ pub fn mint_shard_zero_client_id(&self) -> u128 {
+ self.mint_client_id(0)
+ }
+
/// Ship a replica TCP connection to the next round-robin target shard.
///
- /// On success broadcasts a `ReplicaMappingUpdate` to every shard and
- /// returns `Ok(target_shard)`. On inter-shard channel failure closes
- /// the duplicated fd and returns `Err(SendError::RoutingFailed)`.
+ /// Returns `Ok(target_shard)` on a successful `try_send`. The owning
+ /// shard records itself in the shard-shared
+ /// [`message_bus::ReplicaOwnerTable`] once its installer accepts the
+ /// fd, so no extra broadcast or mapping cache is needed here. On
+ /// inter-shard channel failure closes the duplicated fd and returns
+ /// `Err(SendError::RoutingFailed)`.
///
/// # Errors
///
@@ -184,11 +172,13 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
let target = self.next_replica_target();
let fd = fd_transfer::dup_fd(&stream).map_err(SendError::DupFailed)?;
- let setup = ShardFramePayload::ReplicaConnectionSetup { fd, replica_id };
+ let setup = LifecycleFrame::ReplicaConnectionSetup { fd, replica_id };
if let Err(e) = self.senders[target as usize].try_send(ShardFrame::lifecycle(setup)) {
// The frame (and the `DupedFd` inside) is returned in `e` and
// dropped at end-of-block, which closes the dup. No explicit
// `close_fd` needed.
+ self.metrics
+ .record_frame_drop(frame_drop_variant::FD_TRANSFER, classify_try_send_err(&e));
warn!(
replica_id,
target, "delegate_replica try_send failed: {e:?}"
@@ -200,35 +190,9 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
// the socket open.
drop(stream);
- // Record the mapping on the coordinator's authoritative snapshot
- // BEFORE broadcasting: if `broadcast_mapping_update` drops a frame
- // on a full inbox, the periodic refresh task reads this snapshot
- // and re-broadcasts.
- self.mappings.borrow_mut()[usize::from(replica_id)] = MappingSlot::Active(target);
- self.broadcast_mapping_update(replica_id, target);
-
Ok(target)
}
- fn broadcast_mapping_update(&self, replica_id: u8, owning_shard: u16) {
- // Broadcast to every shard (including shard 0 and the owner
- // itself). Drops on full inbox are tolerable: the owning shard
- // already holds the fd and the periodic refresh task reconciles
- // any missed mapping on its next tick.
- for sender in self.senders.iter() {
- let update = ShardFramePayload::ReplicaMappingUpdate {
- replica_id,
- owning_shard,
- };
- if let Err(e) = sender.try_send(ShardFrame::lifecycle(update)) {
- debug!(
- shard_id = sender.shard_id(),
- replica_id, "mapping update try_send failed: {e:?}"
- );
- }
- }
- }
-
/// Ship a client TCP connection to the next round-robin target shard.
///
/// On success returns the minted client id. On failure closes the
@@ -247,9 +211,11 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
let fd = fd_transfer::dup_fd(&stream).map_err(SendError::DupFailed)?;
let meta = ClientConnMeta::new(client_id, peer_addr, ClientTransportKind::Tcp);
- let setup = ShardFramePayload::ClientConnectionSetup { fd, meta };
+ let setup = LifecycleFrame::ClientConnectionSetup { fd, meta };
if let Err(e) = self.senders[target as usize].try_send(ShardFrame::lifecycle(setup)) {
// The returned frame owns the `DupedFd` and closes it on drop.
+ self.metrics
+ .record_frame_drop(frame_drop_variant::FD_TRANSFER, classify_try_send_err(&e));
warn!(client_id, target, "delegate_client try_send failed: {e:?}");
return Err(SendError::RoutingFailed(target));
}
@@ -262,7 +228,7 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
/// round-robin target shard.
///
/// Identical wire path to [`Self::delegate_client`] but ships
- /// [`ShardFramePayload::ClientWsConnectionSetup`] so the receiving
+ /// [`LifecycleFrame::ClientWsConnectionSetup`] so the receiving
/// shard runs `compio_ws::accept_async_with_config` before
/// installing the connection. The fd at ship-time is plain TCP;
/// the WS state machine only materialises post-upgrade on the
@@ -280,8 +246,10 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
let fd = fd_transfer::dup_fd(&stream).map_err(SendError::DupFailed)?;
let meta = ClientConnMeta::new(client_id, peer_addr, ClientTransportKind::Ws);
- let setup = ShardFramePayload::ClientWsConnectionSetup { fd, meta };
+ let setup = LifecycleFrame::ClientWsConnectionSetup { fd, meta };
if let Err(e) = self.senders[target as usize].try_send(ShardFrame::lifecycle(setup)) {
+ self.metrics
+ .record_frame_drop(frame_drop_variant::FD_TRANSFER, classify_try_send_err(&e));
warn!(
client_id,
target, "delegate_ws_client try_send failed: {e:?}"
@@ -293,84 +261,21 @@ impl<R: Send + 'static> ShardZeroCoordinator<R> {
Ok(client_id)
}
- /// Broadcast a `ReplicaMappingClear` to every shard. Used by the
- /// `ConnectionLost` handler before the next `delegate_replica` runs.
- pub fn broadcast_mapping_clear(&self, replica_id: u8) {
- crate::broadcast_mapping_clear(&self.senders, replica_id);
- }
-
- /// Transition the coordinator's authoritative entry for `replica_id`
- /// to [`MappingSlot::Cleared`] so the periodic refresh task re-broadcasts
- /// `ReplicaMappingClear` until a fresh delegation revives the slot. A
- /// shard whose inbox was full at the original clear-time recovers on the
- /// next refresh tick rather than retaining the stale `Active(owner)`
- /// view forever.
- ///
- /// Call from shard 0's `ConnectionLost` handler paired with
- /// [`broadcast_mapping_clear`](Self::broadcast_mapping_clear).
- pub fn forget_mapping(&self, replica_id: u8) {
- self.mappings.borrow_mut()[usize::from(replica_id)] = MappingSlot::Cleared;
- }
-
- /// Re-broadcast every mapping the coordinator is tracking.
- ///
- /// `delegate_replica` and `forget_mapping` broadcast their respective
- /// `ReplicaMappingUpdate` / `ReplicaMappingClear` as best-effort
- /// `try_send`: a shard whose inbox was full at the moment of the
- /// original broadcast permanently lost the change. Without periodic
- /// refresh, an `Update` miss leaves `send_to_replica` returning
- /// `ReplicaNotConnected` until the peer reconnects, and a `Clear` miss
- /// leaves a stale `owning_shard` view that misroutes traffic to a shard
- /// that no longer holds the connection. This refresh closes both gaps.
- pub fn broadcast_mapping_snapshot(&self) {
- // Stage entries in scratch `Vec`s so the `RefCell` borrow is
- // released before any `try_send` runs, matching the borrow
- // discipline of the other mapping paths. Heap allocation is
- // acceptable: this snapshot runs on a periodic refresh tick,
- // not on the per-message hot path.
- let mut updates: Vec<(u8, u16)> = Vec::new();
- let mut clears: Vec<u8> = Vec::new();
- for (idx, slot) in self.mappings.borrow().iter().enumerate() {
- #[allow(clippy::cast_possible_truncation)]
- let replica_id = idx as u8;
- match *slot {
- MappingSlot::Active(owner) => updates.push((replica_id, owner)),
- MappingSlot::Cleared => clears.push(replica_id),
- MappingSlot::Untouched => {}
- }
- }
- for (replica_id, owning_shard) in updates {
- self.broadcast_mapping_update(replica_id, owning_shard);
- }
- for replica_id in clears {
- crate::broadcast_mapping_clear(&self.senders, replica_id);
- }
- }
-
- /// Spawn a compio task that calls [`Self::broadcast_mapping_snapshot`]
- /// every `period` until `token` fires. Bootstrap is expected to track
- /// the returned handle on the bus's background tasks so graceful
- /// shutdown awaits it.
- pub fn spawn_refresh_task(
- self: &Rc<Self>,
- token: ShutdownToken,
- period: Duration,
- ) -> JoinHandle<()> {
- let coord = Rc::clone(self);
- compio::runtime::spawn(async move {
- while token.sleep_or_shutdown(period).await {
- coord.broadcast_mapping_snapshot();
- }
- debug!("coordinator mapping refresh task exiting");
- })
- }
-
#[must_use]
pub const fn total_shards(&self) -> u16 {
self.total_shards
}
}
+/// Map `crossfire::TrySendError` to the `frame_drop_reason` label set so
+/// every drop site picks the same string.
+pub(crate) const fn classify_try_send_err<T>(err: &crossfire::TrySendError<T>) -> &'static str {
+ match err {
+ crossfire::TrySendError::Full(_) => frame_drop_reason::FULL,
+ crossfire::TrySendError::Disconnected(_) => frame_drop_reason::DISCONNECTED,
+ }
+}
+
/// Advance `counter` and return the next target shard.
///
/// When `skip_zero` is true and `total_shards > 1`, wraps over
@@ -394,7 +299,7 @@ mod tests {
fn build_senders(total: u16) -> Rc<Vec<TaggedSender>> {
let mut senders = Vec::with_capacity(total as usize);
for shard_id in 0..total {
- let (tx, _rx) = crate::shard_channel::<()>(shard_id, 16);
+ let (tx, _rx) = crate::shard_channel(shard_id, 16);
senders.push(tx);
}
Rc::new(senders)
@@ -406,7 +311,7 @@ mod tests {
let mut senders = Vec::with_capacity(total as usize);
let mut receivers = Vec::with_capacity(total as usize);
for shard_id in 0..total {
- let (tx, rx) = crate::shard_channel::<()>(shard_id, 16);
+ let (tx, rx) = crate::shard_channel(shard_id, 16);
senders.push(tx);
receivers.push(rx);
}
@@ -414,23 +319,42 @@ mod tests {
}
#[test]
- #[should_panic(expected = "inter-shard vec must be in canonical order")]
fn ctor_rejects_permuted_sender_vec() {
// Build senders in correct order, then swap two entries so the
// indexed position no longer matches the tagged shard id.
- let (tx0, _rx0) = crate::shard_channel::<()>(0, 16);
- let (tx1, _rx1) = crate::shard_channel::<()>(1, 16);
- let (tx2, _rx2) = crate::shard_channel::<()>(2, 16);
- let (tx3, _rx3) = crate::shard_channel::<()>(3, 16);
+ let (tx0, _rx0) = crate::shard_channel(0, 16);
+ let (tx1, _rx1) = crate::shard_channel(1, 16);
+ let (tx2, _rx2) = crate::shard_channel(2, 16);
+ let (tx3, _rx3) = crate::shard_channel(3, 16);
let permuted = Rc::new(vec![tx0, tx2, tx1, tx3]);
- let _coord = ShardZeroCoordinator::<()>::new(permuted, 4, CoordinatorConfig::default());
+ let result = ShardZeroCoordinator::new(
+ permuted,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ );
+ assert!(
+ matches!(
+ result.as_ref().err(),
+ Some(ShardCtorError::SenderOrderingInvalid { .. }),
+ ),
+ "expected SenderOrderingInvalid, got {:?}",
+ result.as_ref().err(),
+ );
+ drop(result);
}
#[test]
fn replica_rr_default_skips_shard_zero() {
// Default config: replicas skip shard 0, clients include it.
let senders = build_senders(4);
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, CoordinatorConfig::default());
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
// Replicas wrap over [1, 4).
assert_eq!(coord.next_replica_target(), 1);
@@ -453,9 +377,10 @@ mod tests {
let cfg = CoordinatorConfig {
skip_shard_zero_for_replicas: false,
skip_shard_zero_for_clients: false,
- ..CoordinatorConfig::default()
};
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, cfg);
+ let coord =
+ ShardZeroCoordinator::new(senders, 4, cfg, crate::metrics::ShardMetrics::for_shard())
+ .expect("coord ctor ok");
assert_eq!(coord.next_replica_target(), 0);
assert_eq!(coord.next_replica_target(), 1);
@@ -473,9 +398,10 @@ mod tests {
let cfg = CoordinatorConfig {
skip_shard_zero_for_replicas: true,
skip_shard_zero_for_clients: true,
- ..CoordinatorConfig::default()
};
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, cfg);
+ let coord =
+ ShardZeroCoordinator::new(senders, 4, cfg, crate::metrics::ShardMetrics::for_shard())
+ .expect("coord ctor ok");
for _ in 0..8 {
let r = coord.next_replica_target();
@@ -491,9 +417,10 @@ mod tests {
let cfg = CoordinatorConfig {
skip_shard_zero_for_replicas: true,
skip_shard_zero_for_clients: true,
- ..CoordinatorConfig::default()
};
- let coord = ShardZeroCoordinator::<()>::new(senders, 1, cfg);
+ let coord =
+ ShardZeroCoordinator::new(senders, 1, cfg, crate::metrics::ShardMetrics::for_shard())
+ .expect("coord ctor ok");
for _ in 0..4 {
assert_eq!(coord.next_replica_target(), 0);
assert_eq!(coord.next_client_target(), 0);
@@ -503,18 +430,58 @@ mod tests {
#[test]
fn mint_client_id_encodes_target_shard() {
let senders = build_senders(8);
- let coord = ShardZeroCoordinator::<()>::new(senders, 8, CoordinatorConfig::default());
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 8,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
let id = coord.mint_client_id(5);
assert_eq!((id >> 112) as u16, 5);
assert_eq!(id & ((1u128 << 112) - 1), 1, "first seq is 1");
}
+ #[test]
+ fn shard_zero_local_and_delegated_ids_never_collide() {
+ let senders = build_senders(4);
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
+
+ // Interleave shard-0-local mints (QUIC / TCP-TLS path) with
+ // delegated mints that round-robin to shard 0. Both encode target
+ // shard 0 in the top 16 bits; the shared counter keeps every id
+ // distinct so they cannot overwrite each other in shard 0's
+ // connection registry.
+ let mut ids = std::collections::HashSet::new();
+ for _ in 0..8 {
+ assert!(ids.insert(coord.mint_shard_zero_client_id()));
+ assert!(ids.insert(coord.mint_client_id(0)));
+ }
+ assert_eq!(
+ ids.len(),
+ 16,
+ "every minted shard-0 client id must be unique"
+ );
+ }
+
#[compio::test]
#[allow(clippy::future_not_send)]
- async fn delegate_replica_sends_setup_then_broadcasts_mapping() {
+ async fn delegate_replica_ships_setup_to_target_shard() {
let (senders, receivers) = build_senders_with_rx(4);
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, CoordinatorConfig::default());
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
// Loopback TCP pair so delegate_replica has a real fd to dup.
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -531,29 +498,25 @@ mod tests {
// Target shard should observe ReplicaConnectionSetup.
let setup_frame = receivers[target as usize].recv().await.unwrap();
- match setup_frame.payload {
- ShardFramePayload::ReplicaConnectionSetup { fd, replica_id } => {
+ match setup_frame {
+ ShardFrame::Lifecycle(LifecycleFrame::ReplicaConnectionSetup { fd, replica_id }) => {
assert_eq!(replica_id, 7);
- // Drop closes the dup'd fd via `DupedFd::Drop`.
drop(fd);
}
_ => panic!("expected ReplicaConnectionSetup on target shard"),
}
- // Every shard (including shard 0 and the target itself) should
- // observe the ReplicaMappingUpdate broadcast.
+ // Non-target shards must not receive any frame: delegate_replica
+ // no longer broadcasts a mapping update, the shard-shared
+ // owner table covers cross-shard routing.
for (idx, rx) in receivers.iter().enumerate() {
- let frame = rx.recv().await.unwrap();
- match frame.payload {
- ShardFramePayload::ReplicaMappingUpdate {
- replica_id,
- owning_shard,
- } => {
- assert_eq!(replica_id, 7, "shard {idx} mapping update replica id");
- assert_eq!(owning_shard, target, "shard {idx} mapping update owner");
- }
- _ => panic!("shard {idx} expected ReplicaMappingUpdate"),
+ if idx == target as usize {
+ continue;
}
+ assert!(
+ rx.try_recv().is_err(),
+ "shard {idx} unexpectedly received a frame after delegate_replica",
+ );
}
}
@@ -561,7 +524,13 @@ mod tests {
#[allow(clippy::future_not_send)]
async fn delegate_client_ships_setup_with_meta_transport_tcp() {
let (senders, receivers) = build_senders_with_rx(4);
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, CoordinatorConfig::default());
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
@@ -577,8 +546,8 @@ mod tests {
);
let setup_frame = receivers[target as usize].recv().await.unwrap();
- match setup_frame.payload {
- ShardFramePayload::ClientConnectionSetup { fd, meta } => {
+ match setup_frame {
+ ShardFrame::Lifecycle(LifecycleFrame::ClientConnectionSetup { fd, meta }) => {
assert_eq!(meta.client_id, client_id);
assert!(matches!(meta.transport, ClientTransportKind::Tcp));
drop(fd);
@@ -591,7 +560,13 @@ mod tests {
#[allow(clippy::future_not_send)]
async fn delegate_ws_client_ships_setup_with_meta_transport_ws() {
let (senders, receivers) = build_senders_with_rx(4);
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, CoordinatorConfig::default());
+ let coord = ShardZeroCoordinator::new(
+ senders,
+ 4,
+ CoordinatorConfig::default(),
+ crate::metrics::ShardMetrics::for_shard(),
+ )
+ .expect("coord ctor ok");
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
@@ -607,8 +582,8 @@ mod tests {
);
let setup_frame = receivers[target as usize].recv().await.unwrap();
- match setup_frame.payload {
- ShardFramePayload::ClientWsConnectionSetup { fd, meta } => {
+ match setup_frame {
+ ShardFrame::Lifecycle(LifecycleFrame::ClientWsConnectionSetup { fd, meta }) => {
assert_eq!(meta.client_id, client_id);
assert!(
matches!(meta.transport, ClientTransportKind::Ws),
@@ -620,156 +595,4 @@ mod tests {
_ => panic!("expected ClientWsConnectionSetup variant"),
}
}
-
- #[compio::test]
- #[allow(clippy::future_not_send)]
- async fn broadcast_mapping_clear_reaches_every_shard() {
- let (senders, receivers) = build_senders_with_rx(4);
- let coord = ShardZeroCoordinator::<()>::new(senders, 4, CoordinatorConfig::default());
-
- coord.broadcast_mapping_clear(9);
-
- for (idx, rx) in receivers.iter().enumerate() {
- let frame = rx.recv().await.unwrap();
- match frame.payload {
- ShardFramePayload::ReplicaMappingClear { replica_id } => {
- assert_eq!(replica_id, 9, "shard {idx} mapping clear replica id");
- }
- _ => panic!("shard {idx} expected ReplicaMappingClear"),
- }
- }
- }
-
- #[compio::test]
- #[allow(clippy::future_not_send)]
- async fn snapshot_rebroadcasts_every_tracked_mapping() {
- let (senders, receivers) = build_senders_with_rx(4);
- let coord = Rc::new(ShardZeroCoordinator::<()>::new(
- senders,
- 4,
- CoordinatorConfig::default(),
- ));
-
- // Seed the coordinator's tracked mappings directly (delegate_replica
- // needs a real TCP fd; the snapshot path is orthogonal to dup).
- coord.mappings.borrow_mut()[3] = MappingSlot::Active(1);
- coord.mappings.borrow_mut()[7] = MappingSlot::Active(2);
-
- coord.broadcast_mapping_snapshot();
-
- // Each shard must observe an update for each seeded mapping. The
- // order across replica ids is unspecified (HashMap->Vec iteration
- // on the collected snapshot), so collect and compare as a set.
- for (idx, rx) in receivers.iter().enumerate() {
- let mut observed = std::collections::BTreeSet::new();
- for _ in 0..2 {
- let frame = rx.recv().await.unwrap();
- match frame.payload {
- ShardFramePayload::ReplicaMappingUpdate {
- replica_id,
- owning_shard,
- } => {
- observed.insert((replica_id, owning_shard));
- }
- _ => panic!("shard {idx} expected ReplicaMappingUpdate"),
- }
- }
- let expected: std::collections::BTreeSet<_> =
- [(3u8, 1u16), (7u8, 2u16)].into_iter().collect();
- assert_eq!(
- observed, expected,
- "shard {idx} did not receive the full snapshot"
- );
- }
- }
-
- #[compio::test]
- #[allow(clippy::future_not_send)]
- async fn forget_mapping_transitions_slot_to_cleared_state() {
- let (senders, _receivers) = build_senders_with_rx(2);
- let coord = ShardZeroCoordinator::<()>::new(senders, 2, CoordinatorConfig::default());
-
- coord.mappings.borrow_mut()[4] = MappingSlot::Active(1);
- coord.mappings.borrow_mut()[5] = MappingSlot::Active(1);
- coord.forget_mapping(4);
-
- assert_eq!(coord.mappings.borrow()[4], MappingSlot::Cleared);
- assert_eq!(coord.mappings.borrow()[5], MappingSlot::Active(1));
- }
-
- #[compio::test]
- #[allow(clippy::future_not_send)]
- async fn snapshot_includes_clear_for_forgotten_slots() {
- let (senders, receivers) = build_senders_with_rx(3);
- let coord = Rc::new(ShardZeroCoordinator::<()>::new(
- senders,
- 3,
- CoordinatorConfig::default(),
- ));
-
- // Two live mappings + one slot that was previously delegated and is
- // now forgotten. Refresh must rebroadcast a Clear for the forgotten
- // slot so a shard that missed the original Clear (full inbox) can
- // reconcile its stale Active view on the next tick.
- coord.mappings.borrow_mut()[3] = MappingSlot::Active(1);
- coord.mappings.borrow_mut()[7] = MappingSlot::Active(2);
- coord.mappings.borrow_mut()[5] = MappingSlot::Cleared;
-
- coord.broadcast_mapping_snapshot();
-
- for (idx, rx) in receivers.iter().enumerate() {
- let mut updates = std::collections::BTreeSet::new();
- let mut clears = std::collections::BTreeSet::new();
- for _ in 0..3 {
- let frame = rx.recv().await.unwrap();
- match frame.payload {
- ShardFramePayload::ReplicaMappingUpdate {
- replica_id,
- owning_shard,
- } => {
- updates.insert((replica_id, owning_shard));
- }
- ShardFramePayload::ReplicaMappingClear { replica_id } => {
- clears.insert(replica_id);
- }
- _ => panic!("shard {idx} expected mapping update or clear"),
- }
- }
-
- let expected_updates: std::collections::BTreeSet<_> =
- [(3u8, 1u16), (7u8, 2u16)].into_iter().collect();
- let expected_clears: std::collections::BTreeSet<_> = std::iter::once(5u8).collect();
- assert_eq!(
- updates, expected_updates,
- "shard {idx} did not receive both Update frames"
- );
- assert_eq!(
- clears, expected_clears,
- "shard {idx} did not receive a Clear for the forgotten slot"
- );
- }
- }
-
- #[compio::test]
- #[allow(clippy::future_not_send)]
- async fn snapshot_skips_untouched_slots() {
- let (senders, receivers) = build_senders_with_rx(2);
- let coord = Rc::new(ShardZeroCoordinator::<()>::new(
- senders,
- 2,
- CoordinatorConfig::default(),
- ));
-
- // No mappings populated: every slot is `Untouched`. Snapshot must
- // be silent so the cluster does not pay 256 try_sends per shard per
- // refresh tick when nothing has been delegated yet.
- coord.broadcast_mapping_snapshot();
-
- for (idx, rx) in receivers.iter().enumerate() {
- assert!(
- rx.try_recv().is_err(),
- "shard {idx} should not receive any frame from an empty snapshot"
- );
- }
- }
}
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 2dc7a4942..d199b3f33 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -18,6 +18,7 @@
pub mod builder;
pub mod config;
pub mod coordinator;
+pub mod metrics;
mod router;
pub mod shards_table;
@@ -105,14 +106,36 @@ pub fn channel<T: Send + 'static>(capacity: usize) -> (Sender<T>, Receiver<T>) {
/// Bootstrap uses this to build the per-shard sender `Vec` such that
/// `vec[i]` necessarily reaches shard `i`.
#[must_use]
-pub fn shard_channel<R: Send + 'static>(
- owner_shard: u16,
- capacity: usize,
-) -> (TaggedSender<R>, Receiver<ShardFrame<R>>) {
- let (tx, rx) = channel::<ShardFrame<R>>(capacity);
+pub fn shard_channel(owner_shard: u16, capacity: usize) -> (TaggedSender, Receiver<ShardFrame>) {
+ let (tx, rx) = channel::<ShardFrame>(capacity);
(TaggedSender::new(owner_shard, tx), rx)
}
+/// Build canonical-ordered `(senders, inboxes)` pair for an N-shard mesh.
+///
+/// Each `inboxes[i]` drains exclusively on the runtime owning shard `i`. The
+/// returned `senders` Vec satisfies `senders[i].shard_id() == i` by
+/// construction; clone it into every shard before spawning so all shards
+/// share the same mesh.
+///
+/// Receivers are wrapped in `Option` because [`Receiver`] (crossfire
+/// `AsyncRx`) is non-cloneable on purpose; bootstrap takes the slot for
+/// shard `i` exactly once when spawning the owning thread.
+#[must_use]
+pub fn shard_mesh_channels(
+ total_shards: u16,
+ capacity: usize,
+) -> (Vec<TaggedSender>, Vec<Option<Receiver<ShardFrame>>>) {
+ let mut senders = Vec::with_capacity(total_shards as usize);
+ let mut inboxes = Vec::with_capacity(total_shards as usize);
+ for shard_id in 0..total_shards {
+ let (tx, rx) = shard_channel(shard_id, capacity);
+ senders.push(tx);
+ inboxes.push(Some(rx));
+ }
+ (senders, inboxes)
+}
+
/// A [`Sender`] annotated with the id of the shard whose paired receiver it
/// feeds.
///
@@ -121,20 +144,20 @@ pub fn shard_channel<R: Send + 'static>(
/// permuted `Vec<Sender<_>>` would silently misroute every setup, mapping,
/// and forward frame. Construct senders through [`shard_channel`] (or
/// [`TaggedSender::new`]) at the channel-creation site; the coordinator and
-/// [`IggyShard`] ctors then assert `senders[i].shard_id() == i`, turning the
-/// ordering invariant from a doc comment into a ctor-checked type property.
-pub struct TaggedSender<R: Send + 'static = ()> {
+/// [`IggyShard`] ctors then validate `senders[i].shard_id() == i`,
+/// returning [`ShardCtorError`] if violated.
+pub struct TaggedSender {
shard_id: u16,
- inner: Sender<ShardFrame<R>>,
+ inner: Sender<ShardFrame>,
}
-impl<R: Send + 'static> TaggedSender<R> {
+impl TaggedSender {
/// Wrap an already-constructed sender with the id of the shard whose
/// paired receiver drains it. Prefer [`shard_channel`] unless an
/// existing sender is being re-tagged (e.g., tests that build senders
/// manually and know the ordering is correct).
#[must_use]
- pub const fn new(shard_id: u16, inner: Sender<ShardFrame<R>>) -> Self {
+ pub const fn new(shard_id: u16, inner: Sender<ShardFrame>) -> Self {
Self { shard_id, inner }
}
@@ -142,14 +165,9 @@ impl<R: Send + 'static> TaggedSender<R> {
pub const fn shard_id(&self) -> u16 {
self.shard_id
}
-
- #[must_use]
- pub const fn sender(&self) -> &Sender<ShardFrame<R>> {
- &self.inner
- }
}
-impl<R: Send + 'static> Clone for TaggedSender<R> {
+impl Clone for TaggedSender {
fn clone(&self) -> Self {
Self {
shard_id: self.shard_id,
@@ -158,38 +176,78 @@ impl<R: Send + 'static> Clone for TaggedSender<R> {
}
}
-impl<R: Send + 'static> std::ops::Deref for TaggedSender<R> {
- type Target = Sender<ShardFrame<R>>;
+impl std::ops::Deref for TaggedSender {
+ type Target = Sender<ShardFrame>;
fn deref(&self) -> &Self::Target {
&self.inner
}
}
-/// Assert the canonical ordering `senders[i].shard_id() == i`. Called from
-/// every ctor that accepts a pre-built `Vec<TaggedSender<_>>`.
-///
-/// # Panics
+/// Error returned by [`IggyShard::new`] and the shard builder when ctor
+/// preconditions are violated.
///
-/// Panics if any sender in `senders` carries a `shard_id` that does not
-/// match its index. Bootstrap programming error.
-fn assert_sender_ordering<R: Send + 'static>(senders: &[TaggedSender<R>]) {
+/// Both are bootstrap programming errors: the surrounding crate either
+/// built the `senders` vec out of canonical order, or produced more
+/// shards than the inter-shard addressing scheme supports. Surfaced as
+/// `Err` instead of panicking so the host process can log and abort with
+/// a typed error.
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum ShardCtorError {
+ #[error(
+ "senders[{index}] carries shard_id {actual}; inter-shard vec must be in canonical \
+ order (senders[i].shard_id() == i)"
+ )]
+ SenderOrderingInvalid {
+ index: usize,
+ expected: u16,
+ actual: u16,
+ },
+ #[error("shard count {count} does not fit in u16; inter-shard frame addressing is u16-indexed")]
+ ShardCountOverflow { count: usize },
+ #[error(
+ "shard-0 coordinator senders length {senders} does not match total_shards {total_shards} \
+ (total_shards must be >= 1 and equal senders.len())"
+ )]
+ CoordinatorSendersMismatch { senders: usize, total_shards: u16 },
+}
+
+/// Validate the canonical ordering `senders[i].shard_id() == i`.
+/// Returns `Err` for the first index that violates the invariant.
+pub(crate) fn validate_sender_ordering(senders: &[TaggedSender]) -> Result<(), ShardCtorError> {
for (idx, sender) in senders.iter().enumerate() {
- let expected = u16::try_from(idx).expect("shard count must fit in u16");
- assert_eq!(
- sender.shard_id(),
- expected,
- "senders[{idx}] carries shard_id {}; inter-shard vec must be in canonical order",
- sender.shard_id(),
- );
+ let expected = u16::try_from(idx).map_err(|_| ShardCtorError::ShardCountOverflow {
+ count: senders.len(),
+ })?;
+ let actual = sender.shard_id();
+ if actual != expected {
+ return Err(ShardCtorError::SenderOrderingInvalid {
+ index: idx,
+ expected,
+ actual,
+ });
+ }
}
+ Ok(())
}
-/// Payload carried by a [`ShardFrame`].
+/// Lifecycle frame variants.
+///
+/// Connection setup and cross-shard forwards: every frame the inter-shard
+/// channel carries that is NOT a consensus protocol message lives here.
+/// Splitting these out from [`ShardFrame::Consensus`] keeps the consensus
+/// dispatch path hot and cache-tight while leaving lifecycle traffic on
+/// the same single channel (preserving relative ordering between consensus
+/// and lifecycle frames at near-zero cost).
+///
+/// Trade-off: consensus and lifecycle traffic compete for one bounded
+/// inbox. A consensus burst or retransmit storm can fill it exactly when
+/// a terminal-drop [`LifecycleFrame::ForwardClientSend`] needs the space;
+/// `inbox_capacity` is a single knob and cannot isolate the two frame
+/// classes.
#[non_exhaustive]
-pub enum ShardFramePayload {
- /// A consensus protocol message routed between shards.
- Consensus(Message<GenericHeader>),
+pub enum LifecycleFrame {
/// Shard 0 distributes an inbound replica TCP connection fd to the owning
/// shard. The receiving shard wraps the fd in a `TcpStream` and spawns
/// writer + reader tasks on its own compio runtime. The `fd` is an
@@ -221,12 +279,6 @@ pub enum ShardFramePayload {
/// Shard 0 therefore terminates QUIC locally and uses the existing
/// `ForwardClientSend` variant for outbound traffic.
ClientWsConnectionSetup { fd: DupedFd, meta: ClientConnMeta },
- /// Shard 0 broadcasts the owner for a replica to every shard so each
- /// bus' `send_to_replica` slow path can route through the correct owner.
- ReplicaMappingUpdate { replica_id: u8, owning_shard: u16 },
- /// Shard 0 broadcasts that a replica mapping should be forgotten (e.g.
- /// after a connection loss and before the next allocate).
- ReplicaMappingClear { replica_id: u8 },
/// A non-owning shard forwards a replica send to the owning shard's
/// local bus; the owning shard then takes the fast path.
ForwardReplicaSend {
@@ -239,83 +291,58 @@ pub enum ShardFramePayload {
client_id: u128,
msg: Frozen<MESSAGE_ALIGN>,
},
- /// Owning shard notifies shard 0 that a replica connection died.
- /// Shard 0 clears the mapping cluster-wide and drives a reconnect.
- ConnectionLost { replica_id: u8 },
}
-/// Envelope for inter-shard channel messages.
-///
-/// Wraps a [`ShardFramePayload`] together with an optional one-shot response
-/// channel. Fire-and-forget dispatches leave `response_sender` as `None`;
-/// request-response dispatches provide a sender that the message pump will
-/// notify once the message has been processed.
+/// Inter-shard channel envelope.
///
-/// The response type `R` is generic so that higher layers (e.g. HTTP handlers)
-/// can carry a response enum while the consensus layer can default to `()`.
-pub struct ShardFrame<R: Send + 'static = ()> {
- pub payload: ShardFramePayload,
- pub response_sender: Option<Sender<R>>,
+/// Concrete enum; no generic. Consensus dispatches are fire-and-forget by
+/// VSR design (replies travel as their own wire-level messages: `Reply`
+/// to clients, `PrepareOk` to the primary), so no response channel rides
+/// in the frame.
+#[non_exhaustive]
+pub enum ShardFrame {
+ /// A consensus protocol message (Request / Prepare / `PrepareOk` /
+ /// view-change family / Commit). Fire-and-forget. Drops on full inbox
+ /// are recovered by VSR retransmit timers.
+ ///
+ /// `target_shard` is stamped by the sender at enqueue time, so the
+ /// receiving pump never re-derives routing in release builds. The
+ /// receiver still validates `target_shard == self.id` and drops
+ /// frames stamped for the wrong shard (`MISROUTED`) to preserve the
+ /// single-pump invariant under any caller bug.
+ Consensus {
+ target_shard: u16,
+ message: Message<GenericHeader>,
+ },
+ /// A connection setup or cross-shard forward frame. Drop recovery
+ /// depends on the frame class: [`LifecycleFrame::ForwardReplicaSend`]
+ /// is VSR-covered, connection-setup frames are recovered by the
+ /// connector's periodic reconnect sweep, but
+ /// [`LifecycleFrame::ForwardClientSend`] is terminal - no retransmit,
+ /// the client never receives the reply.
+ Lifecycle(LifecycleFrame),
}
-impl<R: Send + 'static> ShardFrame<R> {
- /// Create a fire-and-forget consensus message frame.
+impl ShardFrame {
+ /// Create a consensus frame addressed to `target_shard`. The sender
+ /// is the routing authority; `accept_frame_for_self` compares this
+ /// stamp against the receiving shard id in O(1).
#[must_use]
- pub const fn fire_and_forget(message: Message<GenericHeader>) -> Self {
- Self {
- payload: ShardFramePayload::Consensus(message),
- response_sender: None,
+ pub const fn consensus(target_shard: u16, message: Message<GenericHeader>) -> Self {
+ Self::Consensus {
+ target_shard,
+ message,
}
}
- /// Create a request-response consensus message frame. Returns the frame
- /// and a receiver that the caller can await for completion notification.
+ /// Create a lifecycle frame.
#[must_use]
- pub fn with_response(message: Message<GenericHeader>) -> (Self, Receiver<R>) {
- let (tx, rx) = channel(1);
- (
- Self {
- payload: ShardFramePayload::Consensus(message),
- response_sender: Some(tx),
- },
- rx,
- )
- }
-
- /// Create a fire-and-forget lifecycle frame (connection setup, loss).
- #[must_use]
- pub const fn lifecycle(payload: ShardFramePayload) -> Self {
- Self {
- payload,
- response_sender: None,
- }
+ pub const fn lifecycle(payload: LifecycleFrame) -> Self {
+ Self::Lifecycle(payload)
}
}
-/// Broadcast a `ReplicaMappingClear` to every shard (including sender-self).
-///
-/// Used by shard 0's `ConnectionLost` handler and by
-/// [`coordinator::ShardZeroCoordinator::broadcast_mapping_clear`] so both
-/// paths go through the same try-send-and-log logic. `try_send` failures
-/// (inbox full or disconnected) are logged at debug: the next mapping
-/// broadcast or reconnect sweep will reconcile.
-pub(crate) fn broadcast_mapping_clear<R: Send + 'static>(
- senders: &[TaggedSender<R>],
- replica_id: u8,
-) {
- for sender in senders {
- let clear = ShardFramePayload::ReplicaMappingClear { replica_id };
- if let Err(e) = sender.try_send(ShardFrame::lifecycle(clear)) {
- tracing::debug!(
- shard_id = sender.shard_id(),
- replica_id,
- "mapping clear try_send failed: {e:?}"
- );
- }
- }
-}
-
-pub struct IggyShard<B, MJ, S, M, T = (), R: Send + 'static = ()>
+pub struct IggyShard<B, MJ, S, M, T = ()>
where
B: MessageBus,
{
@@ -324,8 +351,8 @@ where
pub plane: ShardPlane<B, MJ, S, M>,
/// Handle to the local bus. Retained alongside the bus owned by every
- /// consensus plane so the router can reach the `ConnectionInstaller` /
- /// mapping surface without going through consensus.
+ /// consensus plane so the router can reach the `ConnectionInstaller`
+ /// surface without going through consensus.
pub bus: B,
/// Callback attached to every delegated replica connection installed
@@ -344,26 +371,33 @@ where
/// [`assert_sender_ordering`] is invoked in the ctor so `senders[i]`
/// is guaranteed to feed the shard whose `id == i`. Call sites can
/// therefore index by `target_shard` without re-checking.
- senders: Vec<TaggedSender<R>>,
+ senders: Vec<TaggedSender>,
+
+ /// Total shard count, cached from `senders.len()` at construction.
+ /// `senders` is immutable post-ctor, so consensus routing reads this
+ /// rather than recomputing the `usize -> u32` conversion per frame.
+ shard_count: u32,
/// Receiver end of this shard's inbox. Peer shards (and self) send
/// messages here via the corresponding sender.
- inbox: Receiver<ShardFrame<R>>,
+ inbox: Receiver<ShardFrame>,
/// Partition namespace -> owning shard lookup.
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
- /// Shard 0 coordinator, set at bootstrap on shard 0 only. The router's
- /// `ConnectionLost` handler calls [`coordinator::ShardZeroCoordinator::forget_mapping`]
- /// so the periodic refresh task stops re-broadcasting the mapping of
- /// a replica that has disconnected. `None` on non-zero shards and in
- /// single-shard tests that bypass the coordinator.
- coordinator: Option<Rc<crate::coordinator::ShardZeroCoordinator<R>>>,
+ /// Shard 0 coordinator, supplied at construction. Holds round-robin
+ /// state for replica and client delegation. `None` on non-zero shards
+ /// and in single-shard tests that bypass the coordinator.
+ coordinator: Option<Rc<crate::coordinator::ShardZeroCoordinator>>,
+
+ /// Per-shard observability counters. Cloned at metric increment sites,
+ /// so cheap (`Arc` clone) regardless of label cardinality.
+ metrics: crate::metrics::ShardMetrics,
}
-impl<B, MJ, S, M, T, R: Send + 'static> IggyShard<B, MJ, S, M, T, R>
+impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T>
where
B: MessageBus,
T: ShardsTable,
@@ -371,22 +405,29 @@ where
/// Create a new shard with channel links and a shards table.
///
/// * `bus` - shard-local bus handle (kept alongside the buses owned
- /// by the consensus planes so the router can reach installer /
- /// mapping operations directly).
+ /// by the consensus planes so the router can reach the
+ /// `ConnectionInstaller` surface directly).
/// * `senders` - one [`TaggedSender`] per shard. The ctor asserts
/// `senders[i].shard_id() == i`; use [`shard_channel`] at
/// construction time so every sender carries the id of the shard
/// whose receiver drains it.
/// * `inbox` - the receiver that this shard drains in its message pump.
/// * `shards_table` - namespace -> shard routing table.
+ /// * `coordinator` - `Some` on shard 0 (supplied by the builder when
+ /// `is_shard_zero`), `None` everywhere else. Immutable post-ctor:
+ /// the coordinator is injected at construction time so an
+ /// `IggyShard` cannot appear half-wired to a reader.
+ /// * `metrics` - per-shard observability handle; currently the
+ /// `frame_drops_total` counter.
///
- /// # Panics
+ /// # Errors
///
- /// Panics if `senders` is not in canonical order (any
- /// `senders[i].shard_id() != i`). That is a bootstrap programming
- /// error; the resulting permutation would silently misroute every
- /// inter-shard frame.
- #[must_use]
+ /// Returns [`ShardCtorError::SenderOrderingInvalid`] if `senders` is
+ /// not in canonical order (any `senders[i].shard_id() != i`) and
+ /// [`ShardCtorError::ShardCountOverflow`] if `senders.len()` does not
+ /// fit in `u16`. Both are bootstrap programming errors: the
+ /// permutation would silently misroute every inter-shard frame, or
+ /// addressing space (u16) would wrap.
#[allow(clippy::too_many_arguments)]
pub fn new(
identity: ShardIdentity,
@@ -395,15 +436,21 @@ where
on_client_request: RequestHandler,
metadata: IggyMetadata<VsrConsensus<B>, MJ, S, M>,
partitions: IggyPartitions<B>,
- senders: Vec<TaggedSender<R>>,
- inbox: Receiver<ShardFrame<R>>,
+ senders: Vec<TaggedSender>,
+ inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
- ) -> Self {
- assert_sender_ordering(&senders);
+ coordinator: Option<Rc<crate::coordinator::ShardZeroCoordinator>>,
+ metrics: crate::metrics::ShardMetrics,
+ ) -> Result<Self, ShardCtorError> {
+ validate_sender_ordering(&senders)?;
+ let shard_count =
+ u32::try_from(senders.len()).map_err(|_| ShardCtorError::ShardCountOverflow {
+ count: senders.len(),
+ })?;
let plane = MuxPlane::new(variadic!(metadata, partitions));
let ShardIdentity { id, name } = identity;
- Self {
+ Ok(Self {
id,
name,
plane,
@@ -411,29 +458,22 @@ where
on_replica_message,
on_client_request,
senders,
+ shard_count,
inbox,
shards_table,
partition_consensus,
- coordinator: None,
- }
- }
-
- /// Attach a shard-0 coordinator. Bootstrap calls this on the shard 0
- /// `IggyShard` only; other shards keep `coordinator = None`. The
- /// router's `ConnectionLost` handler will call
- /// [`coordinator::ShardZeroCoordinator::forget_mapping`] via this
- /// reference so the periodic refresh stops re-broadcasting dead
- /// replica mappings.
- pub fn set_coordinator(&mut self, coord: Rc<crate::coordinator::ShardZeroCoordinator<R>>) {
- self.coordinator = Some(coord);
+ coordinator,
+ metrics,
+ })
}
- /// `true` when a [`coordinator::ShardZeroCoordinator`] has been attached
- /// via [`set_coordinator`](Self::set_coordinator). Shard 0 bootstrap is
- /// expected to flip this on; every other shard keeps it `false`.
+ /// Return a clone of the shard-0 coordinator handle, if attached.
+ /// Bootstrap uses this to wire the listener accept callbacks
+ /// (replica + client) to coordinator-driven fd-delegation instead
+ /// of installing connections locally on shard 0.
#[must_use]
- pub const fn has_coordinator(&self) -> bool {
- self.coordinator.is_some()
+ pub fn coordinator(&self) -> Option<Rc<crate::coordinator::ShardZeroCoordinator>> {
+ self.coordinator.clone()
}
/// Create a shard without inter-shard channels or delegated connections.
@@ -451,9 +491,9 @@ where
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
) -> Self {
- // TODO: previously we used unbounded channel with flume,
- // but this is not possible with crossfire without mangling types due to Flavor trait in crossfire.
- // This needs to be revisited in the future.
+ // TODO(hubcio): crossfire's Flavor trait blocks unbounded channels
+ // with the current type setup; revisit when crossfire grows an
+ // unbounded variant or we replace it.
let (_tx, inbox) = channel(1);
let plane = MuxPlane::new(variadic!(metadata, partitions));
let ShardIdentity { id, name } = identity;
@@ -466,9 +506,17 @@ where
plane,
coordinator: None,
senders: Vec::new(),
+ // The simulator delivers inbound messages straight to
+ // `on_message`, bypassing the inter-shard router. The router's
+ // `shard_count` should therefore never be read on this path,
+ // but `pub fn dispatch` is still reachable; pinning to 1 keeps
+ // `% shard_count` from panicking if a future caller slips
+ // through, while preserving single-shard routing semantics.
+ shard_count: 1,
inbox,
shards_table,
partition_consensus,
+ metrics: crate::metrics::ShardMetrics::for_shard(),
}
}
@@ -480,7 +528,7 @@ where
/// Local message processing — these methods handle messages that have been
/// routed to this shard via the message pump.
-impl<B, MJ, S, M, T, R: Send + 'static> IggyShard<B, MJ, S, M, T, R>
+impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T>
where
B: MessageBus,
{
@@ -488,6 +536,15 @@ where
///
/// Routes requests, replication messages, and acks to either the metadata
/// plane or the partitions plane based on `PlaneIdentity::is_applicable`.
+ //
+ // TODO(hubcio): perf - this `MessageBag::try_from` is the second parse of
+ // the same frame; the first ran in `IggyShard::dispatch` (router.rs ~85)
+ // to extract (operation, namespace) for routing. The work here re-runs
+ // `bytemuck::checked::try_from_bytes` + per-header `validate()` on bytes
+ // already validated upstream. See the matching TODO in router.rs for the
+ // fix: thread the classified `MessageBag` through `ShardFrame::Consensus`
+ // so this function takes the bag directly and the match below dispatches
+ // without re-parsing.
#[allow(clippy::future_not_send)]
pub async fn on_message(&self, message: Message<GenericHeader>)
where
@@ -587,7 +644,11 @@ where
/// # Panics
/// Panics if a loopback message is not a valid `PrepareOk` message.
#[allow(clippy::future_not_send)]
- pub async fn process_loopback(&self, buf: &mut Vec<Message<GenericHeader>>) -> usize
+ pub async fn process_loopback(
+ &self,
+ buf: &mut Vec<Message<GenericHeader>>,
+ namespace_scratch: &mut Vec<IggyNamespace>,
+ ) -> usize
where
B: MessageBus,
MJ: JournalHandle,
@@ -603,6 +664,10 @@ where
> + StreamsFrontend,
{
debug_assert!(buf.is_empty(), "buf must be empty on entry");
+ debug_assert!(
+ namespace_scratch.is_empty(),
+ "namespace_scratch must be empty on entry",
+ );
let mut total = 0;
let planes = self.plane.inner();
@@ -619,8 +684,8 @@ where
}
}
- let namespaces: Vec<_> = planes.1.0.namespaces().copied().collect();
- for namespace in namespaces {
+ namespace_scratch.extend(planes.1.0.namespaces().copied());
+ for namespace in namespace_scratch.drain(..) {
let partition = planes
.1
.0
@@ -642,9 +707,17 @@ where
/// Initializes a partition and its dedicated consensus instance on this shard.
///
+ /// Used only by the `simulator` crate to seed partitions on the
+ /// in-process replica array. Production boots via the
+ /// `load_partition` helper in `server-ng` bootstrap, which takes the
+ /// cluster `self_replica_id` explicitly from `topology` and never
+ /// folds the local shard id into the consensus replica space.
+ ///
/// # Panics
- /// Panics if the shard id does not fit in `u8`, which is currently required
- /// by the partition consensus replica id.
+ /// Panics if `self.id > u8::MAX` (256+). The simulator currently
+ /// caps replica count at `MAX_REPLICAS = 256`, so the cast always
+ /// succeeds in the only caller; the constraint is recorded here so
+ /// any future caller in production code reads it before the cast.
pub fn init_partition(&mut self, namespace: IggyNamespace)
where
B: MessageBus + Clone,
@@ -678,6 +751,23 @@ where
/// Handle incoming view-change/control message. Metadata use metadata
/// consensus. Partitions loop all partitions, use partition consensus.
+ ///
+ // TODO(hubcio): every VSR callback below
+ // (`on_start_view_change`, `on_do_view_change`, `on_start_view`,
+ // `on_commit`, `tick_partitions`) materialises
+ // `planes.1.0.namespaces().copied().collect::<Vec<_>>()` per call to
+ // avoid borrowing the partitions plane across the partition-consensus
+ // `.await` inside the loop. Allocs scale with VSR traffic, not with
+ // useful work: a quiet cluster still pays one Vec per heartbeat tick.
+ // Convert to the `namespace_scratch: RefCell<Vec<IggyNamespace>>`
+ // pattern already used by `process_loopback` (see :646-684) so the
+ // scratch is reused across calls. Asserts on entry/exit keep the
+ // "empty on entry, drained on exit" invariant explicit.
+ //
+ // Reproducible in `core/simulator`: sim drives these callbacks via
+ // `tick()` against `IggyShard`, so an allocation-counting variant of
+ // `MemStorage`/test harness can pin the per-VSR-cb alloc count and
+ // fail on regression after the scratch refactor lands.
#[allow(clippy::future_not_send)]
async fn on_start_view_change(&self, msg: Message<StartViewChangeHeader>)
where
diff --git a/core/shard/src/metrics.rs b/core/shard/src/metrics.rs
new file mode 100644
index 000000000..6188fd485
--- /dev/null
+++ b/core/shard/src/metrics.rs
@@ -0,0 +1,258 @@
+// 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.
+
+//! Per-shard frame-drop accounting.
+//!
+//! [`ShardMetrics`] holds `frame_drops_total{variant, reason}`,
+//! bumped whenever an inter-shard `try_send` is rejected (`Full` /
+//! `Disconnected`) or its target shard id is out of range
+//! (`Unroutable`):
+//! - [`crate::coordinator::ShardZeroCoordinator`] - fd-transfer delegation.
+//! - the cross-shard forward closures built in [`crate::builder`].
+//! - `IggyShard::try_send_to_target` - consensus frames.
+//!
+//! The counter uses atomic interior mutability, safe to bump from `!Send`
+//! compio reactor contexts. Each shard owns its own instance. It is not
+//! yet exposed through a scrape endpoint; every drop site also logs via
+//! `tracing`, so the counter is a structured complement to those logs
+//! until a per-shard exporter lands.
+//!
+//! TODO(hubcio): register `frame_drops_total` with a per-shard prometheus
+//! exporter so the counter is scrape-able; until then the `tracing`
+//! drop-site logs are the only alertable signal.
+
+use prometheus_client::encoding::EncodeLabelSet;
+use prometheus_client::metrics::counter::Counter;
+use prometheus_client::metrics::family::Family;
+
+/// Label for `frame_drops_total`.
+///
+/// `variant` describes the dropped frame class; `reason` is `"full"` or
+/// `"disconnected"` per crossfire `TrySendError`, `"unroutable"` when the
+/// target shard id has no sender slot, `"delivery_failed"` when the
+/// receiver path could not place the frame, or `"misrouted"` when a frame
+/// reached a shard that does not own its namespace.
+///
+/// `shard_id` is intentionally NOT a label here: each shard owns its own
+/// `Family<FrameDropLabel, Counter>`, so the per-shard scope is implicit
+/// in the registry the family is exported through. A future scrape
+/// exporter must attach `shard_id` as a target label rather than as part
+/// of the per-counter label set to keep cardinality bounded.
+#[derive(Clone, Hash, Eq, PartialEq, EncodeLabelSet, Debug)]
+pub struct FrameDropLabel {
+ pub variant: &'static str,
+ pub reason: &'static str,
+}
+
+/// Variant labels used in `frame_drops_total`. Exposed as constants to
+/// catch typos at compile time and to keep the cardinality bounded.
+///
+/// `FORWARD_CLIENT_SEND` ticks when the cross-shard client-reply forward
+/// closure fails. Unlike `CONSENSUS` drops (which VSR retransmit
+/// recovers), a `FORWARD_CLIENT_SEND` drop is terminal: the client never
+/// receives the reply and request / response semantics break above the
+/// bus. Operators should alert on the drop-site `tracing` logs (the
+/// counter is not scrape-able yet, see the module doc) and size
+/// `inbox_capacity` for the worst-case cross-shard reply burst.
+/// `FORWARD_REPLICA_SEND` is the symmetric variant for replica forwards;
+/// VSR retransmit covers its loss so it stays informational. `PARTITION`
+/// ticks when a partition-targeted frame cannot be dispatched because the
+/// namespace is absent from the local `ShardsTable`.
+pub mod frame_drop_variant {
+ pub const CONSENSUS: &str = "consensus";
+ pub const FD_TRANSFER: &str = "fd_transfer";
+ pub const PARTITION: &str = "partition";
+ pub const FORWARD_CLIENT_SEND: &str = "forward_client_send";
+ pub const FORWARD_REPLICA_SEND: &str = "forward_replica_send";
+}
+
+/// Reason labels used in `frame_drops_total`.
+///
+/// `UNROUTABLE` ticks when a frame's target shard id has no sender slot
+/// (`target >= senders.len()`). Unreachable while every shard seeds its
+/// `ShardsTable` identically at boot, but `shard_for` returns a stored
+/// `u16` so the index is guarded rather than trusted. `DELIVERY_FAILED`
+/// is the receiver-side equivalent: the frame arrived at the owning shard
+/// but the local registry refused it. `MISROUTED` ticks when the pump
+/// receives a Consensus frame whose target shard is not `self.id`.
+pub mod frame_drop_reason {
+ pub const FULL: &str = "full";
+ pub const DISCONNECTED: &str = "disconnected";
+ pub const UNROUTABLE: &str = "unroutable";
+ pub const DELIVERY_FAILED: &str = "delivery_failed";
+ pub const MISROUTED: &str = "misrouted";
+}
+
+const VARIANT_COUNT: usize = 5;
+const REASON_COUNT: usize = 5;
+
+const VARIANTS: [&str; VARIANT_COUNT] = [
+ frame_drop_variant::CONSENSUS,
+ frame_drop_variant::FD_TRANSFER,
+ frame_drop_variant::PARTITION,
+ frame_drop_variant::FORWARD_CLIENT_SEND,
+ frame_drop_variant::FORWARD_REPLICA_SEND,
+];
+
+const REASONS: [&str; REASON_COUNT] = [
+ frame_drop_reason::FULL,
+ frame_drop_reason::DISCONNECTED,
+ frame_drop_reason::UNROUTABLE,
+ frame_drop_reason::DELIVERY_FAILED,
+ frame_drop_reason::MISROUTED,
+];
+
+fn variant_index(s: &str) -> Option<usize> {
+ VARIANTS.iter().position(|v| *v == s)
+}
+
+fn reason_index(s: &str) -> Option<usize> {
+ REASONS.iter().position(|r| *r == s)
+}
+
+/// Per-shard metric handles.
+///
+/// Cheap to clone (`Arc` of a `Family` under the hood). Each shard owns
+/// one instance produced by [`ShardMetrics::for_shard`]. The
+/// `VARIANT_COUNT * REASON_COUNT` cross product of `Counter`s is minted
+/// at construction so the drop-site hot path never re-enters
+/// `Family::get_or_create` (which acquires a `RwLock` read guard per
+/// drop and stalls under VSR retransmit / drop-burst storms).
+#[derive(Clone)]
+pub struct ShardMetrics {
+ frame_drops_total: Family<FrameDropLabel, Counter>,
+ cached_counters: [[Counter; REASON_COUNT]; VARIANT_COUNT],
+}
+
+impl ShardMetrics {
+ /// Create a metrics handle for a shard. The handle is per-shard by
+ /// virtue of being constructed once per shard; the shard id does not
+ /// appear in the label set (see [`FrameDropLabel`] doc).
+ ///
+ /// All `VARIANT_COUNT * REASON_COUNT` counters are pre-registered
+ /// with the underlying [`Family`] so the drop-site hot path is a
+ /// constant-time array index + atomic increment.
+ #[must_use]
+ pub fn for_shard() -> Self {
+ let frame_drops_total: Family<FrameDropLabel, Counter> = Family::default();
+ let cached_counters = std::array::from_fn(|v_idx| {
+ std::array::from_fn(|r_idx| {
+ frame_drops_total
+ .get_or_create(&FrameDropLabel {
+ variant: VARIANTS[v_idx],
+ reason: REASONS[r_idx],
+ })
+ .clone()
+ })
+ });
+ Self {
+ frame_drops_total,
+ cached_counters,
+ }
+ }
+
+ /// Increment `frame_drops_total{variant, reason}` by 1.
+ ///
+ /// Callers should pass label constants from [`frame_drop_variant`]
+ /// and [`frame_drop_reason`]; those hit the cached counter table.
+ /// Any unknown pair falls back to the `Family::get_or_create` slow
+ /// path so accounting is preserved even if a future caller forgets
+ /// to extend the const tables above.
+ pub fn record_frame_drop(&self, variant: &'static str, reason: &'static str) {
+ if let (Some(v_idx), Some(r_idx)) = (variant_index(variant), reason_index(reason)) {
+ self.cached_counters[v_idx][r_idx].inc();
+ } else {
+ self.frame_drops_total
+ .get_or_create(&FrameDropLabel { variant, reason })
+ .inc();
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn frame_drop_counter_increments_per_label_set() {
+ let metrics = ShardMetrics::for_shard();
+ metrics.record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::FULL);
+ metrics.record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::FULL);
+ metrics.record_frame_drop(
+ frame_drop_variant::CONSENSUS,
+ frame_drop_reason::DISCONNECTED,
+ );
+
+ let count = |variant, reason| {
+ metrics
+ .frame_drops_total
+ .get_or_create(&FrameDropLabel { variant, reason })
+ .get()
+ };
+
+ assert_eq!(
+ count(frame_drop_variant::CONSENSUS, frame_drop_reason::FULL),
+ 2,
+ "two FULL drops land on one label set",
+ );
+ assert_eq!(
+ count(
+ frame_drop_variant::CONSENSUS,
+ frame_drop_reason::DISCONNECTED,
+ ),
+ 1,
+ "a distinct reason gets its own counter",
+ );
+ }
+
+ #[test]
+ fn cached_counter_aliases_family_entry() {
+ // Cached fast-path counters must point at the same underlying
+ // atomic as the `Family`'s get_or_create entry, otherwise a future
+ // scrape exporter would observe zero while the fast path
+ // incremented the cache and the slow path queried the family.
+ let metrics = ShardMetrics::for_shard();
+ for _ in 0..5 {
+ metrics.record_frame_drop(frame_drop_variant::PARTITION, frame_drop_reason::UNROUTABLE);
+ }
+ let from_family = metrics
+ .frame_drops_total
+ .get_or_create(&FrameDropLabel {
+ variant: frame_drop_variant::PARTITION,
+ reason: frame_drop_reason::UNROUTABLE,
+ })
+ .get();
+ assert_eq!(from_family, 5);
+ }
+
+ #[test]
+ fn unknown_label_set_falls_back_to_family() {
+ // A label outside the const tables must still record via the slow
+ // path so we never silently drop accounting. A scrape that filters
+ // on the unknown pair must show the bump.
+ let metrics = ShardMetrics::for_shard();
+ metrics.record_frame_drop("unexpected_variant", "unexpected_reason");
+ let from_family = metrics
+ .frame_drops_total
+ .get_or_create(&FrameDropLabel {
+ variant: "unexpected_variant",
+ reason: "unexpected_reason",
+ })
+ .get();
+ assert_eq!(from_family, 1);
+ }
+}
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index c09ae2c6f..bf188841e 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -15,37 +15,86 @@
// specific language governing permissions and limitations
// under the License.
+use crate::metrics::{frame_drop_reason, frame_drop_variant};
use crate::shards_table::{ShardsTable, calculate_shard_from_consensus_ns};
-use crate::{IggyShard, Receiver, ShardFrame};
+use crate::{IggyShard, LifecycleFrame, Receiver, ShardFrame};
use crossfire::TrySendError;
use futures::FutureExt;
use iggy_binary_protocol::{
- ConsensusError, ConsensusHeader, GenericHeader, Message, MessageBag, PrepareHeader,
+ ConsensusHeader, GenericHeader, Message, MessageBag, Operation, PrepareHeader,
};
-use iggy_common::sharding::IggyNamespace;
+use iggy_common::sharding::{IggyNamespace, METADATA_CONSENSUS_NAMESPACE};
use journal::{Journal, JournalHandle};
use message_bus::{ConnectionInstaller, MessageBus};
use metadata::impls::metadata::StreamsFrontend;
use metadata::stm::StateMachine;
+/// Decompose a [`MessageBag`] into the routing-relevant tuple
+/// `(operation, namespace, generic_message)`.
+///
+/// Single source of truth used by every dispatch entry point so the
+/// operation / namespace extraction never drifts between call sites.
+fn extract_routing(bag: MessageBag) -> (Operation, u64, Message<GenericHeader>) {
+ match bag {
+ MessageBag::Request(r) => {
+ let h = *r.header();
+ (h.operation, h.namespace, r.into_generic())
+ }
+ MessageBag::Prepare(p) => {
+ let h = *p.header();
+ (h.operation, h.namespace, p.into_generic())
+ }
+ MessageBag::PrepareOk(p) => {
+ let h = *p.header();
+ (h.operation, h.namespace, p.into_generic())
+ }
+ MessageBag::StartViewChange(m) => {
+ let h = *m.header();
+ (h.operation(), h.namespace, m.into_generic())
+ }
+ MessageBag::DoViewChange(m) => {
+ let h = *m.header();
+ (h.operation(), h.namespace, m.into_generic())
+ }
+ MessageBag::StartView(m) => {
+ let h = *m.header();
+ (h.operation(), h.namespace, m.into_generic())
+ }
+ MessageBag::Commit(m) => {
+ let h = *m.header();
+ (h.operation(), h.namespace, m.into_generic())
+ }
+ }
+}
+
/// Inter-shard dispatch logic.
///
/// All messages — whether destined for a local or remote shard — are routed
/// through the channel into the target shard's message pump. This ensures
/// that every mutation on a shard is serialized through a single point (the
/// pump), preventing concurrent access from independent async tasks.
-impl<B, MJ, S, M, T, R> IggyShard<B, MJ, S, M, T, R>
+impl<B, MJ, S, M, T> IggyShard<B, MJ, S, M, T>
where
B: MessageBus + ConnectionInstaller,
T: ShardsTable,
- R: Send + 'static,
{
- /// Classify a raw network message and route it to
- /// the correct shard's message pump.
- ///
- /// Decomposes the generic message into its typed form (Request, Prepare,
- /// or `PrepareOk`) to access the operation and namespace, then resolves
- /// the target shard and enqueues the message via its channel sender.
+ /// Network-receive entry point. Classifies the raw
+ /// `Message<GenericHeader>` and routes it to the owning shard via
+ /// `route_typed`.
+ //
+ // TODO(hubcio): perf - this `MessageBag::try_from` is run twice per
+ // consensus frame: once here to extract (operation, namespace) for
+ // routing, and a second time on the receiving shard inside `on_message`
+ // (lib.rs ~560) to dispatch to the correct on_* handler. The second
+ // parse re-runs `bytemuck::checked::try_from_bytes` + per-header
+ // `validate()` on bytes already validated upstream. Measured ~50 ns/
+ // frame; at 1M ops/sec/shard ~ 50 ms/sec/shard of pure re-validation.
+ //
+ // Fix: thread the classified bag through `ShardFrame::Consensus` (carry
+ // `MessageBag` instead of `Message<GenericHeader>`) so the inbox path
+ // matches directly with no second parse. Consensus variant grows from
+ // ~24 B to ~32 B, but `ShardFrame` total stays at 160 B (LifecycleFrame
+ // drives the union size).
pub fn dispatch(&self, message: Message<GenericHeader>) {
let bag = match MessageBag::try_from(message) {
Ok(bag) => bag,
@@ -54,42 +103,55 @@ where
return;
}
};
- let (operation, namespace, generic) = match bag {
- MessageBag::Request(r) => {
- let h = *r.header();
- (h.operation, h.namespace, r.into_generic())
- }
- MessageBag::Prepare(p) => {
- let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
- }
- MessageBag::PrepareOk(p) => {
- let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
- }
- MessageBag::StartViewChange(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::DoViewChange(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::StartView(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::Commit(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- };
- let raw_namespace = namespace;
- let partition_namespace = IggyNamespace::from_raw(raw_namespace);
- let target = if operation.is_metadata() {
- 0
- } else if operation.is_partition() {
+ let (operation, namespace, generic) = extract_routing(bag);
+ self.route_typed(operation, namespace, generic);
+ }
+
+ /// Route a consensus-control message (`StartViewChange`, `DoViewChange`,
+ /// `StartView`, `Commit`) by hashing its raw `u64` consensus namespace.
+ /// Every node uses the same hash so the shard owning the consensus
+ /// group is deterministic across the cluster.
+ pub(crate) fn route_consensus_control(
+ &self,
+ message: Message<GenericHeader>,
+ namespace_u64: u64,
+ operation: Operation,
+ ) {
+ let target = calculate_shard_from_consensus_ns(namespace_u64, self.shard_count);
+ self.try_send_to_target(target, message, operation);
+ }
+
+ /// Branch on operation class to pick the right routing rule. Used by
+ /// [`Self::dispatch`].
+ ///
+ /// Three classes:
+ /// 1. `is_metadata`: metadata data-plane operation, owned by shard 0.
+ /// 2. `is_partition`: partition data-plane operation, owned by the
+ /// shard that the `IggyNamespace -> shard_id` table assigns.
+ /// 3. `is_vsr_reserved` (`Reserved` / `Register`): VSR control-plane
+ /// frame (`StartViewChange`, `DoViewChange`, `StartView`, `Commit`)
+ /// or a client `Register` request. The owning consensus group is
+ /// identified by `namespace_u64`:
+ /// - `METADATA_CONSENSUS_NAMESPACE` -> shard 0.
+ /// - packable `IggyNamespace::inner()` -> the shard owning that
+ /// partition's consensus group.
+ fn route_typed(
+ &self,
+ operation: Operation,
+ namespace_u64: u64,
+ generic: Message<GenericHeader>,
+ ) {
+ if operation.is_metadata() {
+ self.try_send_to_target(0, generic, operation);
+ return;
+ }
+ if operation.is_partition() {
+ let partition_namespace = IggyNamespace::from_raw(namespace_u64);
let Some(target) = self.shards_table.shard_for(partition_namespace) else {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::PARTITION,
+ frame_drop_reason::UNROUTABLE,
+ );
tracing::error!(
shard = self.id,
stream = partition_namespace.stream_id(),
@@ -99,144 +161,76 @@ where
);
return;
};
- target
- } else {
- // View-change / Commit messages carry an opaque u64 consensus
- // namespace (not an `IggyNamespace`). Hash it with the same
- // function the partition-plane lookup table uses so the
- // shard owning the consensus group is deterministically the
- // same across every node. Single-shard deployments trivially
- // collapse to 0.
- #[allow(clippy::cast_lossless)]
- let shard_count = u32::try_from(self.senders.len()).unwrap_or(u32::MAX);
- calculate_shard_from_consensus_ns(raw_namespace, shard_count)
- };
- // `senders[target]` is a `crossfire::MTx`, which in compio is
- // running on an io_uring reactor: a blocking `send` on a full
- // inbox would park the reactor thread and stall every other
- // connection on the shard. Drop instead - consensus recovers via
- // WAL retransmit or a view change.
- match self.senders[target as usize].try_send(ShardFrame::fire_and_forget(generic)) {
- Ok(()) => {}
- Err(TrySendError::Full(_)) => {
- tracing::warn!(
- shard = self.id,
- target,
- ?operation,
- "dispatch: shard inbox full, message dropped"
- );
- }
- Err(TrySendError::Disconnected(_)) => {
- tracing::warn!(
- shard = self.id,
- target,
- ?operation,
- "dispatch: shard inbox closed, message dropped"
- );
- }
+ self.try_send_to_target(target, generic, operation);
+ return;
+ }
+ debug_assert!(
+ operation.is_vsr_reserved(),
+ "route_typed: operation {operation:?} fell through unclassified; \
+ expected is_metadata / is_partition / is_vsr_reserved"
+ );
+ if namespace_u64 == METADATA_CONSENSUS_NAMESPACE {
+ self.try_send_to_target(0, generic, operation);
+ return;
}
+ self.route_consensus_control(generic, namespace_u64, operation);
}
- /// Dispatch a message and return a receiver that resolves when the target
- /// shard has finished processing it.
- ///
- /// # Errors
- /// Returns `ConsensusError` if the message cannot be routed.
- pub fn dispatch_request(
+ /// Send `message` into `senders[target]`. Honors the `io_uring` reactor
+ /// constraint: never blocks; drops on `Full` / `Disconnected` and
+ /// records the drop in `frame_drops_total{variant=consensus}`. VSR
+ /// retransmit recovers consensus drops. A `target` past the end of
+ /// `senders` (a stored `u16` from `shard_for`, not a trusted index)
+ /// is dropped with `reason=unroutable` rather than panicking.
+ /// Metadata frames always pass `target = 0` here, since `is_metadata`
+ /// operations are owned by shard 0.
+ fn try_send_to_target(
&self,
+ target: u16,
message: Message<GenericHeader>,
- ) -> Result<Receiver<R>, ConsensusError> {
- let bag = MessageBag::try_from(message)?;
- let (operation, namespace, generic) = match bag {
- MessageBag::Request(r) => {
- let h = *r.header();
- (h.operation, h.namespace, r.into_generic())
- }
- MessageBag::Prepare(p) => {
- let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
- }
- MessageBag::PrepareOk(p) => {
- let h = *p.header();
- (h.operation, h.namespace, p.into_generic())
- }
- MessageBag::StartViewChange(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::DoViewChange(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::StartView(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
- MessageBag::Commit(m) => {
- let h = *m.header();
- (h.operation(), h.namespace, m.into_generic())
- }
+ operation: Operation,
+ ) {
+ let Some(sender) = self.senders.get(target as usize) else {
+ self.metrics
+ .record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::UNROUTABLE);
+ tracing::error!(
+ shard = self.id,
+ target,
+ ?operation,
+ "dispatch: target shard id out of range, message dropped"
+ );
+ return;
};
- let raw_namespace = namespace;
- let partition_namespace = IggyNamespace::from_raw(raw_namespace);
-
- // Determine which shard should handle a message given its operation and
- // namespace.
- //
- // - Metadata operations always route to shard 0 (the control plane).
- // - Partition operations route to the shard that owns the namespace,
- // looked up via the [`ShardsTable`].
- // - Consensus control-plane (`StartViewChange`, `DoViewChange`,
- // `StartView`, `Commit`) carries a raw `u64` consensus namespace;
- // route it via a deterministic hash function so every node agrees
- // on the owning shard without consulting the partitions table.
- let target = if operation.is_metadata() {
- 0
- } else if operation.is_partition() {
- self.shards_table
- .shard_for(partition_namespace)
- .ok_or_else(|| {
- ConsensusError::InvalidField(format!(
- "namespace {raw_namespace} is not registered in shards_table"
- ))
- })?
- } else {
- #[allow(clippy::cast_lossless)]
- let shard_count = u32::try_from(self.senders.len()).unwrap_or(u32::MAX);
- calculate_shard_from_consensus_ns(raw_namespace, shard_count)
- };
- // Create a frame and send it to the target shard. Same non-
- // blocking `try_send` reason as `dispatch` above: blocking on a
- // full inbox would park the io_uring reactor.
- //
- // On `Full` / `Disconnected` we drop the frame and let the
- // caller's `rx` observe the disconnect (the `response_sender`
- // inside the frame is dropped together with the frame). Callers
- // see an early error rather than hanging.
- let (frame, rx) = ShardFrame::<R>::with_response(generic);
- match self.senders[target as usize].try_send(frame) {
+ match sender.try_send(ShardFrame::consensus(target, message)) {
Ok(()) => {}
Err(TrySendError::Full(_)) => {
+ self.metrics
+ .record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::FULL);
tracing::warn!(
shard = self.id,
target,
?operation,
- "dispatch_request: shard inbox full, message dropped"
+ "dispatch: shard inbox full, message dropped"
);
}
Err(TrySendError::Disconnected(_)) => {
+ self.metrics.record_frame_drop(
+ frame_drop_variant::CONSENSUS,
+ frame_drop_reason::DISCONNECTED,
+ );
tracing::warn!(
shard = self.id,
target,
?operation,
- "dispatch_request: shard inbox closed, message dropped"
+ "dispatch: shard inbox closed, message dropped"
);
}
}
- Ok(rx)
}
- /// Drain this shard's inbox and process each frame locally.
+ /// Drain this shard's inbox and process each frame locally until the
+ /// `stop` signal fires or the inbox disconnects, then drain any frames
+ /// still queued so in-flight requests still get a response.
#[allow(clippy::future_not_send)]
pub async fn run_message_pump(&self, stop: Receiver<()>)
where
@@ -253,15 +247,20 @@ where
Error = iggy_common::IggyError,
> + StreamsFrontend,
{
- let mut loopback_buf = Vec::new();
+ // Reused across every pump iteration; pre-size to skip the
+ // first-drain reallocation.
+ let mut loopback_buf = Vec::with_capacity(64);
+ let mut namespace_scratch: Vec<IggyNamespace> = Vec::with_capacity(64);
loop {
futures::select! {
_ = stop.recv().fuse() => break,
frame = self.inbox.recv().fuse() => {
match frame {
Ok(frame) => {
- self.process_frame(frame).await;
- self.process_loopback(&mut loopback_buf).await;
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ self.process_loopback(&mut loopback_buf, &mut namespace_scratch).await;
+ }
}
Err(_) => break,
}
@@ -271,13 +270,49 @@ where
// Drain remaining frames so in-flight requests get a response.
while let Ok(frame) = self.inbox.try_recv() {
- self.process_frame(frame).await;
- self.process_loopback(&mut loopback_buf).await;
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ self.process_loopback(&mut loopback_buf, &mut namespace_scratch)
+ .await;
+ }
}
}
+ /// Sanity check at pump entry: every Consensus frame routed through
+ /// [`Self::dispatch`] must land on the shard whose `id` matches the
+ /// `target_shard` the sender stamped on the frame. The ctor
+ /// `assert_sender_ordering` proves `senders[i].shard_id() == i`, but
+ /// only checks the sender vec - it cannot catch a receiver moved to
+ /// the wrong runtime. This check closes that gap. Non-Consensus
+ /// frames are always accepted; a Consensus frame stamped for a
+ /// *different* shard fires a `debug_assert_eq!` in test/debug builds
+ /// and is logged + dropped in release. Returns `true` when the frame
+ /// may be processed, `false` when it must be dropped.
+ fn accept_frame_for_self(&self, frame: &ShardFrame) -> bool {
+ let ShardFrame::Consensus { target_shard, .. } = frame else {
+ return true;
+ };
+ if *target_shard == self.id {
+ return true;
+ }
+ debug_assert_eq!(
+ *target_shard, self.id,
+ "shard {} pump received Consensus frame whose target_shard={target_shard}; \
+ senders/runtime incorrectly bound",
+ self.id
+ );
+ self.metrics
+ .record_frame_drop(frame_drop_variant::CONSENSUS, frame_drop_reason::MISROUTED);
+ tracing::error!(
+ shard = self.id,
+ target_shard,
+ "Consensus frame routed to wrong shard; dropping frame to preserve safety"
+ );
+ false
+ }
+
#[allow(clippy::future_not_send)]
- async fn process_frame(&self, frame: ShardFrame<R>)
+ async fn process_frame(&self, frame: ShardFrame)
where
B: MessageBus,
MJ: JournalHandle,
@@ -292,11 +327,21 @@ where
Error = iggy_common::IggyError,
> + StreamsFrontend,
{
- match frame.payload {
- crate::ShardFramePayload::Consensus(message) => {
+ match frame {
+ ShardFrame::Consensus { message, .. } => {
self.on_message(message).await;
}
- crate::ShardFramePayload::ReplicaConnectionSetup { fd, replica_id } => {
+ ShardFrame::Lifecycle(payload) => self.process_lifecycle(payload).await,
+ }
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn process_lifecycle(&self, payload: LifecycleFrame)
+ where
+ B: MessageBus,
+ {
+ match payload {
+ LifecycleFrame::ReplicaConnectionSetup { fd, replica_id } => {
tracing::info!(
shard = self.id,
replica_id,
@@ -306,7 +351,7 @@ where
self.bus
.install_replica_tcp_fd(fd, replica_id, self.on_replica_message.clone());
}
- crate::ShardFramePayload::ClientConnectionSetup { fd, meta } => {
+ LifecycleFrame::ClientConnectionSetup { fd, meta } => {
tracing::info!(
shard = self.id,
client_id = meta.client_id,
@@ -316,7 +361,7 @@ where
self.bus
.install_client_fd(fd, meta, self.on_client_request.clone());
}
- crate::ShardFramePayload::ClientWsConnectionSetup { fd, meta } => {
+ LifecycleFrame::ClientWsConnectionSetup { fd, meta } => {
tracing::info!(
shard = self.id,
client_id = meta.client_id,
@@ -326,18 +371,7 @@ where
self.bus
.install_client_ws_fd(fd, meta, self.on_client_request.clone());
}
- crate::ShardFramePayload::ReplicaMappingUpdate {
- replica_id,
- owning_shard,
- } => {
- self.bus.set_shard_mapping(replica_id, owning_shard);
- }
- crate::ShardFramePayload::ReplicaMappingClear { replica_id } => {
- self.bus.remove_shard_mapping(replica_id);
- }
- crate::ShardFramePayload::ForwardReplicaSend { replica_id, msg } => {
- // Fast path on the owning shard: re-enter `send_to_replica`
- // which finds the local `BusSender` in the replicas registry.
+ LifecycleFrame::ForwardReplicaSend { replica_id, msg } => {
if let Err(e) = self.bus.send_to_replica(replica_id, msg).await {
tracing::debug!(
shard = self.id,
@@ -347,9 +381,18 @@ where
);
}
}
- crate::ShardFramePayload::ForwardClientSend { client_id, msg } => {
+ LifecycleFrame::ForwardClientSend { client_id, msg } => {
if let Err(e) = self.bus.send_to_client(client_id, msg).await {
- tracing::debug!(
+ // Terminal per `SendError` docs: the client never
+ // receives the reply and request / response semantics
+ // break above the bus. Bump the receiver-side counter
+ // (sender side already bumps in `builder.rs`) and raise
+ // to `warn!` so it surfaces in default operator logs.
+ self.metrics.record_frame_drop(
+ frame_drop_variant::FORWARD_CLIENT_SEND,
+ frame_drop_reason::DELIVERY_FAILED,
+ );
+ tracing::warn!(
shard = self.id,
client_id,
error = ?e,
@@ -357,35 +400,6 @@ where
);
}
}
- crate::ShardFramePayload::ConnectionLost { replica_id } => {
- // Shard 0 is the sole responder; the next reconnect sweep
- // will re-dial the peer. Non-zero shards should not see this
- // variant.
- if self.id == 0 {
- tracing::warn!(replica_id, "shard 0 observed replica connection loss");
- // If a coordinator is attached, let it broadcast the
- // clear AND drop its tracked mapping so the periodic
- // refresh task stops re-broadcasting a dead replica's
- // mapping. Tests that bypass the coordinator fall back
- // to the direct broadcast.
- if let Some(coord) = &self.coordinator {
- coord.forget_mapping(replica_id);
- coord.broadcast_mapping_clear(replica_id);
- } else {
- crate::broadcast_mapping_clear(&self.senders, replica_id);
- }
- } else {
- tracing::warn!(
- shard = self.id,
- replica_id,
- "non-zero shard received ConnectionLost; dropping"
- );
- }
- }
}
- // TODO: once on_message returns an R (e.g. ShardResponse), send it
- // back via frame.response_sender. For now the sender is dropped and
- // the caller's receiver will observe a disconnect.
- drop(frame.response_sender);
}
}
diff --git a/core/shard/src/shards_table.rs b/core/shard/src/shards_table.rs
index 0a0a94064..6829f28cd 100644
--- a/core/shard/src/shards_table.rs
+++ b/core/shard/src/shards_table.rs
@@ -32,7 +32,11 @@ pub trait ShardsTable {
fn shard_for(&self, namespace: IggyNamespace) -> Option<u16>;
}
-/// No-op shards table for single-shard setups.
+/// Always-`None` impl. Satisfies the trait for test / simulator paths
+/// that never route via the table (e.g. [`crate::IggyShard::without_inbox`]).
+/// Do not use in production: the router drops any partition frame whose
+/// `shard_for` returns `None`, so a real cluster wired to this impl would
+/// silently shed every partition request.
impl ShardsTable for () {
fn shard_for(&self, _namespace: IggyNamespace) -> Option<u16> {
None
diff --git a/core/simulator/src/bus.rs b/core/simulator/src/bus.rs
index d185e77be..5e6769b77 100644
--- a/core/simulator/src/bus.rs
+++ b/core/simulator/src/bus.rs
@@ -143,6 +143,10 @@ impl MessageBus for SimOutbox {
Ok(())
}
+
+ fn set_connection_lost_fn(&self, _f: message_bus::ConnectionLostFn) {}
+ fn set_replica_forward_fn(&self, _f: message_bus::ReplicaForwardFn) {}
+ fn set_client_forward_fn(&self, _f: message_bus::ClientForwardFn) {}
}
/// Newtype wrapper for shared [`SimOutbox`] that implements [`MessageBus`]
@@ -172,4 +176,16 @@ impl MessageBus for SharedSimOutbox {
) -> Result<(), SendError> {
self.0.send_to_replica(replica, data).await
}
+
+ fn set_connection_lost_fn(&self, f: message_bus::ConnectionLostFn) {
+ self.0.set_connection_lost_fn(f);
+ }
+
+ fn set_replica_forward_fn(&self, f: message_bus::ReplicaForwardFn) {
+ self.0.set_replica_forward_fn(f);
+ }
+
+ fn set_client_forward_fn(&self, f: message_bus::ClientForwardFn) {
+ self.0.set_client_forward_fn(f);
+ }
}
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 51c97ce37..28d777a5f 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -325,8 +325,10 @@ impl Simulator {
futures::executor::block_on(replica.on_message(message));
let mut buf = Vec::new();
- futures::executor::block_on(replica.process_loopback(&mut buf));
- let loopback_count = futures::executor::block_on(replica.process_loopback(&mut buf));
+ let mut ns_scratch: Vec<IggyNamespace> = Vec::new();
+ futures::executor::block_on(replica.process_loopback(&mut buf, &mut ns_scratch));
+ let loopback_count =
+ futures::executor::block_on(replica.process_loopback(&mut buf, &mut ns_scratch));
debug_assert_eq!(
loopback_count, 0,
"on_ack must not re-enqueue loopback messages"