From dc6d34c524368928b5c70499369662ef02c970e8 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Tue, 18 Aug 2026 11:39:48 +0200 Subject: [PATCH 1/3] refactor(consensus): single restore constructor + wedge/replay/ack fixes --- core/configs/src/server_config/cluster.rs | 43 ++++ core/configs/src/server_config/defaults.rs | 5 + core/consensus/src/fatal.rs | 4 + core/consensus/src/impls.rs | 110 ++++++++++ core/consensus/src/plane_helpers.rs | 17 +- core/metadata/src/impls/metadata.rs | 12 +- core/metadata/src/impls/recovery.rs | 38 +++- core/partitions/src/iggy_partition.rs | 4 +- core/server/config.toml | 9 + core/server/src/bootstrap.rs | 243 ++++++++++++--------- core/server/src/partition_helpers.rs | 82 +++---- core/shard/src/lib.rs | 64 +++++- 12 files changed, 455 insertions(+), 176 deletions(-) diff --git a/core/configs/src/server_config/cluster.rs b/core/configs/src/server_config/cluster.rs index ad767c6b20..a1874a6406 100644 --- a/core/configs/src/server_config/cluster.rs +++ b/core/configs/src/server_config/cluster.rs @@ -50,6 +50,10 @@ const MIN_HEARTBEAT_TO_COMMIT_BROADCAST_RATIO: u32 = 4; /// than escalating a progressing view change into a fresh cluster-wide election. const MIN_STATUS_TO_RETRANSMIT_RATIO: u32 = 4; +/// Floor for a nonzero `superblock_wedged_fatal_timeout`: a shorter window +/// would fail-stop the process on a transient disk hiccup instead of a wedge. +const MIN_SUPERBLOCK_WEDGED_FATAL_TIMEOUT: Duration = Duration::from_secs(30); + /// Default recovering-replica probe-attempt ceiling. Duplicated here rather /// than imported so `core/configs` keeps off a build-time edge onto /// `core/consensus` (mirroring [`super::partition`]); `core/server`'s @@ -172,6 +176,16 @@ fn default_repair_chunk_max() -> usize { SERVER_CONFIG.cluster.repair_chunk_max as usize } +/// serde fallback for configs written before the field existed; the value +/// itself lives in `core/server/config.toml` like every other default. +fn default_superblock_wedged_fatal_timeout() -> IggyDuration { + SERVER_CONFIG + .cluster + .superblock_wedged_fatal_timeout + .parse() + .unwrap() +} + #[serde_as] #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] #[serde(deny_unknown_fields)] @@ -262,6 +276,17 @@ pub struct ClusterConfig { /// be > 0 and <= `MAX_REPAIR_CHUNK_MAX`. #[serde(default = "default_repair_chunk_max")] pub repair_chunk_max: usize, + /// How long the metadata superblock may stay unwritable before the replica + /// fail-stops. While wedged the replica is already fenced quorum-invisible + /// and peers elect around it; this converts the log-only limp into a + /// distinct exit status a supervisor can act on. Zero (and the `0` / + /// `disabled` / `unlimited` sentinels, which all parse to zero) disables + /// the fail-stop; nonzero values below + /// `MIN_SUPERBLOCK_WEDGED_FATAL_TIMEOUT` are rejected at boot. + #[serde(default = "default_superblock_wedged_fatal_timeout")] + #[serde_as(as = "DisplayFromStr")] + #[config_env(leaf)] + pub superblock_wedged_fatal_timeout: IggyDuration, /// Full roster of cluster members. Intended to be byte-identical across /// every node so operators ship one config. The running node's identity /// is supplied out-of-band via the `--replica-id` CLI flag, which @@ -747,6 +772,22 @@ impl Validatable for ClusterConfig { return Err(ConfigurationError::InvalidConfigurationValue); } + // Also ahead of the enabled gate: a solo replica persists the + // superblock too, so the fail-stop window applies regardless. + let superblock_fatal_window = self.superblock_wedged_fatal_timeout.get_duration(); + if !superblock_fatal_window.is_zero() + && superblock_fatal_window < MIN_SUPERBLOCK_WEDGED_FATAL_TIMEOUT + { + eprintln!( + "Invalid cluster configuration: cluster.superblock_wedged_fatal_timeout '{}' must \ + be zero (disabled) or at least {}s (a shorter window fail-stops the process on a \ + transient disk hiccup)", + self.superblock_wedged_fatal_timeout, + MIN_SUPERBLOCK_WEDGED_FATAL_TIMEOUT.as_secs() + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } + if !self.enabled { return Ok(()); } @@ -1402,6 +1443,7 @@ mod tests { view_probe_attempts_max: default_view_probe_attempts_max(), repair_retry_interval: default_repair_retry_interval(), repair_chunk_max: default_repair_chunk_max(), + superblock_wedged_fatal_timeout: default_superblock_wedged_fatal_timeout(), nodes: Vec::new(), auth: ClusterAuthConfig { enabled: true, @@ -1737,6 +1779,7 @@ mod cluster_validate_tests { view_probe_attempts_max: default_view_probe_attempts_max(), repair_retry_interval: default_repair_retry_interval(), repair_chunk_max: default_repair_chunk_max(), + superblock_wedged_fatal_timeout: default_superblock_wedged_fatal_timeout(), nodes, auth: ClusterAuthConfig::default(), tls: ClusterTlsConfig::default(), diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs index 7238769b51..1705e861ba 100644 --- a/core/configs/src/server_config/defaults.rs +++ b/core/configs/src/server_config/defaults.rs @@ -92,6 +92,11 @@ impl Default for ClusterConfig { .view_change_status_timeout .parse() .unwrap(), + superblock_wedged_fatal_timeout: SERVER_CONFIG + .cluster + .superblock_wedged_fatal_timeout + .parse() + .unwrap(), request_start_view_retransmit_interval: SERVER_CONFIG .cluster .request_start_view_retransmit_interval diff --git a/core/consensus/src/fatal.rs b/core/consensus/src/fatal.rs index d089834112..341bda389a 100644 --- a/core/consensus/src/fatal.rs +++ b/core/consensus/src/fatal.rs @@ -26,6 +26,10 @@ pub enum FatalReason { /// back. The durable log is intact up to the previous op, so recovery re-derives /// the frontier and restarting is the repair. UnreconcilableLogFrontier = 2, + /// The superblock stayed unwritable past the configured fail-stop window. + /// The replica was already fenced quorum-invisible, so exiting hands the + /// wedge to a supervisor instead of a log reader. + SuperblockWedged = 3, } impl FatalReason { diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 50bb8523d2..33838ee636 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -1113,6 +1113,49 @@ where clock: ConsensusClock, } +/// Boot-time timer set for a consensus group, one struct so every plane's +/// restore path applies the same values in the same place. +#[derive(Debug, Clone, Copy)] +pub struct ConsensusTimers { + pub normal_heartbeat_ticks: u64, + pub commit_message_ticks: u64, + pub prepare_ticks: u64, + pub view_change_retransmit_ticks: u64, + pub view_change_status_ticks: u64, + pub request_start_view_ticks: u64, + pub probe_attempts_max: u32, +} + +/// How a restored replica joins its group. +#[derive(Debug, Clone, Copy)] +pub enum JoinMode { + /// Fresh group or solo replica: plain init; the group needs its view-0 + /// primary to exist. + Init, + /// Prior life detected: join quorum-invisible and probe for the current + /// view (`RequestStartView`) instead of resuming a role the cluster may + /// have elected past. + ProbeAsBackup { + /// Also await a state-transfer offer before serving, replacing + /// snapshot-shaped state from the live primary. + await_state_transfer: bool, + }, +} + +/// Restored state handed to [`VsrConsensus::restored`]. +#[derive(Debug, Clone, Copy)] +pub struct VsrRestore<'a> { + pub timers: &'a ConsensusTimers, + /// `(view, log_view)` read back from the group's durable superblock. + pub durable_view: Option<(u32, u32)>, + /// View inferred from the last journaled prepare, consulted only when no + /// durable record exists; `log_view` cannot be inferred and stays 0. + pub view_fallback: Option, + /// Non-zero boot incarnation; `None` keeps the default. + pub incarnation: Option, + pub join: JoinMode, +} + impl> VsrConsensus { /// # Panics /// - If `replica >= replica_count`. @@ -1136,6 +1179,73 @@ impl> VsrConsensus { ) } + /// Restore constructor: the one ordered boot path for every plane, so the + /// metadata and partition planes cannot diverge in restore order. Timers + /// first, then the durable view, then role selection - a probe must never + /// advertise a view older than the recorded one. + /// + /// # Panics + /// - If `replica >= replica_count`. + /// - If `replica_count < 1`. + pub fn restored( + cluster: u128, + replica: u8, + replica_count: u8, + group: u64, + message_bus: B, + pipeline: P, + restore: VsrRestore<'_>, + ) -> Self { + let mut consensus = Self::new( + cluster, + replica, + replica_count, + group, + message_bus, + pipeline, + ); + let timers = restore.timers; + consensus.set_normal_heartbeat_ticks(timers.normal_heartbeat_ticks); + consensus.set_commit_message_ticks(timers.commit_message_ticks); + consensus.set_prepare_ticks(timers.prepare_ticks); + consensus.set_view_change_retransmit_ticks(timers.view_change_retransmit_ticks); + consensus.set_view_change_status_ticks(timers.view_change_status_ticks); + consensus.set_request_start_view_ticks(timers.request_start_view_ticks); + consensus.set_probe_attempts_max(timers.probe_attempts_max); + if let Some(incarnation) = restore.incarnation { + consensus.set_incarnation(incarnation); + } + if let Some((view, log_view)) = restore.durable_view { + // The one line proving the durable record was READ BACK, not merely + // written: a replica that came back at view 0 is otherwise + // indistinguishable from one that resumed correctly until it votes. + tracing::info!( + group, + view, + log_view, + "restored group view from its superblock" + ); + consensus.set_view(view); + consensus.set_log_view(log_view); + consensus.mark_superblock_durable(view, log_view); + } else if let Some(view) = restore.view_fallback { + consensus.set_view(view); + } + match restore.join { + JoinMode::Init => consensus.init(), + JoinMode::ProbeAsBackup { + await_state_transfer, + } => { + consensus.init_as_backup(); + consensus.begin_view_probe(); + if await_state_transfer { + consensus.begin_state_transfer_await(); + } + } + } + consensus + } + /// [`Self::new`] with an explicit time source. Simulator and clock /// tests only; production wiring stays on the system-clock default. /// diff --git a/core/consensus/src/plane_helpers.rs b/core/consensus/src/plane_helpers.rs index 57c6d296a2..4e4bcd7efd 100644 --- a/core/consensus/src/plane_helpers.rs +++ b/core/consensus/src/plane_helpers.rs @@ -771,7 +771,14 @@ pub fn panic_if_hash_chain_would_break_in_same_view( } } -// TODO: Figure out how to make this check the journal if it contains the prepare. +/// Ack a prepare back to its primary once the owning plane vouches for it. +/// +/// `is_persisted` is the caller's journal-containment verdict for `header`: +/// consensus is sans-io and cannot consult the journal itself, so the plane +/// that owns the journal must vouch that this exact prepare is durable before +/// the ack leaves. `false` withholds the ack; the primary's retransmit +/// re-drives it once a later persist succeeds. +/// /// # Panics /// - If `header.command` is not `Command::Prepare`. /// - If `header.view > consensus.view()`. @@ -779,7 +786,7 @@ pub fn panic_if_hash_chain_would_break_in_same_view( pub async fn send_prepare_ok( consensus: &VsrConsensus, header: &PrepareHeader, - is_persisted: Option, + is_persisted: bool, ) where B: MessageBus, P: Pipeline, @@ -794,7 +801,7 @@ pub async fn send_prepare_ok( return; } - if is_persisted == Some(false) { + if !is_persisted { return; } @@ -1227,7 +1234,7 @@ mod tests { ..Default::default() }; - futures::executor::block_on(send_prepare_ok(&consensus, &prepare_header, Some(true))); + futures::executor::block_on(send_prepare_ok(&consensus, &prepare_header, true)); let mut buf = Vec::new(); consensus.drain_loopback_into(&mut buf); @@ -1874,7 +1881,7 @@ mod tests { ..Default::default() }; - futures::executor::block_on(send_prepare_ok(&consensus, &prepare_header, Some(true))); + futures::executor::block_on(send_prepare_ok(&consensus, &prepare_header, true)); let mut buf = Vec::new(); consensus.drain_loopback_into(&mut buf); diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index db84be4af1..ee67e1097c 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -3548,9 +3548,17 @@ where if !self.persist_superblock_if_needed(consensus).await { return; } + // Containment, not occupancy: after a refused append (slot collision, + // view-change race) the slot can hold a DIFFERENT prepare at this op, + // and acking it would vouch durability for bytes this replica never + // journaled. Checksum equality withholds the ack; the primary's + // retransmit re-drives it once the right prepare lands. let journal = self.journal.as_ref().unwrap(); - let persisted = journal.handle().header(header.op as usize).is_some(); - send_prepare_ok_common(consensus, header, Some(persisted)).await; + let persisted = journal + .handle() + .header(header.op as usize) + .is_some_and(|stored| stored.checksum == header.checksum); + send_prepare_ok_common(consensus, header, persisted).await; } } diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs index f192d38f72..afdec53b93 100644 --- a/core/metadata/src/impls/recovery.rs +++ b/core/metadata/src/impls/recovery.rs @@ -358,6 +358,7 @@ pub async fn recover( journal_slots: usize, clients_table_max: usize, seed_baseline: impl FnOnce(&M), + on_replayed_logout: impl Fn(&M, u128, iggy_common::IggyTimestamp), ) -> Result, RecoveryError> where M: StateMachine, Error = IggyError> @@ -618,10 +619,15 @@ where } if header.operation == Operation::Logout { client_table.remove_client(header.client); - // TODO: the commit paths also run `remove_consumer_group_member` - // here; recovery has no `StreamsFrontend` bound, so replayed - // logouts leave stale group members (pre-existing, harmless for - // dead connections but a divergence from the live apply). + // Logout's only state-machine effect, mirrored from the live + // commit path: the caller drops the client from its consumer + // groups (`remove_consumer_group_member`) so replay and live + // apply converge on the same group membership. + on_replayed_logout( + &mux_stm, + header.client, + iggy_common::IggyTimestamp::from(header.timestamp), + ); last_applied_op = Some(header.op); continue; } @@ -956,6 +962,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -981,6 +988,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1016,6 +1024,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1073,6 +1082,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1132,6 +1142,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await; @@ -1182,6 +1193,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1219,6 +1231,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1288,6 +1301,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1363,6 +1377,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1428,6 +1443,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1477,12 +1493,14 @@ mod tests { journal.storage_ref().fsync().await.unwrap(); } + let replayed_logouts = std::cell::RefCell::new(Vec::new()); let recovered = recover::( dir.path(), SOLO, journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, client, _| replayed_logouts.borrow_mut().push(client), ) .await .unwrap(); @@ -1491,6 +1509,11 @@ mod tests { None, "logged-out session must not be resurrected" ); + assert_eq!( + replayed_logouts.into_inner(), + vec![CLIENT], + "replay must hand the logged-out client to the group-removal hook" + ); assert_eq!(recovered.last_applied_op, Some(2)); } @@ -1545,6 +1568,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await; assert!(matches!( @@ -1577,6 +1601,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await; assert!(matches!( @@ -1608,6 +1633,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1660,6 +1686,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await; match result { @@ -1700,6 +1727,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await { @@ -1764,6 +1792,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); @@ -1816,6 +1845,7 @@ mod tests { journal::prepare_journal::DEFAULT_SLOT_COUNT, CLIENTS_TABLE_MAX, |_| {}, + |_, _, _| {}, ) .await .unwrap(); diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 76267af573..4a366076dd 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -4717,7 +4717,9 @@ where // consumer-offset ops (via `apply_replicated_operation`) append // to that journal before `send_prepare_ok` fires, so every op // that reaches here is journal-backed and ACKs as durable. - send_prepare_ok_common(self.consensus(), header, Some(true)).await; + // (`header_by_op` is a linear scan, so re-proving that here would + // put O(journal) on every ack; the call-order invariant stands in.) + send_prepare_ok_common(self.consensus(), header, true).await; } } diff --git a/core/server/config.toml b/core/server/config.toml index d10bfa4e04..f420aeeb30 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -559,6 +559,15 @@ repair_retry_interval = "1s" # the queue and drops frames. Must be > 0 and <= 1024. repair_chunk_max = 128 +# How long the metadata superblock may stay unwritable before the replica +# fail-stops (duration). A replica that cannot persist its view is already +# fenced quorum-invisible and retries with capped backoff; past this window the +# process exits with a distinct status so a supervisor restarts or replaces it +# instead of an operator finding the wedge in logs. "0" disables the fail-stop +# and leaves the replica fenced indefinitely. Nonzero values must be at least +# 30s so a transient disk hiccup cannot kill the process. +superblock_wedged_fatal_timeout = "2m" + # Replica-to-replica authentication (PSK + BLAKE3 keyed-MAC handshake). [cluster.auth] # When true, every replica peer must complete the authenticated handshake or be diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index a78cc847db..01ef2d1fab 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -26,7 +26,7 @@ use crate::dispatch::{ use crate::http; use crate::partition_helpers::{ build_partition_fresh, configure_consumer_offsets, ensure_initial_segment, - open_partition_superblock, restore_partition_view, + open_partition_superblock, }; use crate::segment_recovery::{RecoveredSegment, load_persisted_segments}; use crate::server_error::{ServerError, ShardJoinFailure, ShardJoinFailureKind}; @@ -37,8 +37,8 @@ use configs::sharding::{ INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX, }; use consensus::{ - ClientTable, LocalPipeline, MetadataHandle, PartitionsHandle, PipelineEntry, Sequencer, - VsrConsensus, + ClientTable, ConsensusTimers, JoinMode, LocalPipeline, MetadataHandle, PartitionsHandle, + PipelineEntry, Sequencer, VsrConsensus, VsrRestore, }; // `try_send` / `try_recv` resolve through these traits on `MAsyncTx` / // `MAsyncRx`; the metadata-handoff loops below depend on the @@ -1016,6 +1016,11 @@ async fn shard_main( |mux_stm| { ensure_default_root_user(mux_stm); }, + |mux_stm, client, timestamp| { + mux_stm + .streams() + .remove_consumer_group_member(client, timestamp); + }, ) .await .map_err(ServerError::MetadataRecovery)?; @@ -2012,6 +2017,7 @@ async fn build_shard_for_thread( // Repair pacing is shared by both planes' repair loops, so it is a // per-shard tunable set once here rather than per consensus group. shard.set_repair_retry_ticks(repair_retry_ticks(config)); + shard.set_superblock_wedged_fatal_failures(superblock_wedged_fatal_failures(config)); shard.set_served_segment_cache_bytes_max( config .partition @@ -2093,6 +2099,28 @@ fn duration_to_ticks(interval: Duration) -> u64 { u64::try_from(ticks.max(1)).unwrap_or(u64::MAX) } +/// `[cluster] superblock_wedged_fatal_timeout` as a consecutive-failure count. +/// Retries pin at the backoff cap after warmup, so the window divided by +/// [`journal::superblock::SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS`] bounds how +/// long a wedged replica may limp before it fail-stops. Zero stays zero +/// (fail-stop disabled). +pub(crate) fn superblock_wedged_fatal_failures(config: &ServerConfig) -> u64 { + superblock_window_to_failures( + config + .cluster + .superblock_wedged_fatal_timeout + .get_duration(), + ) +} + +fn superblock_window_to_failures(window: Duration) -> u64 { + if window.is_zero() { + return 0; + } + let cap_micros = u128::from(journal::superblock::SUPERBLOCK_RETRY_BACKOFF_MAX_MICROS); + u64::try_from((window.as_micros() / cap_micros).max(1)).unwrap_or(u64::MAX) +} + /// `[cluster] heartbeat_timeout` in consensus ticks. Every consensus group /// (metadata and per-partition planes alike) gets the same window: the failure /// it guards against - a primary that stopped heartbeating - is host-level, not @@ -2181,6 +2209,20 @@ pub(crate) fn request_start_view_ticks(config: &ServerConfig) -> u64 { ) } +/// The full `[cluster]` timer set every consensus group boots with, built +/// once so the planes cannot diverge in what they apply. +pub(crate) fn consensus_timers(config: &ServerConfig) -> ConsensusTimers { + ConsensusTimers { + normal_heartbeat_ticks: cluster_heartbeat_ticks(config), + commit_message_ticks: commit_broadcast_ticks(config), + prepare_ticks: prepare_retransmit_ticks(config), + view_change_retransmit_ticks: view_change_retransmit_ticks(config), + view_change_status_ticks: view_change_status_ticks(config), + request_start_view_ticks: request_start_view_ticks(config), + probe_attempts_max: config.cluster.view_probe_attempts_max, + } +} + /// `[cluster] repair_retry_interval` in consensus ticks: how long a stalled /// journal-repair stream waits before re-requesting its window. Both planes' /// repair loops share it, so it is applied once per shard (not per consensus @@ -2235,50 +2277,10 @@ fn restore_metadata_consensus( ); let prepare_queue_depth = config.metadata.prepare_queue_depth; - let mut consensus = VsrConsensus::new( - topology.cluster_id, - topology.self_replica_id, - replica_count, - server_common::sharding::METADATA_GROUP, - bus, - // Request queue keeps the stock 2x ratio over the prepare queue - // (32 -> 64 at defaults): buffered requests are cheap relative to - // in-flight prepares and drain as prepares commit. - LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), - ); - consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config)); - consensus.set_commit_message_ticks(commit_broadcast_ticks(config)); - consensus.set_prepare_ticks(prepare_retransmit_ticks(config)); - consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks(config)); - consensus.set_view_change_status_ticks(view_change_status_ticks(config)); - consensus.set_request_start_view_ticks(request_start_view_ticks(config)); - consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max); - // Fresh random incarnation each boot, so a StartView addressed to a previous - // incarnation still in flight is ignored (`handle_start_view` guard). `| 1` - // guarantees the non-zero the guard treats as set. The deterministic simulator - // overrides this with a seed-derived value bumped per restart. - consensus.set_incarnation(rand::random::() | 1); - let last_header = journal .last_op() .and_then(|op| usize::try_from(op).ok()) .and_then(|op| journal.header(op).map(|header| *header)); - // View and log_view come from the durable superblock when present. A present but - // unreadable superblock already refused boot in `recover()`, so reaching the - // `else` means it is genuinely absent: a fresh node, or one that took writes but - // never checkpointed or changed view. There, inferring the view from the last WAL - // prepare is safe, since the persist-before-send gate guarantees this replica - // never externalized a view beyond what a re-probe re-derives, and it re-probes - // as a backup below. log_view cannot be inferred and stays 0 until the next - // superblock write. - if let Some(state) = recovered_state { - consensus.set_view(state.view); - consensus.set_log_view(state.log_view); - consensus.mark_superblock_durable(state.view, state.log_view); - } else if let Some(header) = last_header { - consensus.set_view(header.view); - } - // On a RESTART in a cluster, rejoin as a quorum-invisible backup and // probe for the current view (`RequestStartView`): the view's primary // answers with a `StartView`, the replica adopts it as a backup, and @@ -2295,18 +2297,51 @@ fn restore_metadata_consensus( // and an empty journal; gating on the WAL alone would `init()` it into // `Status::Normal` as primary for a view the cluster may have moved past, // with `ceded_primaryship` false and no probe to correct it. - if replica_count > 1 && (restored_op > 0 || recovered_state.is_some()) { - consensus.init_as_backup(); - consensus.begin_view_probe(); - // Restart in a cluster: replace snapshot-shaped metadata state - // (snapshot + client table) from the live primary the probe finds, - // then journal-repair the tail. If the probe exhausts instead -- - // full-cluster bootstrap, nobody live to fetch from -- the election - // fallback clears the stage and this local recovery stands. - consensus.begin_state_transfer_await(); + // + // The rejoin also awaits a state transfer: snapshot-shaped metadata state + // (snapshot + client table) is replaced from the live primary the probe + // finds, then journal repair fills the tail. If the probe exhausts + // instead -- full-cluster bootstrap, nobody live to fetch from -- the + // election fallback clears the stage and this local recovery stands. + let join = if replica_count > 1 && (restored_op > 0 || recovered_state.is_some()) { + JoinMode::ProbeAsBackup { + await_state_transfer: true, + } } else { - consensus.init(); - } + JoinMode::Init + }; + let timers = consensus_timers(config); + let consensus = VsrConsensus::restored( + topology.cluster_id, + topology.self_replica_id, + replica_count, + server_common::sharding::METADATA_GROUP, + bus, + // Request queue keeps the stock 2x ratio over the prepare queue + // (32 -> 64 at defaults): buffered requests are cheap relative to + // in-flight prepares and drain as prepares commit. + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), + VsrRestore { + timers: &timers, + // View and log_view come from the durable superblock when present. + // A present but unreadable superblock already refused boot in + // `recover()`, so no durable record means genuinely absent: a + // fresh node, or one that took writes but never checkpointed or + // changed view. There, inferring the view from the last WAL + // prepare is safe, since the persist-before-send gate guarantees + // this replica never externalized a view beyond what a re-probe + // re-derives, and it re-probes as a backup. + durable_view: recovered_state.map(|state| (state.view, state.log_view)), + view_fallback: last_header.map(|header| header.view), + // Fresh random incarnation each boot, so a StartView addressed to + // a previous incarnation still in flight is ignored + // (`handle_start_view` guard). `| 1` guarantees the non-zero the + // guard treats as set. The deterministic simulator overrides this + // with a seed-derived value bumped per restart. + incarnation: Some(rand::random::() | 1), + join, + }, + ); consensus.sequencer().set_sequence(restored_op); // A SOLO replica's durable journal head IS its commit point: quorum is // 1-of-1, so an entry commits the instant it is durable, and the acks @@ -2388,39 +2423,6 @@ fn restore_metadata_consensus( consensus } -#[allow(clippy::too_many_arguments)] -/// Build the ticked-and-bounded consensus a loaded partition group joins -/// with; the fresh-create path configures its own inside -/// `build_partition_fresh`. -fn loaded_partition_consensus( - config: &ServerConfig, - namespace: IggyNamespace, - cluster_id: u128, - self_replica_id: u8, - replica_count: u8, - bus: Rc, -) -> VsrConsensus> { - // Request queue holds 2x the prepare depth (buffered requests drain as - // prepares commit); depth is the per-partition `[partition]` knob. - let prepare_queue_depth = config.partition.prepare_queue_depth; - let consensus = VsrConsensus::new( - cluster_id, - self_replica_id, - replica_count, - namespace.inner(), - bus, - LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), - ); - consensus.set_normal_heartbeat_ticks(cluster_heartbeat_ticks(config)); - consensus.set_commit_message_ticks(commit_broadcast_ticks(config)); - consensus.set_prepare_ticks(prepare_retransmit_ticks(config)); - consensus.set_view_change_retransmit_ticks(view_change_retransmit_ticks(config)); - consensus.set_view_change_status_ticks(view_change_status_ticks(config)); - consensus.set_request_start_view_ticks(request_start_view_ticks(config)); - consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max); - consensus -} - /// Recover this partition's persisted segment chain, stamping each segment /// with the topic's effective segment size (the per-topic value when the /// topic was created with one, else the shard-wide configured size). @@ -2472,20 +2474,9 @@ async fn load_partition( let stream_id = namespace.stream_id(); let topic_id = namespace.topic_id(); let partition_id = namespace.partition_id(); - let mut consensus = loaded_partition_consensus( - config, - namespace, - cluster_id, - self_replica_id, - replica_count, - bus, - ); - // (view, log_view) come from the group's durable superblock when present; // a present but unverifiable record already refused boot inside - // `open_partition_superblock`. Restored BEFORE choosing how to join, so - // the backup probe below never advertises a view older than the recorded - // one. + // `open_partition_superblock`. let partition_dir = config .system .get_partition_path(stream_id, topic_id, partition_id); @@ -2498,9 +2489,6 @@ async fn load_partition( }, ) .await?; - if let Some(state) = recovered_state.as_ref() { - restore_partition_view(&mut consensus, state); - } // A recovered partition lost its journal state with the process: the // partition journal is in-memory and segments carry no op numbers, so @@ -2512,12 +2500,34 @@ async fn load_partition( // at the serving peer's retention point. The probe re-broadcasts on its // timeout, so it needs no live mesh at boot. Single-replica groups // have no peer to ask and keep the plain init. - if replica_count > 1 { - consensus.init_as_backup(); - consensus.begin_view_probe(); + let join = if replica_count > 1 { + JoinMode::ProbeAsBackup { + await_state_transfer: false, + } } else { - consensus.init(); - } + JoinMode::Init + }; + // Request queue holds 2x the prepare depth (buffered requests drain as + // prepares commit); depth is the per-partition `[partition]` knob. + let prepare_queue_depth = config.partition.prepare_queue_depth; + let timers = consensus_timers(config); + let consensus = VsrConsensus::restored( + cluster_id, + self_replica_id, + replica_count, + namespace.inner(), + bus, + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), + VsrRestore { + timers: &timers, + durable_view: recovered_state + .as_ref() + .map(|state| (state.view, state.log_view)), + view_fallback: None, + incarnation: None, + join, + }, + ); // No prepare-timestamp floor is restored here: the partition consensus // journal is non-durable today, so there is no persisted head to observe @@ -3965,6 +3975,25 @@ const fn operation_triggers_partition_reconcile(op: Operation) -> bool { mod tests { use super::*; + #[test] + fn superblock_fatal_window_converts_to_capped_backoff_retries() { + assert_eq!( + superblock_window_to_failures(Duration::ZERO), + 0, + "zero window must stay the disabled sentinel" + ); + assert_eq!( + superblock_window_to_failures(Duration::from_mins(2)), + 120, + "past warmup one retry rides each 1s backoff cap" + ); + assert_eq!( + superblock_window_to_failures(Duration::from_micros(500)), + 1, + "a sub-cap window still needs one failure to fire" + ); + } + #[test] fn fresh_cluster_bootstrap_requires_explicit_root_credentials() { assert!(matches!( diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 8fb4a0e332..820d40f5f4 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -28,7 +28,7 @@ use crate::offset_recovery::{load_consumer_group_offsets, load_consumer_offsets} use crate::server_error::ServerError; use compio::fs::create_dir_all; use configs::server::ServerConfig; -use consensus::{LocalPipeline, VsrConsensus, VsrState}; +use consensus::{JoinMode, LocalPipeline, VsrConsensus, VsrRestore, VsrState}; use iggy_common::{ ConsumerGroupOffsets, ConsumerOffsets, IggyByteSize, IggyError, IggyTimestamp, PartitionStats, TopicRuntimeOptions, @@ -44,7 +44,7 @@ use std::path::{Path, PathBuf}; use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::Ordering; -use tracing::{error, info, warn}; +use tracing::{error, warn}; /// Create the on-disk directory hierarchy for a partition. /// @@ -498,29 +498,6 @@ pub(crate) async fn open_partition_superblock( Ok((Rc::new(superblock), recovered_state)) } -/// Restore `(view, log_view)` from a recovered superblock record and mark -/// them durable (read back from disk, durable by definition). Runs BEFORE -/// `init` / `init_as_backup` so the join path never advertises a view older -/// than the recorded one. -pub(crate) fn restore_partition_view( - consensus: &mut VsrConsensus>, - state: &VsrState, -) { - // The one line proving the durable record was READ BACK, not merely written: - // the group's whole anti-regression guarantee rests on this call running, and - // a replica that came back at view 0 is otherwise indistinguishable from one - // that resumed correctly until it votes. - info!( - namespace_raw = consensus.group(), - view = state.view, - log_view = state.log_view, - "restored partition view from its superblock" - ); - consensus.set_view(state.view); - consensus.set_log_view(state.log_view); - consensus.mark_superblock_durable(state.view, state.log_view); -} - /// Materialise a brand-new [`IggyPartition`] for a namespace that has no on-disk state yet. /// /// Counterpart to bootstrap's `load_partition`, which hydrates from @@ -586,26 +563,6 @@ pub async fn build_partition_fresh( source })?; - // Request queue holds 2x the prepare depth (buffered requests drain as - // prepares commit); depth is the per-partition `[partition]` knob. - let prepare_queue_depth = config.partition.prepare_queue_depth; - let mut consensus = VsrConsensus::new( - cluster_id, - self_replica_id, - replica_count, - namespace.inner(), - bus, - LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), - ); - consensus.set_normal_heartbeat_ticks(crate::bootstrap::cluster_heartbeat_ticks(config)); - consensus.set_commit_message_ticks(crate::bootstrap::commit_broadcast_ticks(config)); - consensus.set_prepare_ticks(crate::bootstrap::prepare_retransmit_ticks(config)); - consensus - .set_view_change_retransmit_ticks(crate::bootstrap::view_change_retransmit_ticks(config)); - consensus.set_view_change_status_ticks(crate::bootstrap::view_change_status_ticks(config)); - consensus.set_request_start_view_ticks(crate::bootstrap::request_start_view_ticks(config)); - consensus.set_probe_attempts_max(config.cluster.view_probe_attempts_max); - // The hierarchy create above guarantees the directory exists; recover this // group's durable (view, log_view) before choosing how to join, so a // restart materialization resumes from the view it last recorded instead @@ -622,9 +579,6 @@ pub async fn build_partition_fresh( }, ) .await?; - if let Some(state) = recovered_state.as_ref() { - restore_partition_view(&mut consensus, state); - } // A partition directory that already holds segment bytes is a RESTART // materialization, not a fresh create: this replica's group state died @@ -635,12 +589,34 @@ pub async fn build_partition_fresh( // peer, byte-identical by the deterministic-roll/replicated-ciphertext // design. A truly fresh create keeps the plain init: every group needs // its view-0 primary to exist. - if restarted { - consensus.init_as_backup(); - consensus.begin_view_probe(); + let join = if restarted { + JoinMode::ProbeAsBackup { + await_state_transfer: false, + } } else { - consensus.init(); - } + JoinMode::Init + }; + // Request queue holds 2x the prepare depth (buffered requests drain as + // prepares commit); depth is the per-partition `[partition]` knob. + let prepare_queue_depth = config.partition.prepare_queue_depth; + let timers = crate::bootstrap::consensus_timers(config); + let consensus = VsrConsensus::restored( + cluster_id, + self_replica_id, + replica_count, + namespace.inner(), + bus, + LocalPipeline::with_capacities(prepare_queue_depth, prepare_queue_depth * 2), + VsrRestore { + timers: &timers, + durable_view: recovered_state + .as_ref() + .map(|state| (state.view, state.log_view)), + view_fallback: None, + incarnation: None, + join, + }, + ); let mut partition = IggyPartition::new(stats, consensus); partition.set_runtime_options(runtime_options); diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 726bcb0176..4e20691b0e 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -29,10 +29,10 @@ pub use router::CONSENSUS_TICK_INTERVAL; use consensus::LocalPipeline; use consensus::{ ChunkProgress, CommitOutcome, Consensus, ConsensusClock, DVC_HEADERS_MAX, DvcHeaderKind, - DvcSuffix, MergedLog, MetadataHandle, MuxPlane, PartitionsHandle, Pipeline, Plane, PlaneKind, - STATE_TRANSFER_MAX_DECODE_RETRIES, STATE_TRANSFER_MAX_STALL_RETRIES, Sequencer, Status, - VsrAction, VsrConsensus, build_deny_reply_from_request_header, dvc_blank, dvc_header_kind, - encode_prepare_headers, restamp_prepare_view, verify_prepare_integrity, + DvcSuffix, FatalReason, MergedLog, MetadataHandle, MuxPlane, PartitionsHandle, Pipeline, Plane, + PlaneKind, STATE_TRANSFER_MAX_DECODE_RETRIES, STATE_TRANSFER_MAX_STALL_RETRIES, Sequencer, + Status, VsrAction, VsrConsensus, build_deny_reply_from_request_header, dvc_blank, + dvc_header_kind, encode_prepare_headers, fatal, restamp_prepare_view, verify_prepare_integrity, }; #[cfg(any(test, feature = "simulator"))] use crossfire::AsyncRxTrait; @@ -1379,6 +1379,12 @@ where /// `[cluster] repair_retry_interval` at bootstrap. repair_retry_ticks: Cell, + /// Consecutive metadata superblock write failures tolerated before the + /// process fail-stops. Defaults to 0 (disabled) so the simulator and tests + /// keep a wedged-but-fenced replica alive; the server arms it from + /// `[cluster] superblock_wedged_fatal_timeout` at bootstrap. + superblock_wedged_fatal_failures: Cell, + /// Live `[partition] transfer_served_cache_bytes_max`: the byte budget for /// segment payloads this shard keeps resident to serve chunk requests. /// Defaults to [`SERVED_SEGMENT_CACHE_BYTES_DEFAULT`]; the server @@ -1525,6 +1531,7 @@ where partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT), repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), + superblock_wedged_fatal_failures: Cell::new(0), bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE), metadata_transfer_attempts: Cell::new(0), metadata_transfer_decode_failures: Cell::new(None), @@ -1538,6 +1545,14 @@ where self.repair_retry_ticks.set(ticks); } + /// Arm the superblock fail-stop bound (consecutive write failures). + /// Called once per shard at bootstrap; the simulator and tests keep the + /// disabled default (0) so a wedged-but-fenced replica stays observable + /// in-process. + pub fn set_superblock_wedged_fatal_failures(&self, failures: u64) { + self.superblock_wedged_fatal_failures.set(failures); + } + /// Override the serving-side resident payload budget from configuration. /// Called once per shard at bootstrap. pub fn set_served_segment_cache_bytes_max(&self, bytes: u64) { @@ -1841,6 +1856,7 @@ where partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT), repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX), repair_retry_ticks: Cell::new(partitions::REPAIR_RETRY_TICKS), + superblock_wedged_fatal_failures: Cell::new(0), bus_max_message_size: Cell::new(DEFAULT_BUS_MAX_MESSAGE_SIZE), metadata_transfer_attempts: Cell::new(0), metadata_transfer_decode_failures: Cell::new(None), @@ -2393,6 +2409,12 @@ const fn parked_footprint(len: usize) -> usize { len.next_multiple_of(MESSAGE_ALIGN) } +/// Whether consecutive superblock write failures crossed the fail-stop bound. +/// `fatal_after == 0` disables the fail-stop. +const fn superblock_wedged(failures: u64, fatal_after: u64) -> bool { + fatal_after != 0 && failures >= fatal_after +} + /// Reconciler passes a frame may survive before it is answered rather than held. /// /// Passes, not seconds, and deliberately not described in seconds: a pass fires @@ -8228,6 +8250,20 @@ where if metadata.persist_superblock_if_needed(consensus).await { dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &wire_actions).await; } + let superblock_failures = metadata.superblock_write_failures(); + if superblock_wedged( + superblock_failures, + self.superblock_wedged_fatal_failures.get(), + ) { + fatal( + FatalReason::SuperblockWedged, + &format!( + "metadata superblock persist failed {superblock_failures} consecutive times, \ + past the [cluster] superblock_wedged_fatal_timeout window; exiting so a \ + supervisor handles the wedge instead of the replica limping fenced" + ), + ); + } // Repair a lost primary self-ack: `RetransmitPrepares` to self is a // no-op, so the timer-driven retransmit above cannot recover the @@ -9769,3 +9805,23 @@ mod control_frame_tests { assert!(body.is_empty()); } } + +#[cfg(test)] +mod superblock_fail_stop_tests { + //! The bound must stay disabled at 0: the simulator asserts a wedged + //! replica survives fenced in-process, and only the server arms it. + + use super::superblock_wedged; + + #[test] + fn zero_bound_never_fires() { + assert!(!superblock_wedged(u64::MAX, 0)); + } + + #[test] + fn bound_fires_at_and_past_the_threshold() { + assert!(!superblock_wedged(119, 120)); + assert!(superblock_wedged(120, 120)); + assert!(superblock_wedged(121, 120)); + } +} From b7c64210a367527aaab660cea13f4e8fa8a58687 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Tue, 18 Aug 2026 12:57:29 +0200 Subject: [PATCH 2/3] refactor(server): consensus restore, fail-stop, linear cleanup batch --- core/configs/src/common/server.rs | 14 +- core/configs/src/server_config/cluster.rs | 37 ++++ core/configs/src/server_config/defaults.rs | 4 +- core/metadata/src/impls/metadata.rs | 9 +- core/partitions/src/iggy_partition.rs | 9 +- core/partitions/src/lib.rs | 6 +- core/partitions/src/types.rs | 82 +++----- core/server/config.toml | 13 ++ core/server/src/bootstrap.rs | 81 ++++--- core/server/src/dispatch.rs | 7 +- core/server/src/main.rs | 4 +- core/server/src/partition_reconciler.rs | 233 ++++++++++++++++++++- core/server_common/src/buffer.rs | 4 +- core/server_common/src/executor.rs | 100 +++++++-- core/server_common/src/lib.rs | 2 +- core/server_common/src/memory_pool.rs | 23 +- core/shard/src/config.rs | 9 +- core/shard/src/lib.rs | 8 +- core/simulator/src/bin/simulator-ui.rs | 4 +- core/simulator/src/bin/workload-fuzz.rs | 4 +- core/simulator/src/lib.rs | 56 ++--- core/simulator/src/replica.rs | 3 +- 22 files changed, 535 insertions(+), 177 deletions(-) diff --git a/core/configs/src/common/server.rs b/core/configs/src/common/server.rs index 315abeadc5..887cfe8ab3 100644 --- a/core/configs/src/common/server.rs +++ b/core/configs/src/common/server.rs @@ -20,7 +20,7 @@ use iggy_common::{IggyByteSize, IggyDuration}; use serde::{Deserialize, Serialize}; use serde_with::DisplayFromStr; use serde_with::serde_as; -use server_common::MemoryPoolConfigOther; +use server_common::MemoryPoolSettings; use server_common::log::{TelemetryEndpointSettings, TelemetrySettings}; pub use server_common::log::TelemetryTransport; @@ -34,12 +34,12 @@ pub struct MemoryPoolConfig { pub bucket_capacity: u32, } -impl MemoryPoolConfig { - pub fn into_other(&self) -> MemoryPoolConfigOther { - MemoryPoolConfigOther { - enabled: self.enabled, - size: self.size, - bucket_capacity: self.bucket_capacity, +impl From<&MemoryPoolConfig> for MemoryPoolSettings { + fn from(config: &MemoryPoolConfig) -> Self { + Self { + enabled: config.enabled, + size: config.size, + bucket_capacity: config.bucket_capacity, } } } diff --git a/core/configs/src/server_config/cluster.rs b/core/configs/src/server_config/cluster.rs index a1874a6406..d71f86ae6b 100644 --- a/core/configs/src/server_config/cluster.rs +++ b/core/configs/src/server_config/cluster.rs @@ -299,6 +299,41 @@ pub struct ClusterConfig { /// Replica-to-replica TLS settings for the consensus (`tcp_replica`) port. #[serde(default)] pub tls: ClusterTlsConfig, + /// Shard-0 coordinator placement tunables. + #[serde(default)] + pub coordinator: ClusterCoordinatorConfig, +} + +/// Placement tunables for the shard-0 coordinator, converted into the shard +/// crate's `CoordinatorConfig` at bootstrap (the domain type lives there +/// because `configs` and `shard` share no dependency edge). +#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] +#[serde(deny_unknown_fields)] +pub struct ClusterCoordinatorConfig { + /// 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 them to + /// peer shards by default. + #[serde(default = "default_skip_shard_zero_for_replicas")] + pub skip_shard_zero_for_replicas: bool, + /// When `total_shards > 1`, exclude shard 0 from client placement. + /// Default false: client connections are short-lived and benefit from + /// shard-0 parallelism more than replicas do. + #[serde(default)] + pub skip_shard_zero_for_clients: bool, +} + +fn default_skip_shard_zero_for_replicas() -> bool { + true +} + +impl Default for ClusterCoordinatorConfig { + fn default() -> Self { + Self { + skip_shard_zero_for_replicas: default_skip_shard_zero_for_replicas(), + skip_shard_zero_for_clients: false, + } + } } /// Replica-to-replica authentication for the consensus (`tcp_replica`) port. @@ -1451,6 +1486,7 @@ mod tests { previous_shared_secret: "retiring-psk-MUST-NOT-be-persisted".to_owned(), }, tls: ClusterTlsConfig::default(), + coordinator: ClusterCoordinatorConfig::default(), }; let serialized = serde_json::to_string(&config).expect("serialize cluster config"); assert!( @@ -1783,6 +1819,7 @@ mod cluster_validate_tests { nodes, auth: ClusterAuthConfig::default(), tls: ClusterTlsConfig::default(), + coordinator: ClusterCoordinatorConfig::default(), } } diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs index 1705e861ba..f752d32659 100644 --- a/core/configs/src/server_config/defaults.rs +++ b/core/configs/src/server_config/defaults.rs @@ -24,7 +24,8 @@ //! [`crate::common::defaults`]. use super::cluster::{ - ClusterAuthConfig, ClusterConfig, ClusterNodeConfig, ClusterTlsConfig, TransportPorts, + ClusterAuthConfig, ClusterConfig, ClusterCoordinatorConfig, ClusterNodeConfig, + ClusterTlsConfig, TransportPorts, }; use super::message_bus::MessageBusConfig; use super::metadata::MetadataConfig; @@ -148,6 +149,7 @@ impl Default for ClusterConfig { .collect(), auth: ClusterAuthConfig::default(), tls: ClusterTlsConfig::default(), + coordinator: ClusterCoordinatorConfig::default(), } } } diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index ee67e1097c..1fda606d12 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -1138,10 +1138,6 @@ where return; } - // TODO add assertions for valid state here. - - // TODO handle gap in ops. - // Verify hash chain integrity BEFORE checkpoint. `checkpoint_if_needed` // can drain WAL entries, making previous_header return None. if let Some(previous) = journal.handle().previous_header(&header) { @@ -3423,12 +3419,11 @@ where let header = *message.header(); - // TODO: calculate the index; #[allow(clippy::cast_possible_truncation)] - let idx = header.op as usize; + let op = header.op as usize; assert_eq!(header.command, Command::Prepare); assert!( - journal.handle().header(idx).is_some(), + journal.handle().header(op).is_some(), "replicate: prepare must be durable in local journal before chain-forward" ); if let Err(e) = replicate_to_next_in_chain(consensus, message).await { diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 4a366076dd..369db555a9 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -3704,10 +3704,10 @@ where let enforce_fsync = self.effective_enforce_fsync(config); let preallocate_segments = self.effective_preallocate_segments(config); let segment = Segment::new(start_offset, segment_size); - // `PartitionsConfig::get_messages_path` is a stub (`/tmp/iggy_stub`); - // the partition's real directory is only known to the server config - // that created the initial segment, so derive the rotated paths from - // the active writer's location. + // Prefer the active writer's location: a per-topic path override or a + // config change after the initial segment was created must not scatter + // one partition's segments across two directories. The config layout + // only decides for a partition with no writer yet. let (messages_path, index_path) = self.partition_dir().map_or_else( || { ( @@ -6531,6 +6531,7 @@ mod tests { segment_size: IggyByteSize::from(1024 * 1024), preallocate_segments: false, encryptor: None, + path_layout: crate::PartitionPathLayout::default(), } } diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 6d14d4b6b9..4649af8290 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -46,9 +46,9 @@ pub use segment::Segment; use server_common::Message; pub use server_common::send_messages::{IggyMessage, IggyMessageHeader, IggyMessages}; pub use types::{ - AppendResult, Fragment, PartitionOffsets, PartitionsConfig, PollFragments, PollQueryResult, - PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS, RepairConclusion, RepairSession, - SendMessagesResult, + AppendResult, Fragment, PartitionOffsets, PartitionPathLayout, PartitionsConfig, PollFragments, + PollQueryResult, PollingArgs, PollingConsumer, REPAIR_RETRY_TICKS, RepairConclusion, + RepairSession, SendMessagesResult, }; /// Partition-level data plane operations. diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index dcf4eb0b92..df15b43931 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -256,6 +256,32 @@ pub enum RepairConclusion { FloorRefused { floor: u64, to_op: u64 }, } +/// Where partition directories live on disk, mirroring the server's +/// `SystemConfig` path scheme so segment files created by the partition plane +/// land next to the ones the server bootstrap created. +#[derive(Debug, Clone)] +pub struct PartitionPathLayout { + /// `{system.path}/{stream.path}`: the directory holding per-stream dirs. + pub streams_root: String, + /// Directory name of the per-topic level (`topic.path`). + pub topics_dir: String, + /// Directory name of the per-partition level (`partition.path`). + pub partitions_dir: String, +} + +/// Synthetic layout for tests and the simulator, where paths only key the +/// sim storage and never touch a real filesystem. The server always wires +/// the real layout from its `SystemConfig`. +impl Default for PartitionPathLayout { + fn default() -> Self { + Self { + streams_root: "/tmp/iggy_stub/streams".to_string(), + topics_dir: "topics".to_string(), + partitions_dir: "partitions".to_string(), + } + } +} + /// Configuration for partition operations. /// /// Mirrors the relevant fields from the server's `PartitionConfig` and @@ -286,6 +312,8 @@ pub struct PartitionsConfig { /// decrypts uniformly whether a fragment came from the resident journal /// or from disk. pub encryptor: Option>, + /// On-disk location scheme for partition directories. + pub path_layout: PartitionPathLayout, } impl PartitionsConfig { @@ -296,14 +324,15 @@ impl PartitionsConfig { topic_id: usize, partition_id: usize, ) -> String { - format!("/tmp/iggy_stub/streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}") + format!( + "{}/{stream_id}/{}/{topic_id}/{}/{partition_id}", + self.path_layout.streams_root, + self.path_layout.topics_dir, + self.path_layout.partitions_dir, + ) } /// Constructs the file path for segment messages. - /// - /// TODO: This is a stub waiting for completion of issue to move server config - /// to shared module. Real implementation should use: - /// `{base_path}/{streams_path}/{stream_id}/{topics_path}/{topic_id}/{partitions_path}/{partition_id}/{start_offset:0>20}.log` #[must_use] pub fn get_messages_path( &self, @@ -319,10 +348,6 @@ impl PartitionsConfig { } /// Constructs the file path for segment indexes. - /// - /// TODO: This is a stub waiting for completion of issue to move server config - /// to shared module. Real implementation should use: - /// `{base_path}/{streams_path}/{stream_id}/{topics_path}/{topic_id}/{partitions_path}/{partition_id}/{start_offset:0>20}.index` #[must_use] pub fn get_index_path( &self, @@ -336,43 +361,4 @@ impl PartitionsConfig { self.get_partition_path(stream_id, topic_id, partition_id) ) } - - #[must_use] - pub fn get_offsets_path( - &self, - stream_id: usize, - topic_id: usize, - partition_id: usize, - ) -> String { - format!( - "{}/offsets", - self.get_partition_path(stream_id, topic_id, partition_id) - ) - } - - #[must_use] - pub fn get_consumer_offsets_path( - &self, - stream_id: usize, - topic_id: usize, - partition_id: usize, - ) -> String { - format!( - "{}/consumers", - self.get_offsets_path(stream_id, topic_id, partition_id) - ) - } - - #[must_use] - pub fn get_consumer_group_offsets_path( - &self, - stream_id: usize, - topic_id: usize, - partition_id: usize, - ) -> String { - format!( - "{}/groups", - self.get_offsets_path(stream_id, topic_id, partition_id) - ) - } } diff --git a/core/server/config.toml b/core/server/config.toml index f420aeeb30..d5046c5091 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -622,6 +622,19 @@ key_file = "" # use hostnames in the roster if the certificates only have DNS SANs. ca_file = "" +# Shard-0 coordinator placement. +[cluster.coordinator] +# When the server runs more than one shard, exclude shard 0 from replica +# placement. Shard 0 already hosts the coordinator, the metadata writer, and +# both listeners; replica connections are long-lived steady flows, so they are +# offloaded to peer shards by default. +skip_shard_zero_for_replicas = true + +# When the server runs more than one shard, exclude shard 0 from client +# placement. Off by default: client connections are short-lived and benefit +# from shard-0 parallelism more than replicas do. +skip_shard_zero_for_clients = false + # Full roster of cluster members. Byte-identical on every node. The running # node's identity is resolved at launch from the '--replica-id ' CLI # flag, which selects the entry in this list that describes the current diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 01ef2d1fab..f3cf877cef 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -989,7 +989,12 @@ async fn shard_main( 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); + let mut shutdown_watchdog = Some(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 @@ -1194,20 +1199,17 @@ async fn shard_main( ); // 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 + // message pump: it keeps the bus' `background_tasks` vec empty on the + // shutdown path, and shard 0 would otherwise still open TCP/QUIC/WS // listeners for a server that is already tearing down, briefly // accepting connections that immediately get torn by the watchdog. + // + // The flag is set, so the watchdog is (about to be) driving + // `bus.shutdown()`; await it so the runtime does not drop mid-drain. if shutdown_flag_for_handoff.load(Ordering::Relaxed) { + if let Some(watchdog) = shutdown_watchdog.take() { + let _ = watchdog.await; + } return Ok(()); } @@ -1420,6 +1422,12 @@ async fn shard_main( // The bind failure is the primary fault; the drain verdict only // matters for the log it emits. let _ = await_pump_drain(pump_handle.take(), config, shard_id).await; + // Neither the flag nor the bus token has fired yet on this path, + // so the watchdog is still idle-looping; awaiting it would hang. + // Detach and let `run_shard_thread`'s unwind flip the flag. + if let Some(watchdog) = shutdown_watchdog.take() { + watchdog.detach(); + } return Err(error); } @@ -1446,7 +1454,15 @@ async fn shard_main( let _ = tx.try_send(()); } - await_pump_drain(pump_handle.take(), config, shard_id).await?; + // Await the watchdog even when the drain verdict is an error: the token + // has fired, so it either stands down within one poll interval or is + // mid-`bus.shutdown()`, and dropping it there truncates in-flight + // `ClientForwardFailed` replies. + let pump_verdict = await_pump_drain(pump_handle.take(), config, shard_id).await; + if let Some(watchdog) = shutdown_watchdog.take() { + let _ = watchdog.await; + } + pump_verdict?; info!(shard = shard_id, "server shard exited cleanly"); Ok(()) @@ -1662,16 +1678,27 @@ async fn await_bootstrap_complete( /// is the only Send signal we have; the bus' shutdown machinery is /// `!Send` (`Rc>` + per-shard `async_channel`), so it must be /// triggered from within the runtime that owns the bus. +/// +/// The caller owns the returned handle and must await it on the exit paths +/// where shutdown is in progress (flag set or bus token triggered): +/// dropping it there cancels the watchdog mid-`bus.shutdown()`, truncating +/// in-flight `ClientForwardFailed` replies (terminal per `SendError` docs). +/// It cannot go through `bus.track_background` instead: 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). The await is bounded: once the token fires the loop +/// stands down within one poll interval, and the shutdown call itself is +/// capped by `drain_timeout`. #[allow(clippy::needless_pass_by_value)] fn spawn_shutdown_watchdog( bus: Rc, shutdown_flag: Arc, drain_timeout: Duration, poll_interval: Duration, -) { +) -> compio::runtime::JoinHandle<()> { let bus_for_task = Rc::clone(&bus); let bus_token = bus.token(); - let watchdog = compio::runtime::spawn(async move { + compio::runtime::spawn(async move { loop { if shutdown_flag.load(Ordering::Relaxed) { break; @@ -1684,19 +1711,7 @@ fn spawn_shutdown_watchdog( 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 so background tasks can be reaped without coupling - // to the bus shutdown order. - watchdog.detach(); + }) } /// Copy the configured cluster roster plus this node's own client ports into @@ -1790,6 +1805,11 @@ async fn build_shard_for_thread( segment_size: IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE), preallocate_segments: iggy_common::DEFAULT_PREALLOCATE_SEGMENTS, encryptor, + path_layout: partitions::PartitionPathLayout { + streams_root: config.system.get_streams_path(), + topics_dir: config.system.topic.path.clone(), + partitions_dir: config.system.partition.path.clone(), + }, }, owned_partitions_capacity, ); @@ -2007,7 +2027,10 @@ async fn build_shard_for_thread( shard::ReplicaTopology::new(topology.self_replica_id, topology.replica_count), Rc::clone(&bus), ), - CoordinatorConfig::default(), + CoordinatorConfig { + skip_shard_zero_for_replicas: config.cluster.coordinator.skip_shard_zero_for_replicas, + skip_shard_zero_for_clients: config.cluster.coordinator.skip_shard_zero_for_clients, + }, metrics, ) .build() diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 9251bf925b..d3821b15b3 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -3842,7 +3842,7 @@ mod tests { use metadata::stm::stream::Streams; use metadata::stm::user::Users; use metadata::{IggyMetadata, MuxStateMachine}; - use partitions::{IggyPartitions, PartitionsConfig}; + use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig}; use server_common::iobuf::Frozen; use server_common::sharding::ShardId; use server_common::{MESSAGE_ALIGN, Message, MessageBag}; @@ -3999,6 +3999,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); TestShard::without_inbox( @@ -4165,6 +4166,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); let shard = Rc::new(TestShard::without_inbox( @@ -4286,6 +4288,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); let shard = Rc::new(TestShard::without_inbox( @@ -4410,6 +4413,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); let shard = Rc::new(TestShard::without_inbox( @@ -4474,6 +4478,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); // Real sender ring so the staged deny is observable: the test holds diff --git a/core/server/src/main.rs b/core/server/src/main.rs index f6a8e63e36..e2547efc7d 100644 --- a/core/server/src/main.rs +++ b/core/server/src/main.rs @@ -71,7 +71,9 @@ fn main() -> Result<(), ServerError> { let bootstrap_result: Result = bootstrap_runtime.block_on(async { let config = load_config().await?; prepare_runtime_dirs(&config, &mut logging, args.fresh).await?; - server_common::MemoryPool::init_pool(&config.system.memory_pool.into_other()); + let memory_pool_settings = + server_common::MemoryPoolSettings::from(&config.system.memory_pool); + server_common::MemoryPool::init_pool(&memory_pool_settings); Ok(config) }); diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index b456a30b81..fbf99df6bb 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -1183,8 +1183,8 @@ mod tests { PurgeTopicRequest, }; use iggy_binary_protocol::{ - Command, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader, WireIdentifier, - WireOptions, + Command, Operation, PrepareHeader, RepairRangeReplyHeader, ReplyHeader, + RequestPreparesHeader, RoutedRequestHeader, WireIdentifier, WireOptions, }; use message_bus::IggyMessageBus; use metadata::IggyMetadata; @@ -1193,7 +1193,7 @@ mod tests { use metadata::stm::StateMachine; use metadata::stm::stream::Streams; use metadata::stm::user::Users; - use partitions::{IggyPartitions, PartitionsConfig}; + use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig, RepairSession}; use server_common::sharding::{IggyNamespace, ShardId}; use server_common::{Message, MessageBag}; use shard::shards_table::{PapayaShardsTable, ShardsTable, calculate_shard_assignment}; @@ -1318,6 +1318,54 @@ mod tests { MessageBag::Request(msg) } + /// Build a partition-plane `RepairRangeReply` as the serving peer would + /// send it. Only the fields the receive path reads are stamped: routing + /// (`group`), session (`nonce`), and the verdict (`command`, `op`). + fn build_repair_range_reply( + namespace: IggyNamespace, + command: Command, + nonce: u128, + op: u64, + ) -> MessageBag { + let header_size = size_of::(); + let mut msg = Message::::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::( + &mut msg.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes form a valid RepairRangeReplyHeader"); + header.command = command; + header.size = u32::try_from(header_size).expect("header size fits u32"); + header.nonce = nonce; + header.op = op; + header.group = namespace.inner(); + MessageBag::RepairRangeReply(msg) + } + + /// Build a partition-plane `RequestPrepares` as a rejoining peer would + /// send it. `replica` is the requester the serve path replies to. + fn build_request_prepares( + namespace: IggyNamespace, + replica: u8, + nonce: u128, + from_op: u64, + to_op: u64, + ) -> MessageBag { + let header_size = size_of::(); + let mut msg = Message::::new(header_size); + let header = bytemuck::checked::try_from_bytes_mut::( + &mut msg.as_mut_slice()[..header_size], + ) + .expect("zeroed bytes form a valid RequestPreparesHeader"); + header.command = Command::RequestPrepares; + header.size = u32::try_from(header_size).expect("header size fits u32"); + header.replica = replica; + header.nonce = nonce; + header.from_op = from_op; + header.to_op = to_op; + header.group = namespace.inner(); + MessageBag::RequestPrepares(msg) + } + fn assignment(partition_id: u32, consensus_group_id: u64) -> CreatedPartitionAssignment { CreatedPartitionAssignment { partition_id, @@ -1468,6 +1516,7 @@ mod tests { segment_size: iggy_common::IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }, ); let shards_table = PapayaShardsTable::new(); @@ -2285,6 +2334,184 @@ mod tests { ); } + /// Receive half of the purge gate in `on_repair_range_reply`: while a + /// committed purge has not applied locally, a repair verdict must be + /// deferred wholesale -- installing the peer's floor against pre-purge + /// segments silently loses the post-purge batches (offsets restarting at + /// 0 flush-skip below the stale durable line). + #[compio::test] + async fn repair_completion_defers_until_committed_purge_applies() { + const NONCE: u128 = 7; + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-repair-gate"); + seed_topic(&mux, 2, 0, "topic-repair-gate", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + reconcile_pass(&ctx).await; + + let ns = IggyNamespace::new(0, 0, 0); + shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("partition is materialised") + .repair = Some(RepairSession { + nonce: NONCE, + to_op: 5, + floor: None, + peer: 1, + first_batch_offset: None, + idle_ticks: 0, + }); + + // Committed purge: generation 1 > applied 0. + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + shard + .plane + .metadata() + .mux_stm + .update(build_prepare(3, Operation::PurgeTopic, &purge)) + .expect("PurgeTopic apply succeeds"); + + let deferred_before = shard + .metrics() + .partition_repair_serves_deferred_purge_value(); + shard + .on_message(build_repair_range_reply( + ns, + Command::RangeEvicted, + NONCE, + 4, + )) + .await; + let session = shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("partition survives the deferral") + .repair + .expect("deferral must leave the repair session armed"); + assert_eq!( + session.floor, None, + "a deferred RangeEvicted must not install the peer's floor" + ); + assert_eq!( + shard + .metrics() + .partition_repair_serves_deferred_purge_value(), + deferred_before + 1, + "the deferral must be visible on the purge-deferred counter" + ); + + // Apply the purge; the same frame now lands. + let partitions_config = shard.plane.partitions().config().clone(); + shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("purged partition is materialised") + .purge(&partitions_config, 1) + .await + .expect("apply staged purge"); + shard + .on_message(build_repair_range_reply( + ns, + Command::RangeEvicted, + NONCE, + 4, + )) + .await; + let session = shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("partition survives the retry") + .repair + .expect("RangeEvicted records the floor but keeps the session"); + assert_eq!( + session.floor, + Some(3), + "after the purge applies, the retried frame must install the floor" + ); + assert_eq!( + shard + .metrics() + .partition_repair_serves_deferred_purge_value(), + deferred_before + 1, + "the retried frame must pass the gate without another deferral" + ); + } + + /// Serve half of the purge gate in `on_request_prepares`: while a + /// committed purge has not applied locally, the journal still holds + /// pre-purge entries with no floor to fence them, so serving a rejoiner + /// must be deferred (no reply; the requester's stall retry re-asks). + #[compio::test] + async fn repair_serve_defers_until_committed_purge_applies() { + const NONCE: u128 = 11; + let tmp = TempDir::new().expect("tempdir for system path"); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream-serve-gate"); + seed_topic(&mux, 2, 0, "topic-serve-gate", vec![assignment(0, 1)]); + + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + reconcile_pass(&ctx).await; + + let ns = IggyNamespace::new(0, 0, 0); + let purge = PurgeTopicRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + }; + shard + .plane + .metadata() + .mux_stm + .update(build_prepare(3, Operation::PurgeTopic, &purge)) + .expect("PurgeTopic apply succeeds"); + + let deferred_before = shard + .metrics() + .partition_repair_serves_deferred_purge_value(); + shard + .on_message(build_request_prepares(ns, 1, NONCE, 1, 5)) + .await; + assert_eq!( + shard + .metrics() + .partition_repair_serves_deferred_purge_value(), + deferred_before + 1, + "an unapplied purge must defer the serve" + ); + + let partitions_config = shard.plane.partitions().config().clone(); + shard + .plane + .partitions() + .get_mut_by_ns(&ns) + .expect("purged partition is materialised") + .purge(&partitions_config, 1) + .await + .expect("apply staged purge"); + shard + .on_message(build_request_prepares(ns, 1, NONCE, 1, 5)) + .await; + assert_eq!( + shard + .metrics() + .partition_repair_serves_deferred_purge_value(), + deferred_before + 1, + "once the purge applies, the retried request must be served, not deferred" + ); + } + /// Permanent-tombstone-wedge regression: a teardown whose disk delete /// fails sets the tombstone and removes the `shards_table` row but never /// enqueues `ConfirmRemove`, so the tombstone never lifts. If the same diff --git a/core/server_common/src/buffer.rs b/core/server_common/src/buffer.rs index aa84101b77..79b018716f 100644 --- a/core/server_common/src/buffer.rs +++ b/core/server_common/src/buffer.rs @@ -434,7 +434,7 @@ mod miri_tests { #[cfg(not(miri))] mod split_to { use super::*; - use crate::memory_pool::{MemoryPool, MemoryPoolConfigOther}; + use crate::memory_pool::{MemoryPool, MemoryPoolSettings}; use iggy_common::IggyByteSize; use serial_test::serial; use std::str::FromStr; @@ -444,7 +444,7 @@ mod miri_tests { fn init_pool_for_split_to_tests() { MIRI_POOL_INIT.call_once(|| { - let config = MemoryPoolConfigOther { + let config = MemoryPoolSettings { enabled: true, size: IggyByteSize::from_str("64MiB").unwrap(), bucket_capacity: 16, diff --git a/core/server_common/src/executor.rs b/core/server_common/src/executor.rs index ed8ea36373..8fb8cc8422 100644 --- a/core/server_common/src/executor.rs +++ b/core/server_common/src/executor.rs @@ -20,6 +20,13 @@ use compio::runtime::Runtime; const DEFAULT_SHARD_RUNTIME_CAPACITY: u32 = 4096; const SHARD_RUNTIME_CAPACITY_ENV: &str = "IGGY_SHARD_RUNTIME_CAPACITY"; +/// How many tasks the runtime polls between driver (io_uring) sweeps. The +/// default suits the stock task population; a deployment expecting far more +/// connected clients (each connection is roughly one task) can raise it via +/// [`SHARD_EVENT_INTERVAL_ENV`] to amortise driver sweeps across more work. +const DEFAULT_SHARD_EVENT_INTERVAL: usize = 128; +const SHARD_EVENT_INTERVAL_ENV: &str = "IGGY_SHARD_EVENT_INTERVAL"; + /// Resolves the per-shard io_uring SQ/CQ capacity from `IGGY_SHARD_RUNTIME_CAPACITY`, /// falling back to [`DEFAULT_SHARD_RUNTIME_CAPACITY`] when the var is missing or /// fails to parse as `u32`. @@ -30,12 +37,26 @@ fn shard_capacity_from_env() -> u32 { .unwrap_or(DEFAULT_SHARD_RUNTIME_CAPACITY) } +/// Resolves the runtime event interval from `IGGY_SHARD_EVENT_INTERVAL`, +/// falling back to [`DEFAULT_SHARD_EVENT_INTERVAL`] when the var is missing, +/// fails to parse, or is zero (compio treats the interval as a divisor). +fn shard_event_interval_from_env() -> usize { + std::env::var(SHARD_EVENT_INTERVAL_ENV) + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&interval| interval > 0) + .unwrap_or(DEFAULT_SHARD_EVENT_INTERVAL) +} + /// Creates a compio runtime for a shard thread, with shard-specific `io_uring` flags. /// /// The per-ring SQ/CQ capacity defaults to `4096` and can be overridden via the /// `IGGY_SHARD_RUNTIME_CAPACITY` env var, which the multi-node integration /// harness sets to `256` so N nodes * M shards fit under an 8 MiB -/// `RLIMIT_MEMLOCK` budget without `ENOMEM` at ring setup. +/// `RLIMIT_MEMLOCK` budget without `ENOMEM` at ring setup. The runtime event +/// interval defaults to `128` and can be overridden via +/// `IGGY_SHARD_EVENT_INTERVAL` for deployments whose task count (roughly the +/// connected-client count) makes a different poll-to-sweep ratio pay off. /// /// # Errors /// @@ -47,9 +68,6 @@ fn shard_capacity_from_env() -> u32 { /// Falling back to default flags would silently degrade shard performance - /// do not add a retry with reduced flags here. pub fn create_shard_executor() -> Result { - // TODO: The event interval tick, could be configured based on the fact - // How many clients we expect to have connected. - // This roughly estimates the number of tasks we will create. let mut proactor = compio::driver::ProactorBuilder::new(); proactor @@ -57,45 +75,56 @@ pub fn create_shard_executor() -> Result { .coop_taskrun(true) .taskrun_flag(true); - // FIXME(hubcio): Only set thread_pool_limit(0) on non-macOS platforms - // This causes a freeze on macOS with compio fs operations - // see https://github.com/compio-rs/compio/issues/446 + // Permanent divergence, not a workaround to revisit: macOS runs compio's + // polling driver, which routes fs operations through the blocking pool, so + // a zero limit cannot work there by design. Upstream closed + // https://github.com/compio-rs/compio/issues/446 by making blocking-pool + // dispatch with no workers panic ("the thread pool is needed but no worker + // thread is running", compio-driver asyncify.rs) instead of freeze. + // io_uring targets keep the zero limit: no blocking pool exists on shard + // threads, which `core/partitions` messages_writer relies on to justify + // running fallocate inline (`spawn_blocking` would hit that same panic). #[cfg(not(all(target_os = "macos", target_arch = "aarch64")))] proactor.thread_pool_limit(0); compio::runtime::RuntimeBuilder::new() .with_proactor(proactor.to_owned()) - .event_interval(128) + .event_interval(shard_event_interval_from_env()) .build() } #[cfg(test)] mod tests { use super::{ - DEFAULT_SHARD_RUNTIME_CAPACITY, SHARD_RUNTIME_CAPACITY_ENV, shard_capacity_from_env, + DEFAULT_SHARD_EVENT_INTERVAL, DEFAULT_SHARD_RUNTIME_CAPACITY, SHARD_EVENT_INTERVAL_ENV, + SHARD_RUNTIME_CAPACITY_ENV, shard_capacity_from_env, shard_event_interval_from_env, }; use serial_test::serial; - fn with_capacity_env(value: Option<&str>, f: impl FnOnce() -> R) -> R { + fn with_env(name: &str, value: Option<&str>, f: impl FnOnce() -> R) -> R { // SAFETY: tests in this module are #[serial], so no other thread races // on the process-wide environment while the guard is active. - let prev = std::env::var(SHARD_RUNTIME_CAPACITY_ENV).ok(); + let prev = std::env::var(name).ok(); unsafe { match value { - Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v), - None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV), + Some(v) => std::env::set_var(name, v), + None => std::env::remove_var(name), } } let out = f(); unsafe { match prev { - Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v), - None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV), + Some(v) => std::env::set_var(name, v), + None => std::env::remove_var(name), } } out } + fn with_capacity_env(value: Option<&str>, f: impl FnOnce() -> R) -> R { + with_env(SHARD_RUNTIME_CAPACITY_ENV, value, f) + } + #[test] #[serial] fn shard_capacity_from_env_uses_parsed_value() { @@ -127,4 +156,45 @@ mod tests { assert_eq!(shard_capacity_from_env(), DEFAULT_SHARD_RUNTIME_CAPACITY); }); } + + #[test] + #[serial] + fn shard_event_interval_from_env_uses_parsed_value() { + with_env(SHARD_EVENT_INTERVAL_ENV, Some("512"), || { + assert_eq!(shard_event_interval_from_env(), 512); + }); + } + + #[test] + #[serial] + fn shard_event_interval_from_env_falls_back_when_unset() { + with_env(SHARD_EVENT_INTERVAL_ENV, None, || { + assert_eq!( + shard_event_interval_from_env(), + DEFAULT_SHARD_EVENT_INTERVAL + ); + }); + } + + #[test] + #[serial] + fn shard_event_interval_from_env_falls_back_on_unparsable() { + with_env(SHARD_EVENT_INTERVAL_ENV, Some("not-a-number"), || { + assert_eq!( + shard_event_interval_from_env(), + DEFAULT_SHARD_EVENT_INTERVAL + ); + }); + } + + #[test] + #[serial] + fn shard_event_interval_from_env_falls_back_on_zero() { + with_env(SHARD_EVENT_INTERVAL_ENV, Some("0"), || { + assert_eq!( + shard_event_interval_from_env(), + DEFAULT_SHARD_EVENT_INTERVAL + ); + }); + } } diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs index c8fa01d42b..f7b7106586 100644 --- a/core/server_common/src/lib.rs +++ b/core/server_common/src/lib.rs @@ -39,7 +39,7 @@ pub use consensus_message::{ MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking, ResponseBackingKind, }; pub use executor::create_shard_executor; -pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther, memory_pool}; +pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolSettings, memory_pool}; pub use segment_storage::{ IndexReader, IndexWriter, MessagesReader, MessagesWriter, SegmentStorage, }; diff --git a/core/server_common/src/memory_pool.rs b/core/server_common/src/memory_pool.rs index b55c1c64a8..a30ed4b2ec 100644 --- a/core/server_common/src/memory_pool.rs +++ b/core/server_common/src/memory_pool.rs @@ -70,12 +70,11 @@ pub fn memory_pool() -> &'static MemoryPool { .expect("Memory pool not initialized - MemoryPool::init_pool should be called first") } -// TODO: Extract shared domain types (IggyByteSize, IggyDuration, etc.) into an `iggy_types` -// leaf crate so `iggy_common` can depend on `configs` directly. That lets us delete this -// duplicate and use `configs::server::MemoryPoolConfig` here instead. -/// Configuration for the memory pool. +/// Pool settings, converted from the serde-facing config type in +/// `configs::server::MemoryPoolConfig` (which depends on this crate, so the +/// domain type has to live here - same split as [`crate::log::TelemetrySettings`]). #[derive(Debug)] -pub struct MemoryPoolConfigOther { +pub struct MemoryPoolSettings { /// Whether the pool is enabled. pub enabled: bool, /// Maximum size of the pool. @@ -162,11 +161,11 @@ impl MemoryPool { } } - /// Initialize the global pool from the given config. - pub fn init_pool(config: &MemoryPoolConfigOther) { - let is_enabled = config.enabled; - let memory_limit = config.size.as_bytes_usize(); - let bucket_capacity = config.bucket_capacity as usize; + /// Initialize the global pool from the given settings. + pub fn init_pool(settings: &MemoryPoolSettings) { + let is_enabled = settings.enabled; + let memory_limit = settings.size.as_bytes_usize(); + let bucket_capacity = settings.bucket_capacity as usize; let _ = MEMORY_POOL.get_or_init(|| MemoryPool::new(is_enabled, memory_limit, bucket_capacity)); @@ -502,12 +501,12 @@ mod tests { fn initialize_pool_for_tests() { TEST_INIT.call_once(|| { - let config = MemoryPoolConfigOther { + let settings = MemoryPoolSettings { enabled: true, size: IggyByteSize::from_str("4GiB").unwrap(), bucket_capacity: 8192, }; - MemoryPool::init_pool(&config); + MemoryPool::init_pool(&settings); }); } diff --git a/core/shard/src/config.rs b/core/shard/src/config.rs index 27535c2e9a..0406210b14 100644 --- a/core/shard/src/config.rs +++ b/core/shard/src/config.rs @@ -17,11 +17,10 @@ //! Runtime tunables for the shard-0 coordinator. //! -//! TODO: move this module into `core/configs` (as a `CoordinatorConfig` -//! section nested under `ClusterConfig`) once downstream bootstrap -//! wiring that constructs [`crate::coordinator::ShardZeroCoordinator`] -//! from `ServerConfig` lands. Kept in-crate for now to avoid churning -//! the configs crate ahead of that wiring. +//! The serde-facing section lives in `core/configs` as +//! `[cluster.coordinator]` (`ClusterCoordinatorConfig`); the server's +//! bootstrap converts it into this domain type. The split exists because +//! `configs` and `shard` share no dependency edge. /// Tunables for [`crate::coordinator::ShardZeroCoordinator`]. #[derive(Debug, Clone)] diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 4e20691b0e..a51dfa25fa 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -4611,11 +4611,9 @@ where // Defer the whole reply: the purge is one reconciler wake away and // resets the line to `None`, and the stall retry re-asks, so the peer // re-emits both `RangeEvicted` and `RepairDone` for the same window. - // TODO(hubcio): no direct test drives this gate -- `on_repair_range_reply` - // is only reachable through the real message bus and the shard crate has - // no fixture for it (the serve-side gate shares the gap). The loss shape - // is pinned at the partition level instead; a bus fixture would let both - // gates be exercised end to end. + // Pinned by `repair_completion_defers_until_committed_purge_applies` + // (server crate, partition_reconciler tests), driven through the pub + // `on_message` entry; the serve-side twin has its own pin there. let committed_purge = self .plane .metadata() diff --git a/core/simulator/src/bin/simulator-ui.rs b/core/simulator/src/bin/simulator-ui.rs index 3de3100b65..041a7c8860 100644 --- a/core/simulator/src/bin/simulator-ui.rs +++ b/core/simulator/src/bin/simulator-ui.rs @@ -20,7 +20,7 @@ use iggy_binary_protocol::ReplyHeader; use iggy_common::{IggyByteSize, PollingStrategy}; use partitions::{PollingArgs, PollingConsumer}; use server_common::sharding::IggyNamespace; -use server_common::{MemoryPool, MemoryPoolConfigOther, Message}; +use server_common::{MemoryPool, MemoryPoolSettings, Message}; use simulator::Simulator; use simulator::client::SimClient; use simulator::packet::PacketSimulatorOptions; @@ -41,7 +41,7 @@ fn step_until_reply(sim: &mut Simulator, max_ticks: u64) -> Vec u64 { - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -2650,7 +2650,7 @@ mod tests { /// plus its metadata, log a client in against root, produce one message, /// then poll. Returns the poll reply's raw bytes and the schedule hash. fn shell_produce_poll(seed: u64) -> (Vec, u64) { - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -2760,7 +2760,7 @@ mod tests { use consensus::PartitionsHandle; use std::panic::{AssertUnwindSafe, catch_unwind}; - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -2859,7 +2859,7 @@ mod tests { use consensus::PartitionsHandle; use std::panic::{AssertUnwindSafe, catch_unwind}; - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -2957,7 +2957,7 @@ mod tests { use consensus::MetadataHandle; use journal::{Journal, JournalHandle}; - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -3069,7 +3069,7 @@ mod tests { header.group == BLOCKED_NS.load(Ordering::Relaxed) } - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -3177,7 +3177,7 @@ mod tests { oracle, }; - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -3295,7 +3295,7 @@ mod view_change_data_loss_tests { #[test] fn given_committed_op_missing_on_next_primary_when_primary_crashes_should_survive_view_change() { - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, @@ -3393,7 +3393,7 @@ mod view_change_data_loss_tests { #[test] fn given_a_register_inside_the_view_start_persist_when_the_pipeline_rebuilds_should_commit_once() { - server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther { + server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), bucket_capacity: 1, diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index 2764b96da6..362f3a144f 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -27,7 +27,7 @@ use metadata::stm::mux::WithFactory; use metadata::stm::stream::{Streams, StreamsInner}; use metadata::stm::user::{Users, UsersInner}; use metadata::{IggyMetadata, apply_committed_prepare}; -use partitions::{IggyPartitions, PartitionsConfig}; +use partitions::{IggyPartitions, PartitionPathLayout, PartitionsConfig}; use server::bootstrap::{ShellHandlers, ShellShardHandle, wire_shell_handlers}; use server_common::crypto; use server_common::sharding::{METADATA_GROUP, ShardId}; @@ -287,6 +287,7 @@ pub fn new_shard( segment_size: IggyByteSize::from(1024 * 1024 * 1024), preallocate_segments: false, encryptor: None, + path_layout: PartitionPathLayout::default(), }; // Shard id is the NODE-LOCAL shard index, never the replica id: the From 3bd1f142b61e581d6c5a1d643b76621093aec50b Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Tue, 18 Aug 2026 15:59:09 +0200 Subject: [PATCH 3/3] fix CI --- .../server/partition_view_durability_vsr.rs | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/core/integration/tests/server/partition_view_durability_vsr.rs b/core/integration/tests/server/partition_view_durability_vsr.rs index 9d266e447f..1e8a9480a7 100644 --- a/core/integration/tests/server/partition_view_durability_vsr.rs +++ b/core/integration/tests/server/partition_view_durability_vsr.rs @@ -52,9 +52,11 @@ const MESSAGES_COUNT: u32 = 10; /// then the restarted node's probe and repair), and CI runners are slow. 60s bounds /// the worst case without hanging the suite. const CONVERGE_TIMEOUT: Duration = Duration::from_secs(60); -/// Boot line reporting the `(view, log_view)` a partition group restored from its -/// superblock: the recovered replica's own account of what it read back. -const RESTORED_VIEW_MARKER: &str = "restored partition view from its superblock"; +/// Boot line reporting the `(view, log_view)` a consensus group restored from +/// its superblock: the recovered replica's own account of what it read back. +/// One line per group since the unified restore constructor, so the parser +/// filters the metadata group's line out by its `group` field. +const RESTORED_VIEW_MARKER: &str = "restored group view from its superblock"; const POLL_INTERVAL: Duration = Duration::from_millis(250); #[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))] @@ -200,17 +202,23 @@ fn restored_partition_view(harness: &TestHarness, node: usize) -> Option<(u32, u } log.lines() .filter(|line| line.contains(RESTORED_VIEW_MARKER)) + // The metadata group logs the same restore line; its view advances + // independently of the partition group under test, so folding it into + // the max would let a metadata view change satisfy a partition assert. + .filter(|line| { + field(line, " group=") != Some(iggy_binary_protocol::namespace::METADATA_GROUP) + }) .filter_map(|line| { // Leading space so the `view` key cannot match inside `log_view`. - let view = field(line, " view=")?; - let log_view = field(line, " log_view=")?; + let view = u32::try_from(field(line, " view=")?).ok()?; + let log_view = u32::try_from(field(line, " log_view=")?).ok()?; Some((view, log_view)) }) .max() } -/// Value of a space-prefixed `key=` tracing field. -fn field(line: &str, key: &str) -> Option { +/// Value of a space-prefixed `key=` tracing field. +fn field(line: &str, key: &str) -> Option { let start = line.find(key)? + key.len(); line[start..] .split(|character: char| !character.is_ascii_digit())