From 1be86344cefb614e27e0c7fec96da0e02a4b232a Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Thu, 15 Oct 2026 01:36:39 +0800 Subject: [PATCH 1/5] feat(config): add startup readiness stall bounds under [tuning.startup] Startup waits for the metadata group, the data groups, and Calvin catch-up used fixed wall-clock deadlines, so a node that was making steady progress could still be refused. Each wait now ends only when it makes no progress for a configured stall bound, and every applied entry resets the clock. The bounds live in the new StartupTuning config and are range checked on load. Setting them under [server] is rejected with a message that names the [tuning.startup] path. The Calvin scheduler's caught-up registry exposes a watermark so its wait can see progress. Test harness call sites and the getting-started guide are updated. --- docs/getting-started.md | 19 ++- .../node/lifecycle/spawn_full.rs | 5 +- nodedb-test-support/src/single_node.rs | 12 +- nodedb-types/src/config/tuning/config.rs | 19 +++ nodedb-types/src/config/tuning/mod.rs | 5 + nodedb-types/src/config/tuning/startup.rs | 128 ++++++++++++++++ nodedb/src/bootstrap/calvin_catch_up.rs | 38 +++-- nodedb/src/bootstrap/cluster_ready.rs | 92 ++++++++--- nodedb/src/bootstrap/data_group_recovery.rs | 58 +++++-- nodedb/src/bootstrap/mod.rs | 1 + nodedb/src/bootstrap/stall_clock.rs | 144 ++++++++++++++++++ nodedb/src/config/server/config.rs | 115 ++++++++++++++ nodedb/src/config/server/domain.rs | 38 +++++ .../cluster/calvin/scheduler/caught_up.rs | 85 +++++++++-- .../calvin/scheduler/driver/core/scheduler.rs | 1 + .../control/cluster/calvin/scheduler/mod.rs | 2 +- .../src/control/cluster/left_vshard_forget.rs | 11 +- nodedb/src/control/cluster/test_one_node.rs | 10 +- .../src/control/security/auth_lease/status.rs | 6 + nodedb/src/main.rs | 2 + 20 files changed, 722 insertions(+), 69 deletions(-) create mode 100644 nodedb-types/src/config/tuning/startup.rs create mode 100644 nodedb/src/bootstrap/stall_clock.rs diff --git a/docs/getting-started.md b/docs/getting-started.md index c8ffa0233..b95b51d3c 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -629,7 +629,24 @@ Every value must be positive. `verdict_stall_warn_ms` has a floor of `4`. After a restart, a scheduler re-applies epochs up to the highest epoch it applied before. The client gateway opens only after every scheduler on the node reaches that epoch. The wait shares the data-group recovery bound, and -startup fails when the bound expires. +startup fails when no scheduler makes progress for that long. + +**Startup bounds** (config file only): + +| Config field | Default | +| ----------------------------------------------- | ------- | +| `tuning.startup.raft_ready_timeout_ms` | `30000` | +| `tuning.startup.data_group_recovery_timeout_ms` | `60000` | + +Boot fails when a wait makes no progress for its bound. A long replay that +keeps applying entries never fails. + +- `raft_ready_timeout_ms` bounds the metadata raft group's replay. +- `data_group_recovery_timeout_ms` bounds the data raft groups' replay, then + the Calvin scheduler catch-up. + +Each value must be between `1` and `86400000` (one day). Both keys belong under +`[tuning.startup]`. Startup refuses them under `[server]`. **Observability settings:** diff --git a/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs b/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs index b88c03706..6e70a4f11 100644 --- a/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs +++ b/nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs @@ -340,7 +340,10 @@ impl TestClusterNode { // the harness starts nodes one at a time, and a group elects only // once its peers run. It sets `data_groups_replayed`, which gates // the forget of a departed vShard's Calvin state. - nodedb::bootstrap::data_group_recovery::spawn_data_group_recovery(&shared); + nodedb::bootstrap::data_group_recovery::spawn_data_group_recovery( + &shared, + nodedb_types::config::tuning::StartupTuning::default().data_group_recovery_timeout(), + ); // CRDT constraint reconcile loop (leader-gated). The production server // spawns this from `spawn_background_loops`, which the harness does not diff --git a/nodedb-test-support/src/single_node.rs b/nodedb-test-support/src/single_node.rs index 0934489a7..e2fcbe140 100644 --- a/nodedb-test-support/src/single_node.rs +++ b/nodedb-test-support/src/single_node.rs @@ -118,8 +118,16 @@ pub async fn start( }; // The harness cores replay their WAL before they report ready, so no // replay receiver is owed here. - nodedb::bootstrap::cluster_ready::await_cluster_ready(shared, ready_rx, Vec::new(), gates) - .await?; + let startup = nodedb_types::config::tuning::StartupTuning::default(); + nodedb::bootstrap::cluster_ready::await_cluster_ready( + shared, + ready_rx, + Vec::new(), + gates, + startup.raft_ready_timeout(), + startup.data_group_recovery_timeout(), + ) + .await?; Ok(raft) } diff --git a/nodedb-types/src/config/tuning/config.rs b/nodedb-types/src/config/tuning/config.rs index 606fb41e9..63e5bd7b0 100644 --- a/nodedb-types/src/config/tuning/config.rs +++ b/nodedb-types/src/config/tuning/config.rs @@ -14,6 +14,7 @@ use super::network::{BridgeTuning, ClusterTransportTuning, NetworkTuning, WalTun use super::promql::PromqlTuning; use super::scheduler::SchedulerTuning; use super::shutdown::ShutdownTuning; +use super::startup::StartupTuning; /// Top-level tuning configuration. /// @@ -57,6 +58,8 @@ pub struct TuningConfig { pub calvin: CalvinTuning, #[serde(default)] pub promql: PromqlTuning, + #[serde(default)] + pub startup: StartupTuning, } impl TuningConfig { @@ -194,4 +197,20 @@ doc_cache_entries = 8192 assert_eq!(cfg.memory.overflow_max_bytes, 2 * 1024 * 1024 * 1024); assert_eq!(cfg.memory.doc_cache_entries, 8192); } + + #[test] + fn startup_bounds_are_read_from_the_aggregate() { + let cfg: TuningConfig = toml::from_str("").expect("deserialize"); + assert_eq!(cfg.startup.raft_ready_timeout_ms, 30_000); + assert_eq!(cfg.startup.data_group_recovery_timeout_ms, 60_000); + + let toml_str = r#" +[startup] +raft_ready_timeout_ms = 1234 +data_group_recovery_timeout_ms = 5678 +"#; + let cfg: TuningConfig = toml::from_str(toml_str).expect("deserialize"); + assert_eq!(cfg.startup.raft_ready_timeout_ms, 1234); + assert_eq!(cfg.startup.data_group_recovery_timeout_ms, 5678); + } } diff --git a/nodedb-types/src/config/tuning/mod.rs b/nodedb-types/src/config/tuning/mod.rs index 3ea956143..1dc5da6ad 100644 --- a/nodedb-types/src/config/tuning/mod.rs +++ b/nodedb-types/src/config/tuning/mod.rs @@ -11,6 +11,7 @@ mod network; mod promql; mod scheduler; mod shutdown; +mod startup; pub use bitemporal::BitemporalTuning; pub use calvin::CalvinTuning; @@ -27,3 +28,7 @@ pub use network::{BridgeTuning, ClusterTransportTuning, NetworkTuning, WalTuning pub use promql::PromqlTuning; pub use scheduler::SchedulerTuning; pub use shutdown::ShutdownTuning; +pub use startup::{ + DEFAULT_DATA_GROUP_RECOVERY_TIMEOUT_MS, DEFAULT_RAFT_READY_TIMEOUT_MS, + DataGroupRecoveryTimeout, RaftReadyTimeout, StartupTuning, +}; diff --git a/nodedb-types/src/config/tuning/startup.rs b/nodedb-types/src/config/tuning/startup.rs new file mode 100644 index 000000000..d8bd611ed --- /dev/null +++ b/nodedb-types/src/config/tuning/startup.rs @@ -0,0 +1,128 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Startup tuning: the stall bounds of the boot readiness waits. + +use std::time::Duration; + +use serde::{Deserialize, Serialize}; + +/// Default stall bound of the metadata group's replay wait. +pub const DEFAULT_RAFT_READY_TIMEOUT_MS: u64 = 30_000; + +/// Default stall bound of the data groups' replay wait and the Calvin +/// scheduler catch-up wait. +pub const DEFAULT_DATA_GROUP_RECOVERY_TIMEOUT_MS: u64 = 60_000; + +/// Stall bound of the metadata group's replay wait. +/// +/// A distinct type from [`DataGroupRecoveryTimeout`], so the two bounds +/// cannot swap places as adjacent arguments. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RaftReadyTimeout(pub Duration); + +/// Stall bound of the data groups' replay wait and the Calvin scheduler +/// catch-up wait. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DataGroupRecoveryTimeout(pub Duration); + +impl From for RaftReadyTimeout { + fn from(value: Duration) -> Self { + Self(value) + } +} + +impl From for DataGroupRecoveryTimeout { + fn from(value: Duration) -> Self { + Self(value) + } +} + +fn default_raft_ready_timeout_ms() -> u64 { + DEFAULT_RAFT_READY_TIMEOUT_MS +} + +fn default_data_group_recovery_timeout_ms() -> u64 { + DEFAULT_DATA_GROUP_RECOVERY_TIMEOUT_MS +} + +/// `[tuning.startup]`: how long each boot wait can go without progress. +/// +/// A wait that keeps progressing never fails, however long the replay takes. +/// The server range-checks both values at load. An unknown key is refused, so +/// a misspelt bound never leaves the default in force silently. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StartupTuning { + /// Stall bound of the metadata group's replay. Every applied entry resets + /// it. Default: 30_000. + #[serde(default = "default_raft_ready_timeout_ms")] + pub raft_ready_timeout_ms: u64, + + /// Stall bound of the data groups' replay, then of the Calvin scheduler + /// catch-up. An election, a commit or apply advance, a scheduler epoch, or + /// a finished group or scheduler resets it. Default: 60_000. + #[serde(default = "default_data_group_recovery_timeout_ms")] + pub data_group_recovery_timeout_ms: u64, +} + +impl Default for StartupTuning { + fn default() -> Self { + Self { + raft_ready_timeout_ms: default_raft_ready_timeout_ms(), + data_group_recovery_timeout_ms: default_data_group_recovery_timeout_ms(), + } + } +} + +impl StartupTuning { + /// Stall bound of the metadata group's replay wait. + pub fn raft_ready_timeout(&self) -> RaftReadyTimeout { + RaftReadyTimeout(Duration::from_millis(self.raft_ready_timeout_ms)) + } + + /// Stall bound of the data groups' replay and Calvin catch-up waits. + pub fn data_group_recovery_timeout(&self) -> DataGroupRecoveryTimeout { + DataGroupRecoveryTimeout(Duration::from_millis(self.data_group_recovery_timeout_ms)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_are_thirty_and_sixty_seconds() { + let cfg = StartupTuning::default(); + assert_eq!( + cfg.raft_ready_timeout(), + RaftReadyTimeout(Duration::from_secs(30)) + ); + assert_eq!( + cfg.data_group_recovery_timeout(), + DataGroupRecoveryTimeout(Duration::from_secs(60)) + ); + } + + #[test] + fn one_bound_set_leaves_the_other_at_its_default() { + let cfg: StartupTuning = + toml::from_str("raft_ready_timeout_ms = 1234").expect("deserialize"); + assert_eq!(cfg.raft_ready_timeout_ms, 1234); + assert_eq!( + cfg.data_group_recovery_timeout_ms, + DEFAULT_DATA_GROUP_RECOVERY_TIMEOUT_MS + ); + } + + #[test] + fn an_unknown_key_is_refused() { + let error = toml::from_str::("data_group_recovery_timeout = 1800000") + .expect_err("a key without the _ms suffix is not a bound"); + let message = error.to_string(); + assert!(message.contains("data_group_recovery_timeout"), "{message}"); + assert!( + message.contains("data_group_recovery_timeout_ms"), + "the message must list the keys that exist: {message}" + ); + } +} diff --git a/nodedb/src/bootstrap/calvin_catch_up.rs b/nodedb/src/bootstrap/calvin_catch_up.rs index 26b853a13..52730fc9f 100644 --- a/nodedb/src/bootstrap/calvin_catch_up.rs +++ b/nodedb/src/bootstrap/calvin_catch_up.rs @@ -12,14 +12,17 @@ //! caught up. A scheduler with nothing to rebuild reports at once, and a node //! that runs no scheduler passes at once. A vShard whose scheduler halted can //! never catch up, so its halt fails the wait at once. The wait shares the -//! data-group recovery bound and fails closed when the bound expires. +//! data-group recovery bound as a stall bound: it fails closed when no lagging +//! scheduler advances its fully-applied watermark or catches up for that long. use std::sync::Arc; use std::time::{Duration, Instant}; use tracing::{debug, info}; -use crate::bootstrap::data_group_recovery::DATA_GROUP_RECOVERY_TIMEOUT; +use nodedb_types::config::tuning::DataGroupRecoveryTimeout; + +use crate::bootstrap::stall_clock::StallClock; use crate::control::cluster::CalvinApplyHalt; use crate::control::state::SharedState; @@ -67,14 +70,28 @@ fn describe_lagging(lagging: &[u32]) -> String { } /// Hold startup until every Calvin scheduler on this node reaches its rebuild -/// target. -pub async fn await_calvin_catch_up(shared: &Arc) -> anyhow::Result<()> { +/// target. Fails once `stall_timeout` passes with no lagging scheduler +/// advancing its watermark or catching up. +pub async fn await_calvin_catch_up( + shared: &Arc, + stall_timeout: DataGroupRecoveryTimeout, +) -> anyhow::Result<()> { let registry = &shared.calvin.caught_up; - let deadline = Instant::now() + DATA_GROUP_RECOVERY_TIMEOUT; + let mut clock = StallClock::<1>::new(stall_timeout.0, Instant::now()); loop { let halt = shared.sequencer_halt.apply_halt().report(); - match catch_up_wait(registry.lagging(), halt) { + let schedulers = registry.lagging_schedulers(); + let now = Instant::now(); + clock.observe( + now, + schedulers.iter().map(|lagging| { + let mark = lagging.watermark.map_or(0, |w| w.saturating_add(1)); + (u64::from(lagging.vshard_id), [mark]) + }), + ); + let lagging = schedulers.iter().map(|lagging| lagging.vshard_id).collect(); + match catch_up_wait(lagging, halt) { None => { info!("every calvin scheduler reached its rebuild target"); return Ok(()); @@ -92,10 +109,11 @@ pub async fn await_calvin_catch_up(shared: &Arc) -> anyhow::Result< )); } Some(CatchUpWait::Lagging(lagging)) => { - if Instant::now() >= deadline { + if clock.stalled(now) { return Err(anyhow::anyhow!( - "calvin scheduler catch-up timeout after \ - {DATA_GROUP_RECOVERY_TIMEOUT:?}: {} behind their rebuild target", + "calvin scheduler catch-up made no progress for {:?}: {} behind their \ + rebuild target", + clock.bound(), describe_lagging(&lagging) )); } @@ -174,7 +192,7 @@ mod tests { ); let (dispatcher, _data_sides) = crate::bridge::Dispatcher::new(1, 64); let shared = SharedState::new(dispatcher, wal).expect("build shared state"); - await_calvin_catch_up(&shared) + await_calvin_catch_up(&shared, DataGroupRecoveryTimeout(Duration::from_secs(1))) .await .expect("no scheduler holds the gate"); } diff --git a/nodedb/src/bootstrap/cluster_ready.rs b/nodedb/src/bootstrap/cluster_ready.rs index b3627e86b..53544d387 100644 --- a/nodedb/src/bootstrap/cluster_ready.rs +++ b/nodedb/src/bootstrap/cluster_ready.rs @@ -7,7 +7,10 @@ use std::time::{Duration, Instant}; use tracing::info; +use nodedb_types::config::tuning::{DataGroupRecoveryTimeout, RaftReadyTimeout}; + use crate::bootstrap::schema_rehydrate::rehydrate_schema_registry; +use crate::bootstrap::stall_clock::StallClock; use crate::control::startup::ReadyGate; use crate::control::state::SharedState; @@ -30,6 +33,8 @@ pub async fn await_cluster_ready( mut raft_ready_rx: tokio::sync::watch::Receiver, data_plane_replay_done: Vec>>, gates: ClusterReadyGates, + raft_ready_timeout: RaftReadyTimeout, + data_group_recovery_timeout: DataGroupRecoveryTimeout, ) -> anyhow::Result<()> { let ClusterReadyGates { raft_gate, @@ -50,7 +55,7 @@ pub async fn await_cluster_ready( shared, &mut raft_ready_rx, &raft_gate, - RAFT_READY_STALL_TIMEOUT, + raft_ready_timeout, RAFT_READY_POLL_INTERVAL, ) .await?; @@ -156,7 +161,12 @@ pub async fn await_cluster_ready( // elections, so without this wait the gateway can open while a data // group's engines are still empty and an acknowledged write reads back as // if it never happened. Fail closed, like the replay wait above. - if let Err(e) = crate::bootstrap::data_group_recovery::await_data_group_recovery(shared).await { + if let Err(e) = crate::bootstrap::data_group_recovery::await_data_group_recovery( + shared, + data_group_recovery_timeout, + ) + .await + { data_groups_gate.fail(format!("data raft group recovery failed: {e}")); return Err(e); } @@ -165,7 +175,12 @@ pub async fn await_cluster_ready( // re-applies epochs from the sequencer log up to the highest epoch it had // applied before the restart. Hold the gateway until every scheduler on // this node reaches that target. Fail closed, like the wait above. - if let Err(e) = crate::bootstrap::calvin_catch_up::await_calvin_catch_up(shared).await { + if let Err(e) = crate::bootstrap::calvin_catch_up::await_calvin_catch_up( + shared, + data_group_recovery_timeout, + ) + .await + { data_groups_gate.fail(format!("calvin scheduler catch-up failed: {e}")); return Err(e); } @@ -281,7 +296,7 @@ pub async fn await_cluster_ready( Ok(_) => { crate::control::security::auth_lease::await_planning_admitted( shared, - RAFT_READY_STALL_TIMEOUT, + PLANNING_ADMISSION_TIMEOUT, ) .await } @@ -298,24 +313,28 @@ pub async fn await_cluster_ready( Ok(()) } -/// How long the metadata group may make NO replay progress before the boot -/// fails. Reset on every applied-index advance, so a large replay never trips -/// it — only a genuinely stuck group does. -const RAFT_READY_STALL_TIMEOUT: Duration = Duration::from_secs(30); - /// How often the stall check samples the applied index while waiting. const RAFT_READY_POLL_INTERVAL: Duration = Duration::from_secs(1); +/// How long boot waits for the first authorization lease before refusing to +/// open the gateway. +/// +/// Not the metadata group's stall bound, and not configurable with it: this is +/// a hard deadline on one grant, and it does not reset. The lease loop runs from +/// Raft start, so by the time boot reaches it every input of a grant is live; +/// 30 seconds is generous for a grant that is going to arrive at all. +const PLANNING_ADMISSION_TIMEOUT: Duration = Duration::from_secs(30); + /// Wait until the metadata raft group applies its first entry, and fail /// `raft_gate` when it does not. See [`raft_ready_progress`]. async fn wait_for_raft_ready( shared: &Arc, ready_rx: &mut tokio::sync::watch::Receiver, raft_gate: &ReadyGate, - stall_timeout: Duration, + stall_timeout: RaftReadyTimeout, poll_interval: Duration, ) -> anyhow::Result<()> { - match raft_ready_progress(shared, ready_rx, stall_timeout, poll_interval).await { + match raft_ready_progress(shared, ready_rx, stall_timeout.0, poll_interval).await { Ok(()) => { info!("metadata raft group ready — opening client listeners"); Ok(()) @@ -332,14 +351,20 @@ async fn wait_for_raft_ready( /// /// The same wait boot runs before it opens a listener. An in-process host /// that registers no startup gate calls this before it serves requests. +/// +/// `stall_timeout` is the caller's bound. Boot passes the configured +/// `[tuning.startup] raft_ready_timeout_ms` into [`await_cluster_ready`]; a +/// host with no config decides for itself rather than inheriting a value it +/// never read. pub async fn await_raft_ready( shared: &Arc, mut ready_rx: tokio::sync::watch::Receiver, + stall_timeout: RaftReadyTimeout, ) -> crate::Result<()> { raft_ready_progress( shared, &mut ready_rx, - RAFT_READY_STALL_TIMEOUT, + stall_timeout.0, RAFT_READY_POLL_INTERVAL, ) .await @@ -364,8 +389,8 @@ async fn raft_ready_progress( .map(|c| c.applied_index) .unwrap_or_else(|p| p.into_inner().applied_index) }; - let mut last_progress = Instant::now(); - let mut last_index = applied_index(); + let mut clock = StallClock::<1>::new(stall_timeout, Instant::now()); + clock.observe(Instant::now(), [(0, [applied_index()])]); loop { match tokio::time::timeout(poll_interval, ready_rx.wait_for(|v| *v)).await { @@ -378,17 +403,13 @@ async fn raft_ready_progress( // Not ready yet. Replay that is still advancing is healthy. Err(_) => { let current = applied_index(); - if current != last_index { - last_index = current; - last_progress = Instant::now(); - continue; - } - if last_progress.elapsed() >= stall_timeout { + let now = Instant::now(); + clock.observe(now, [(0, [current])]); + if clock.stalled(now) { return Err(crate::Error::Internal { detail: format!( "metadata group applied no entry for {stall_timeout:?} \ - (applied_index stuck at {current}) — it failed to apply its \ - first entry" + (applied_index stuck at {current})" ), }); } @@ -442,7 +463,7 @@ mod tests { &shared, &mut ready_rx, &raft_gate, - Duration::from_millis(100), + RaftReadyTimeout(Duration::from_millis(100)), Duration::from_millis(10), ) .await; @@ -465,7 +486,7 @@ mod tests { &shared, &mut ready_rx, &raft_gate, - Duration::from_millis(100), + RaftReadyTimeout(Duration::from_millis(100)), Duration::from_millis(10), ) .await @@ -475,4 +496,27 @@ mod tests { "the failure must name the stall, not a generic timeout: {err}" ); } + + /// `await_raft_ready` reports the bound its caller passes, not a default. + /// + /// This pins the argument into the metadata gate only. It does not cover + /// the `await_cluster_ready` call in `main.rs`, where the two bounds arrive + /// as adjacent arguments of different types. + #[tokio::test(flavor = "multi_thread")] + async fn the_bound_a_caller_passes_is_the_bound_the_failure_reports() { + let shared = test_shared_state(); + let (_ready_tx, ready_rx) = tokio::sync::watch::channel(false); + + let err = await_raft_ready( + &shared, + ready_rx, + RaftReadyTimeout(Duration::from_millis(50)), + ) + .await + .expect_err("a group that applies nothing must fail"); + assert!( + err.to_string().contains("50ms"), + "the failure must name the caller's bound, not the shipped default: {err}" + ); + } } diff --git a/nodedb/src/bootstrap/data_group_recovery.rs b/nodedb/src/bootstrap/data_group_recovery.rs index 6dbc89986..1ee6fea31 100644 --- a/nodedb/src/bootstrap/data_group_recovery.rs +++ b/nodedb/src/bootstrap/data_group_recovery.rs @@ -17,8 +17,10 @@ //! //! This gate closes that window: startup is held until every locally hosted //! data group has elected a leader, committed through the end of its retained -//! log, and applied that log on this node. It fails closed — a timeout aborts -//! startup rather than opening a gateway that would serve incomplete state. +//! log, and applied that log on this node. It fails closed: a wait that makes +//! no progress for the configured bound aborts startup rather than opening a +//! gateway that would serve incomplete state. A replay that keeps applying +//! entries keeps the wait open, however long the backlog is. //! //! A group whose replica here awaits a snapshot install does not hold startup. //! Its data diverged from the log, so it applies nothing until a snapshot @@ -28,19 +30,16 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use nodedb_raft::ReplicaState; +use nodedb_types::config::tuning::DataGroupRecoveryTimeout; use tracing::{debug, info, warn}; +use crate::bootstrap::stall_clock::StallClock; use crate::control::state::SharedState; /// How often the Raft status snapshot is re-read while waiting. Elections take /// seconds, so a coarse poll costs nothing and avoids a busy loop. const POLL_INTERVAL: Duration = Duration::from_millis(50); -/// Upper bound on the whole wait. Generous relative to a randomized election -/// timeout plus replay of a retained log, but finite: a group that cannot elect -/// or cannot apply is a failure, not a reason to hang forever. -pub const DATA_GROUP_RECOVERY_TIMEOUT: Duration = Duration::from_secs(60); - /// True when `group_id` names a data group whose log carries user writes that /// must be replayed into the Data Plane before queries are served. /// @@ -151,6 +150,19 @@ fn group_recovery(progress: GroupProgress) -> GroupRecovery { struct PendingGroup { group_id: u64, wait: GroupWait, + progress: GroupProgress, +} + +impl PendingGroup { + /// The measures the stall clock reads: whether the group has a leader, + /// how far its log is committed, and how far this node applied it. + fn measures(&self) -> [u64; 3] { + [ + u64::from(self.progress.leader_id != 0), + self.progress.commit_index, + self.progress.host_applied, + ] + } } impl std::fmt::Display for PendingGroup { @@ -205,6 +217,7 @@ fn gate_view( GroupRecovery::Pending(wait) => view.pending.push(PendingGroup { group_id: status.group_id, wait, + progress, }), } } @@ -215,13 +228,21 @@ fn gate_view( /// Raft log on this node. A group awaiting a snapshot install does not hold it. /// On success it sets `SharedState::data_groups_replayed`. /// +/// `stall_timeout` bounds how long the wait can go without progress. Progress +/// is a pending group electing a leader, committing more of its log, applying +/// more of it on this node, or finishing. A group that cannot elect or cannot +/// apply fails the wait. A group that keeps applying never does. +/// /// Refuses when `start_raft` installed no Raft status source. -pub async fn await_data_group_recovery(shared: &Arc) -> anyhow::Result<()> { +pub async fn await_data_group_recovery( + shared: &Arc, + stall_timeout: DataGroupRecoveryTimeout, +) -> anyhow::Result<()> { let Some(status_fn) = shared.raft_status_fn.get() else { anyhow::bail!("data group recovery: no Raft status source: start_raft has not run"); }; let status_fn = Arc::clone(status_fn); - let deadline = Instant::now() + DATA_GROUP_RECOVERY_TIMEOUT; + let mut clock = StallClock::<3>::new(stall_timeout.0, Instant::now()); loop { let view = gate_view(status_fn(), |group_id| { @@ -242,7 +263,14 @@ pub async fn await_data_group_recovery(shared: &Arc) -> anyhow::Res return Ok(()); } - if Instant::now() >= deadline { + let now = Instant::now(); + clock.observe( + now, + view.pending + .iter() + .map(|pending| (pending.group_id, pending.measures())), + ); + if clock.stalled(now) { let detail = view .pending .iter() @@ -250,7 +278,8 @@ pub async fn await_data_group_recovery(shared: &Arc) -> anyhow::Res .collect::>() .join("; "); return Err(anyhow::anyhow!( - "data raft group recovery timeout after {DATA_GROUP_RECOVERY_TIMEOUT:?}: {detail}" + "data raft group recovery made no progress for {:?}: {detail}", + clock.bound() )); } @@ -266,7 +295,10 @@ pub async fn await_data_group_recovery(shared: &Arc) -> anyhow::Res /// watch until it succeeds, without holding the caller. A host that brings /// nodes up one at a time uses it: a group here elects only once its peers /// run. Each failed attempt is logged, and the next one starts at once. -pub fn spawn_data_group_recovery(shared: &Arc) { +pub fn spawn_data_group_recovery( + shared: &Arc, + stall_timeout: DataGroupRecoveryTimeout, +) { let task_shared = Arc::clone(shared); crate::control::shutdown::spawn_loop( &shared.loop_registry, @@ -278,7 +310,7 @@ pub fn spawn_data_group_recovery(shared: &Arc) { tokio::select! { biased; _ = shutdown.wait_cancelled() => break, - recovered = await_data_group_recovery(&task_shared) => match recovered { + recovered = await_data_group_recovery(&task_shared, stall_timeout) => match recovered { Ok(()) => break, Err(error) => warn!(%error, "data raft group recovery did not finish; retrying"), }, diff --git a/nodedb/src/bootstrap/mod.rs b/nodedb/src/bootstrap/mod.rs index d57dfdff0..735077884 100644 --- a/nodedb/src/bootstrap/mod.rs +++ b/nodedb/src/bootstrap/mod.rs @@ -17,6 +17,7 @@ pub mod permission_tree_load; pub mod quota_replay; pub mod schema_rehydrate; pub mod signal; +pub mod stall_clock; pub mod state_wiring; pub mod tls; pub mod tracing_init; diff --git a/nodedb/src/bootstrap/stall_clock.rs b/nodedb/src/bootstrap/stall_clock.rs new file mode 100644 index 000000000..f70a3961e --- /dev/null +++ b/nodedb/src/bootstrap/stall_clock.rs @@ -0,0 +1,144 @@ +// SPDX-License-Identifier: BUSL-1.1 + +//! Progress clock for the boot readiness waits. +//! +//! A boot wait fails when it goes `bound` without progress, never on total +//! elapsed time. A node replaying a large backlog keeps making progress and +//! finishes, however long the replay takes. A wait that makes no progress fails +//! after `bound`. +//! +//! Each wait reports the items it still waits on. Each item carries `N` +//! progress measures, and each measure only counts upward. The clock resets +//! when a measure passes the highest value seen for its item, or when an item +//! stops waiting. A measure that drops and climbs back, such as a leader that +//! steps down and is elected again, does not reset the clock. + +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +/// Tracks the last time a boot wait made progress. +#[derive(Debug)] +pub(crate) struct StallClock { + bound: Duration, + last_progress: Instant, + /// Highest value seen of each measure, per item still waiting. + best: BTreeMap, +} + +impl StallClock { + /// A clock that starts at `now` with no item seen. + pub(crate) fn new(bound: Duration, now: Instant) -> Self { + Self { + bound, + last_progress: now, + best: BTreeMap::new(), + } + } + + /// The bound this clock fails at. + pub(crate) fn bound(&self) -> Duration { + self.bound + } + + /// Fold one sample of the items still waiting, taken at `now`. + /// + /// An item seen for the first time sets its baseline and is not progress. + pub(crate) fn observe( + &mut self, + now: Instant, + waiting: impl IntoIterator, + ) { + let mut progressed = false; + let mut still_waiting = BTreeMap::new(); + for (id, measures) in waiting { + let best = match self.best.remove(&id) { + Some(mut best) => { + for (seen, current) in best.iter_mut().zip(measures) { + if current > *seen { + *seen = current; + progressed = true; + } + } + best + } + None => measures, + }; + still_waiting.insert(id, best); + } + // Items left in `best` were waiting before and are done now. + if !self.best.is_empty() { + progressed = true; + } + self.best = still_waiting; + if progressed { + self.last_progress = now; + } + } + + /// Whether the wait has gone `bound` without progress at `now`. + pub(crate) fn stalled(&self, now: Instant) -> bool { + now.saturating_duration_since(self.last_progress) >= self.bound + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const BOUND: Duration = Duration::from_secs(10); + + fn at(start: Instant, secs: u64) -> Instant { + start + Duration::from_secs(secs) + } + + #[test] + fn no_progress_stalls_at_the_bound() { + let start = Instant::now(); + let mut clock = StallClock::<1>::new(BOUND, start); + clock.observe(at(start, 0), [(1, [5])]); + clock.observe(at(start, 9), [(1, [5])]); + assert!(!clock.stalled(at(start, 9))); + clock.observe(at(start, 10), [(1, [5])]); + assert!(clock.stalled(at(start, 10))); + } + + #[test] + fn a_rising_measure_resets_the_clock() { + let start = Instant::now(); + let mut clock = StallClock::<2>::new(BOUND, start); + clock.observe(at(start, 0), [(1, [0, 3])]); + clock.observe(at(start, 8), [(1, [0, 4])]); + clock.observe(at(start, 16), [(1, [0, 4])]); + assert!(!clock.stalled(at(start, 16))); + assert!(clock.stalled(at(start, 18))); + } + + #[test] + fn a_measure_that_drops_and_climbs_back_is_not_progress() { + let start = Instant::now(); + let mut clock = StallClock::<1>::new(BOUND, start); + clock.observe(at(start, 0), [(1, [1])]); + clock.observe(at(start, 4), [(1, [0])]); + clock.observe(at(start, 8), [(1, [1])]); + assert!(clock.stalled(at(start, 10))); + } + + #[test] + fn an_item_that_stops_waiting_is_progress() { + let start = Instant::now(); + let mut clock = StallClock::<1>::new(BOUND, start); + clock.observe(at(start, 0), [(1, [0]), (2, [0])]); + clock.observe(at(start, 8), [(2, [0])]); + assert!(!clock.stalled(at(start, 17))); + assert!(clock.stalled(at(start, 18))); + } + + #[test] + fn an_item_seen_for_the_first_time_is_not_progress() { + let start = Instant::now(); + let mut clock = StallClock::<1>::new(BOUND, start); + clock.observe(at(start, 0), [(1, [0])]); + clock.observe(at(start, 8), [(1, [0]), (2, [9])]); + assert!(clock.stalled(at(start, 10))); + } +} diff --git a/nodedb/src/config/server/config.rs b/nodedb/src/config/server/config.rs index 4c09766e3..21e916b5f 100644 --- a/nodedb/src/config/server/config.rs +++ b/nodedb/src/config/server/config.rs @@ -139,6 +139,30 @@ pub struct ServerConfig { pub scheduler: SchedulerConfig, } +/// Rejects a startup bound written at `[server]`, where nothing reads it, and +/// names the path that is read instead. `deny_unknown_fields` catches the key +/// anyway, but its message lists every valid field instead of the replacement. +fn reject_startup_bounds_at_the_server_path(content: &str) -> crate::Result<()> { + let Ok(doc) = toml::from_str::(content) else { + // A malformed document fails in the real parse below; nothing to + // inspect here. + return Ok(()); + }; + let Some(server) = doc.get("server").and_then(|v| v.as_table()) else { + return Ok(()); + }; + for key in ["raft_ready_timeout_ms", "data_group_recovery_timeout_ms"] { + if server.contains_key(key) { + return Err(crate::Error::Config { + detail: format!( + "`[server] {key}` is not read; set `[tuning.startup] {key}` instead" + ), + }); + } + } + Ok(()) +} + impl ServerConfig { /// Load configuration from a TOML file, falling back to defaults. pub fn from_file(path: &std::path::Path) -> crate::Result { @@ -150,6 +174,7 @@ impl ServerConfig { // that field — this is textual substitution before parsing, not a // second, competing expansion. let content = super::env_expand::expand_env(path, &content)?; + reject_startup_bounds_at_the_server_path(&content)?; let parsed: Self = toml::from_str(&content).map_err(|e| crate::Error::Config { detail: format!("invalid TOML config: {e}"), })?; @@ -596,4 +621,94 @@ mod tests { "unexpected error: {err}" ); } + + /// A startup bound written at `[server]` is rejected with the path that is + /// read named, not serde's full valid-field list. + #[test] + fn from_file_rejects_a_startup_bound_at_the_server_path() { + let path = write_temp_config( + "nodedb-server-path-startup-bound.toml", + "[server]\nraft_ready_timeout_ms = 300000\n", + ); + let err = ServerConfig::from_file(&path).unwrap_err(); + std::fs::remove_file(&path).ok(); + let msg = err.to_string(); + assert!(msg.contains("tuning.startup"), "{msg}"); + assert!(msg.contains("raft_ready_timeout_ms"), "{msg}"); + } + + /// The data-group bound is rejected at the same path. + #[test] + fn from_file_rejects_a_data_group_bound_at_the_server_path() { + let path = write_temp_config( + "nodedb-server-path-recovery-bound.toml", + "[server]\ndata_group_recovery_timeout_ms = 600000\n", + ); + let err = ServerConfig::from_file(&path).unwrap_err(); + std::fs::remove_file(&path).ok(); + let msg = err.to_string(); + assert!(msg.contains("tuning.startup"), "{msg}"); + assert!(msg.contains("data_group_recovery_timeout_ms"), "{msg}"); + } + + /// Both boot bounds live under `[tuning.startup]`; the other bound keeps + /// its default when only one is set. + #[test] + fn from_file_reads_the_startup_bounds_from_tuning() { + let path = write_temp_config( + "nodedb-startup-tuning.toml", + "[tuning.startup]\ndata_group_recovery_timeout_ms = 1234000\n", + ); + let cfg = ServerConfig::from_file(&path).expect("load config"); + std::fs::remove_file(&path).ok(); + assert_eq!(cfg.tuning.startup.data_group_recovery_timeout_ms, 1_234_000); + assert_eq!(cfg.tuning.startup.raft_ready_timeout_ms, 30_000); + } + + /// Neither bound is accepted outside the range a boot can wait for: zero + /// fails every boot on its first poll, `86400001` is past the stated ceiling, + /// and `u64::MAX` is not a bound at all. The message names the key, the value + /// and the range, so the operator does not have to read the source. + #[test] + fn from_file_rejects_a_startup_bound_out_of_range() { + for field in ["raft_ready_timeout_ms", "data_group_recovery_timeout_ms"] { + for value in ["0", "86400001", "18446744073709551615"] { + let path = write_temp_config( + "nodedb-startup-bound-out-of-range.toml", + &format!("[tuning.startup]\n{field} = {value}\n"), + ); + let err = ServerConfig::from_file(&path).unwrap_err(); + std::fs::remove_file(&path).ok(); + let msg = err.to_string(); + assert!( + msg.contains(field), + "the message must name the key for {field} = {value}: {msg}" + ); + assert!( + msg.contains(&format!( + "invalid value '{value}' for tuning.startup.{field}" + )), + "the message must name the value and the key for {field} = {value}: {msg}" + ); + assert!( + msg.contains("86400000"), + "the message must name the accepted range: {msg}" + ); + } + } + } + + /// Both ends of the accepted range load, and the value arrives intact. + #[test] + fn from_file_accepts_the_ends_of_the_startup_bound_range() { + for value in [1u64, 86_400_000] { + let path = write_temp_config( + "nodedb-startup-bound-in-range.toml", + &format!("[tuning.startup]\nraft_ready_timeout_ms = {value}\n"), + ); + let cfg = ServerConfig::from_file(&path).expect("a bound in range loads"); + std::fs::remove_file(&path).ok(); + assert_eq!(cfg.tuning.startup.raft_ready_timeout_ms, value); + } + } } diff --git a/nodedb/src/config/server/domain.rs b/nodedb/src/config/server/domain.rs index 375d79b4c..188584526 100644 --- a/nodedb/src/config/server/domain.rs +++ b/nodedb/src/config/server/domain.rs @@ -34,6 +34,18 @@ pub(super) const REDO_ENTRY_ENVELOPE_BYTES: usize = 1024 * 1024; pub(super) const MAX_REDO_ENTRY_BYTES: usize = nodedb_cluster::rpc_codec::MAX_RPC_PAYLOAD_SIZE as usize - REDO_ENTRY_ENVELOPE_BYTES; +/// Smallest boot bound a caller can set, in milliseconds. +/// +/// A zero bound fails every boot on its first poll: the wait is over before the +/// group it waits for can answer. +pub(super) const MIN_STARTUP_BOUND_MS: u64 = 1; + +/// Largest boot bound, in milliseconds: one day. +/// +/// A bound past this stops being a bound: a stuck group and a healthy one are +/// indistinguishable for the whole window. +pub(super) const MAX_STARTUP_BOUND_MS: u64 = 86_400_000; + /// Rejects an endpoint that carries no `http://` or `https://` host. pub(super) fn otlp_endpoint_has_host(raw: &str) -> bool { raw.strip_prefix("http://") @@ -54,6 +66,22 @@ pub(super) fn positive_u64(value: u64, field: &str) -> crate::Result<()> { Ok(()) } +/// A boot bound, in milliseconds, inside the range a boot can wait for. +/// +/// Zero fails every boot on its first poll: the wait is over before the group it +/// waits for can answer. A bound near `u64::MAX` is not a bound at all, and the +/// server refuses it here instead of letting a stuck boot look healthy. +fn startup_bound(value: u64, field: &str) -> crate::Result<()> { + if !(MIN_STARTUP_BOUND_MS..=MAX_STARTUP_BOUND_MS).contains(&value) { + return Err(reject( + field, + value, + &format!("a bound from {MIN_STARTUP_BOUND_MS} to {MAX_STARTUP_BOUND_MS} ms (one day)"), + )); + } + Ok(()) +} + fn positive_usize(value: usize, field: &str) -> crate::Result<()> { if value == 0 { return Err(reject(field, value, "a positive integer")); @@ -88,6 +116,16 @@ pub(super) fn validate_domain(config: &ServerConfig) -> crate::Result<()> { super::pitr::validate_pitr(config)?; super::backup::validate_backup(config)?; + let s = &config.tuning.startup; + startup_bound( + s.raft_ready_timeout_ms, + "tuning.startup.raft_ready_timeout_ms", + )?; + startup_bound( + s.data_group_recovery_timeout_ms, + "tuning.startup.data_group_recovery_timeout_ms", + )?; + let ts = &config.tuning.timeseries; positive_usize( ts.memtable_budget_bytes, diff --git a/nodedb/src/control/cluster/calvin/scheduler/caught_up.rs b/nodedb/src/control/cluster/calvin/scheduler/caught_up.rs index 1f9a24ba6..0acebcf14 100644 --- a/nodedb/src/control/cluster/calvin/scheduler/caught_up.rs +++ b/nodedb/src/control/cluster/calvin/scheduler/caught_up.rs @@ -13,12 +13,34 @@ //! removes its entry when the scheduler is dropped, so the registry lists //! only running schedulers. A restarted scheduler replaces its predecessor's //! entry, and the predecessor's drop leaves the new entry in place. +//! +//! Each entry also carries the scheduler's fully-applied watermark. Startup +//! reads it as progress: a lagging scheduler whose watermark rises is still +//! catching up, and one whose watermark stays put is stuck. use std::collections::BTreeMap; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -type Entries = Arc>>>; +type Entries = Arc>>>; + +/// One running scheduler's state in the registry. +#[derive(Debug, Default)] +struct Entry { + caught_up: AtomicBool, + /// The highest fully-applied watermark reported plus one. Zero means none + /// reported yet. + watermark_mark: AtomicU64, +} + +/// A running scheduler that has not reached its rebuild target. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct LaggingScheduler { + pub vshard_id: u32, + /// The highest fully-applied watermark it reported, or `None` before its + /// first. + pub watermark: Option, +} /// Caught-up state of every running scheduler on this node, by vShard. #[derive(Debug, Default)] @@ -29,26 +51,38 @@ pub struct CaughtUpRegistry { impl CaughtUpRegistry { /// Register the scheduler starting for `vshard_id`, not caught up yet. pub fn register(&self, vshard_id: u32) -> CaughtUpHandle { - let flag = Arc::new(AtomicBool::new(false)); + let entry = Arc::new(Entry::default()); self.entries .lock() .unwrap_or_else(|p| p.into_inner()) - .insert(vshard_id, Arc::clone(&flag)); + .insert(vshard_id, Arc::clone(&entry)); CaughtUpHandle { vshard_id, - flag, + entry, entries: Arc::clone(&self.entries), } } /// The vShards whose running scheduler is not caught up, in order. pub fn lagging(&self) -> Vec { + self.lagging_schedulers() + .into_iter() + .map(|lagging| lagging.vshard_id) + .collect() + } + + /// The running schedulers that are not caught up, by vShard, with the + /// watermark each reported. + pub fn lagging_schedulers(&self) -> Vec { self.entries .lock() .unwrap_or_else(|p| p.into_inner()) .iter() - .filter(|(_, flag)| !flag.load(Ordering::Acquire)) - .map(|(vshard_id, _)| *vshard_id) + .filter(|(_, entry)| !entry.caught_up.load(Ordering::Acquire)) + .map(|(vshard_id, entry)| LaggingScheduler { + vshard_id: *vshard_id, + watermark: entry.watermark_mark.load(Ordering::Acquire).checked_sub(1), + }) .collect() } } @@ -57,20 +91,28 @@ impl CaughtUpRegistry { #[derive(Debug)] pub struct CaughtUpHandle { vshard_id: u32, - flag: Arc, + entry: Arc, entries: Entries, } impl CaughtUpHandle { /// Whether this scheduler reported caught up. pub fn is_caught_up(&self) -> bool { - self.flag.load(Ordering::Acquire) + self.entry.caught_up.load(Ordering::Acquire) + } + + /// Report the scheduler's fully-applied watermark. A lower watermark than + /// one already reported changes nothing. + pub fn report_watermark(&self, watermark: u64) { + self.entry + .watermark_mark + .fetch_max(watermark.saturating_add(1), Ordering::Release); } /// Report this scheduler caught up. The state never returns to lagging: /// the fully-applied watermark only rises. pub fn mark_caught_up(&self) { - self.flag.store(true, Ordering::Release); + self.entry.caught_up.store(true, Ordering::Release); } } @@ -79,7 +121,7 @@ impl Drop for CaughtUpHandle { let mut entries = self.entries.lock().unwrap_or_else(|p| p.into_inner()); if entries .get(&self.vshard_id) - .is_some_and(|flag| Arc::ptr_eq(flag, &self.flag)) + .is_some_and(|entry| Arc::ptr_eq(entry, &self.entry)) { entries.remove(&self.vshard_id); } @@ -105,6 +147,27 @@ mod tests { assert!(registry.lagging().is_empty()); } + #[test] + fn a_lagging_scheduler_reports_its_highest_watermark() { + let registry = CaughtUpRegistry::default(); + let handle = registry.register(4); + assert_eq!( + registry.lagging_schedulers(), + vec![LaggingScheduler { + vshard_id: 4, + watermark: None, + }] + ); + + handle.report_watermark(0); + handle.report_watermark(7); + handle.report_watermark(3); + assert_eq!(registry.lagging_schedulers()[0].watermark, Some(7)); + + handle.mark_caught_up(); + assert!(registry.lagging_schedulers().is_empty()); + } + #[test] fn a_dropped_scheduler_leaves_the_registry() { let registry = CaughtUpRegistry::default(); diff --git a/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs b/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs index 089354880..d5227ff00 100644 --- a/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs +++ b/nodedb/src/control/cluster/calvin/scheduler/driver/core/scheduler.rs @@ -396,6 +396,7 @@ impl Scheduler { ) { self.metrics.update_last_applied_epoch(watermark); self.ledger.fold(watermark); + self.caught_up.report_watermark(watermark); self.shared .calvin .last_applied_epoch diff --git a/nodedb/src/control/cluster/calvin/scheduler/mod.rs b/nodedb/src/control/cluster/calvin/scheduler/mod.rs index f9d862066..fcfcdd4dd 100644 --- a/nodedb/src/control/cluster/calvin/scheduler/mod.rs +++ b/nodedb/src/control/cluster/calvin/scheduler/mod.rs @@ -18,7 +18,7 @@ pub mod recovery; pub use applied_gate::AppliedGate; pub use applied_ledger::{CalvinAppliedLedger, CalvinAppliedLedgers, ClaimRefusal}; pub use barrier_log::{BarrierEvent, BarrierLog, BarrierOutcome, ReceivedReads}; -pub use caught_up::{CaughtUpHandle, CaughtUpRegistry}; +pub use caught_up::{CaughtUpHandle, CaughtUpRegistry, LaggingScheduler}; pub use driver::{ CalvinReadResultProposal, RaftSequencerProposer, Scheduler, SchedulerConfig, SchedulerParams, SequencerProposer, propose_calvin_read_result, diff --git a/nodedb/src/control/cluster/left_vshard_forget.rs b/nodedb/src/control/cluster/left_vshard_forget.rs index 265ce2f53..3992109d9 100644 --- a/nodedb/src/control/cluster/left_vshard_forget.rs +++ b/nodedb/src/control/cluster/left_vshard_forget.rs @@ -345,9 +345,14 @@ mod tests { // The group elects, and boot recovery finishes. leader.store(NODE, std::sync::atomic::Ordering::Release); - crate::bootstrap::data_group_recovery::await_data_group_recovery(&state) - .await - .expect("recovery"); + crate::bootstrap::data_group_recovery::await_data_group_recovery( + &state, + nodedb_types::config::tuning::DataGroupRecoveryTimeout(std::time::Duration::from_secs( + 60, + )), + ) + .await + .expect("recovery"); assert!( state .data_groups_replayed diff --git a/nodedb/src/control/cluster/test_one_node.rs b/nodedb/src/control/cluster/test_one_node.rs index 7e97b2ee7..1403b63a7 100644 --- a/nodedb/src/control/cluster/test_one_node.rs +++ b/nodedb/src/control/cluster/test_one_node.rs @@ -104,9 +104,13 @@ pub(crate) async fn boot_with_core( .lock() .unwrap_or_else(|p| p.into_inner()) .take(); - crate::bootstrap::cluster_ready::await_raft_ready(&state, ready) - .await - .expect("the metadata group applies its first entry"); + crate::bootstrap::cluster_ready::await_raft_ready( + &state, + ready, + nodedb_types::config::tuning::StartupTuning::default().raft_ready_timeout(), + ) + .await + .expect("the metadata group applies its first entry"); OneNodeCluster { state, diff --git a/nodedb/src/control/security/auth_lease/status.rs b/nodedb/src/control/security/auth_lease/status.rs index 00727d625..620076982 100644 --- a/nodedb/src/control/security/auth_lease/status.rs +++ b/nodedb/src/control/security/auth_lease/status.rs @@ -82,6 +82,12 @@ pub(crate) fn lease_status_of( /// Wait until this node can plan permission-checked statements, or refuse /// once `timeout` passes. +/// +/// `timeout` is a hard deadline on the first lease grant, not the metadata +/// group's stall bound. The two waits end differently: the stall bound resets +/// every time the group applies an entry, and this one never does, so a caller +/// that has no reason of its own passes its own constant rather than the one +/// the group uses. pub async fn await_planning_admitted(state: &SharedState, timeout: Duration) -> crate::Result<()> { if planning_admitted_within(state, timeout).await { return Ok(()); diff --git a/nodedb/src/main.rs b/nodedb/src/main.rs index 5d90d43eb..4251004ad 100644 --- a/nodedb/src/main.rs +++ b/nodedb/src/main.rs @@ -291,6 +291,8 @@ async fn server_main() -> anyhow::Result<()> { health_loop_gate, gateway_enable_gate, }, + config.tuning.startup.raft_ready_timeout(), + config.tuning.startup.data_group_recovery_timeout(), ) .await?; From a13914f599c8e80c2586fccc7571da7a3e68c9c9 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Thu, 15 Oct 2026 01:36:50 +0800 Subject: [PATCH 2/5] fix(wal): name the WAL format version when startup refuses a store A segment written in a WAL format this build cannot read was reported as corruption, which sends an operator toward repair instead of toward the build that wrote the data. Readers now judge the format at the segment's first record and return SegmentFormatVersion with the segment path and both versions. Startup turns it into a version-gap message that lists the actions that exist. A zeroed or damaged version field is still corruption, not a format gap, and only a record whose checksum matches its declared version counts as another format. The restore walkers report the same error. --- nodedb-wal/src/error.rs | 36 +++ nodedb-wal/src/lazy_reader.rs | 110 ++++++++- nodedb-wal/src/mmap_reader/reader.rs | 107 +++++++- nodedb-wal/src/mmap_reader/replay.rs | 10 +- nodedb-wal/src/reader.rs | 144 ++++++++++- nodedb-wal/src/record/foreign_version.rs | 146 +++++++++++ nodedb-wal/src/record/mod.rs | 2 + nodedb-wal/src/recovery.rs | 85 ++++++- nodedb-wal/src/segmented.rs | 5 +- nodedb/src/bootstrap/wal_init.rs | 21 +- nodedb/src/ctl/restore/cut_rule.rs | 32 ++- nodedb/src/ctl/restore/segment.rs | 166 ++++++++++++- nodedb/src/wal/manager/replay.rs | 297 ++++++++++++++++++++++- 13 files changed, 1126 insertions(+), 35 deletions(-) create mode 100644 nodedb-wal/src/record/foreign_version.rs diff --git a/nodedb-wal/src/error.rs b/nodedb-wal/src/error.rs index 58c38dd1e..b092e9eae 100644 --- a/nodedb-wal/src/error.rs +++ b/nodedb-wal/src/error.rs @@ -31,6 +31,22 @@ pub enum WalError { #[error("unsupported WAL format version {version} (supported: {supported})")] UnsupportedVersion { version: u16, supported: u16 }, + /// A segment holds records in a WAL format version this build cannot read. + /// + /// The same finding as [`WalError::UnsupportedVersion`], reported by the + /// callers that know which file carries it. A reader sees one header at a + /// time and holds no path, and a version alone does not tell an operator + /// which segment to act on. + #[error( + "WAL segment '{path}' holds records in WAL format version {version}; this build reads \ + version {supported}" + )] + SegmentFormatVersion { + path: String, + version: u16, + supported: u16, + }, + /// Unknown required record type encountered during replay. /// Optional unknown record types are safely skipped. #[error("unknown required record type {record_type} at LSN {lsn}")] @@ -214,4 +230,24 @@ pub enum WalError { }, } +impl WalError { + /// Name the segment a version gap was found in. + /// + /// A reader reports [`WalError::UnsupportedVersion`] from header validation + /// alone and holds no path. The callers that opened the file do, and only + /// they can turn the finding into something an operator can act on. Every + /// other error passes through unchanged. + #[cfg(not(target_arch = "wasm32"))] + pub(crate) fn with_segment_path(self, path: &std::path::Path) -> Self { + match self { + Self::UnsupportedVersion { version, supported } => Self::SegmentFormatVersion { + path: path.display().to_string(), + version, + supported, + }, + other => other, + } + } +} + pub type Result = std::result::Result; diff --git a/nodedb-wal/src/lazy_reader.rs b/nodedb-wal/src/lazy_reader.rs index 3fb315548..05ea9f380 100644 --- a/nodedb-wal/src/lazy_reader.rs +++ b/nodedb-wal/src/lazy_reader.rs @@ -39,7 +39,11 @@ use crate::crypto::KeyRing; use crate::error::{Result, WalError}; use crate::preamble::{SegmentPreamble, WAL_PREAMBLE_MAGIC, read_leading_preamble}; use crate::reader::StopReason; -use crate::record::{HEADER_SIZE, MAX_WAL_PAYLOAD_SIZE, RecordHeader, RecordType, WalRecord}; +use crate::record::foreign_version::read_declared_payload; +use crate::record::{ + HEADER_SIZE, MAX_WAL_PAYLOAD_SIZE, RecordHeader, RecordType, WalRecord, is_format_gap, + may_be_format_gap, +}; use crate::segment::SegmentDecryptor; fn checked_offset_add(offset: u64, len: u64) -> Result { @@ -55,6 +59,10 @@ fn checked_offset_add(offset: u64, len: u64) -> Result { pub struct LazyWalReader { file: File, offset: u64, + /// Offset of the segment's first record, past any preamble. The format + /// version is judged here and nowhere else: one writer never mixes formats + /// inside one segment. + records_start: u64, /// Preamble read from offset 0 of this segment (present when encryption is /// active). The epoch is part of the AAD used to decrypt payloads. segment_preamble: Option, @@ -94,6 +102,7 @@ impl LazyWalReader { Ok(Self { file, offset: start_offset, + records_start: start_offset, segment_preamble, double_write, stop_reason: None, @@ -123,9 +132,10 @@ impl LazyWalReader { /// Read the next record header without reading the payload. /// - /// Returns `None` at EOF or first corruption. After this call, use - /// either `read_payload()` to get the payload or `skip_payload()` to - /// seek past it. + /// Returns `None` at EOF or first corruption. Returns `Err` for a format + /// version this build cannot read at the segment's first record, and the + /// caller must stop at that `Err`. After a header, use either + /// `read_payload()` to get the payload or `skip_payload()` to seek past it. /// /// Alignment padding records are consumed internally — the caller only /// ever sees real records. @@ -153,6 +163,20 @@ impl LazyWalReader { match header.validate(record_offset) { Ok(()) => {} Err(error @ WalError::PayloadTooLarge { .. }) => return Err(error), + // Another build's format travels out so the refusal can name + // the version. A damaged header in this build's format is a + // stop for `torn_tail` to classify, like any damage. + Err(error @ WalError::UnsupportedVersion { .. }) + if may_be_format_gap(&header, record_offset == self.records_start) => + { + let payload = read_declared_payload(&mut self.file, &header, self.offset)?; + if is_format_gap(&header, payload.as_deref()) { + return Err(error); + } + return self.stop(StopReason::Corruption { + offset: record_offset, + }); + } Err(_) => { return self.stop(StopReason::Corruption { offset: record_offset, @@ -345,7 +369,10 @@ where { let mut reader = LazyWalReader::open(path, keys)?; let mut last_lsn = 0u64; - while let Some(header) = reader.next_header()? { + while let Some(header) = reader + .next_header() + .map_err(|error| error.with_segment_path(path))? + { last_lsn = header.lsn; handler(&mut reader, &header)?; } @@ -541,4 +568,77 @@ mod tests { let mut reader = LazyWalReader::open(&path, None).unwrap(); assert!(reader.next_header().unwrap().is_none()); } + + /// A version this build cannot read at the segment's first record is a + /// typed error that names the version. + #[test] + fn an_unknown_format_version_at_the_first_record_is_an_error() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("unknown.wal"); + { + let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"x") + .unwrap(); + writer.sync().unwrap(); + } + + let mut bytes = std::fs::read(&path).unwrap(); + crate::record::foreign_version::reseal_with_version(&mut bytes, 0, 1); + std::fs::write(&path, &bytes).unwrap(); + + let mut reader = LazyWalReader::open(&path, None).unwrap(); + match reader.next_header() { + Err(WalError::UnsupportedVersion { version, supported }) => { + assert_eq!(version, 1); + assert_eq!(supported, crate::record::WAL_FORMAT_VERSION); + } + other => panic!("expected UnsupportedVersion, got {other:?}"), + } + } + + /// A version this build cannot read, found after an intact record, is + /// damage: the stream stops there as corruption, not as a format gap. + #[test] + fn a_version_mismatch_after_the_first_record_is_damage() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("two-records.wal"); + { + let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"first") + .unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"second") + .unwrap(); + writer.sync().unwrap(); + } + + let mut reader = LazyWalReader::open(&path, None).unwrap(); + let first = reader.next_header().unwrap().expect("the first header"); + reader.skip_payload(&first).unwrap(); + let second_at = reader.offset(); + + let mut bytes = std::fs::read(&path).unwrap(); + let version_at = usize::try_from(second_at).unwrap() + 4; + bytes[version_at..version_at + 2].copy_from_slice(&1u16.to_le_bytes()); + std::fs::write(&path, &bytes).unwrap(); + + let mut reader = LazyWalReader::open(&path, None).unwrap(); + let first = reader + .next_header() + .unwrap() + .expect("the first record is still readable"); + assert_eq!(first.lsn, 1); + reader.skip_payload(&first).unwrap(); + assert!( + reader.next_header().unwrap().is_none(), + "the mismatch stops the stream instead of failing the read" + ); + assert_eq!( + reader.stop_reason(), + Some(StopReason::Corruption { offset: second_at }), + "a mismatch after the first record is damage, not a format gap" + ); + } } diff --git a/nodedb-wal/src/mmap_reader/reader.rs b/nodedb-wal/src/mmap_reader/reader.rs index d60b43f57..08fd9e250 100644 --- a/nodedb-wal/src/mmap_reader/reader.rs +++ b/nodedb-wal/src/mmap_reader/reader.rs @@ -34,7 +34,9 @@ use crate::crypto::KeyRing; use crate::error::{Result, WalError}; use crate::preamble::{PREAMBLE_SIZE, SegmentPreamble, WAL_PREAMBLE_MAGIC, parse_leading_preamble}; use crate::reader::StopReason; -use crate::record::{HEADER_SIZE, RecordHeader, RecordType, WAL_MAGIC, WalRecord}; +use crate::record::{ + HEADER_SIZE, RecordHeader, RecordType, WAL_MAGIC, WalRecord, is_format_gap, may_be_format_gap, +}; use crate::segment::SegmentDecryptor; fn checked_range_end(start: usize, len: usize, limit: usize) -> Result> { @@ -113,6 +115,10 @@ fn fadv_dontneed(fd: &std::fs::File, len: usize, path: &Path) { pub struct MmapWalReader { mmap: Mmap, offset: usize, + /// Offset of the segment's first record, past any preamble. The format + /// version is judged here and nowhere else: one writer never mixes formats + /// inside one segment. + records_start: usize, /// Preamble mapped at offset 0 of this segment (present when encryption is /// active). The epoch is part of the AAD used to decrypt payloads. segment_preamble: Option, @@ -194,6 +200,7 @@ impl MmapWalReader { Ok(Self { mmap, offset, + records_start: offset, segment_preamble, file, path: path.to_path_buf(), @@ -245,7 +252,9 @@ impl MmapWalReader { /// Read the next record from the mmap'd region. /// - /// Returns `None` at EOF or at the first corruption point. + /// Returns `None` at EOF or at the first corruption point. Returns `Err` + /// for a format version this build cannot read at the segment's first + /// record, and the caller must stop at that `Err`. /// Header parsing avoids extra copies, but the payload is copied out of /// the mmap'd region into an owned `Vec` on `WalRecord`: records must /// outlive the reader (they cross thread boundaries in parallel replay @@ -299,6 +308,23 @@ impl MmapWalReader { match header.validate(header_offset) { Ok(()) => {} Err(error @ WalError::PayloadTooLarge { .. }) => return Err(error), + // Another build's format travels out so the refusal can name + // the version. A damaged header in this build's format is a + // stop for `torn_tail` to classify, like any damage. + Err(error @ WalError::UnsupportedVersion { .. }) + if may_be_format_gap(&header, self.offset == self.records_start) => + { + let payload = usize::try_from(header.payload_len) + .ok() + .filter(|len| *len <= crate::record::MAX_WAL_PAYLOAD_SIZE) + .and_then(|len| data.get(header_end..header_end.checked_add(len)?)); + if is_format_gap(&header, payload) { + return Err(error); + } + return self.stop(StopReason::Corruption { + offset: header_offset, + }); + } Err(_) => { return self.stop(StopReason::Corruption { offset: header_offset, @@ -517,6 +543,83 @@ mod tests { )); } + /// A version this build cannot read at the segment's first record is a + /// typed error that names the version. + #[test] + fn an_unknown_format_version_at_the_first_record_is_an_error() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("unknown.wal"); + { + let mut writer = test_writer(&path); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"x") + .unwrap(); + writer.sync().unwrap(); + } + + let mut bytes = std::fs::read(&path).unwrap(); + crate::record::foreign_version::reseal_with_version(&mut bytes, 0, 1); + std::fs::write(&path, &bytes).unwrap(); + + let mut reader = MmapWalReader::open(&path, None).unwrap(); + match reader.next_record() { + Err(WalError::UnsupportedVersion { version, supported }) => { + assert_eq!(version, 1); + assert_eq!(supported, crate::record::WAL_FORMAT_VERSION); + } + other => panic!("expected UnsupportedVersion, got {other:?}"), + } + } + + /// A version this build cannot read, found after an intact record, is + /// damage: the stream stops there as corruption, not as a format gap. + #[test] + fn a_version_mismatch_after_the_first_record_is_damage() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("two-records.wal"); + { + let mut writer = test_writer(&path); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"first") + .unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"second") + .unwrap(); + writer.sync().unwrap(); + } + + let second_at = { + let mut reader = MmapWalReader::open(&path, None).unwrap(); + reader + .next_record() + .unwrap() + .expect("the first record reads"); + reader.offset() + }; + + let mut bytes = std::fs::read(&path).unwrap(); + let version_at = second_at + 4; + bytes[version_at..version_at + 2].copy_from_slice(&1u16.to_le_bytes()); + std::fs::write(&path, &bytes).unwrap(); + + let mut reader = MmapWalReader::open(&path, None).unwrap(); + let first = reader + .next_record() + .unwrap() + .expect("the first record is still readable"); + assert_eq!(first.header.lsn, 1); + assert!( + reader.next_record().unwrap().is_none(), + "the mismatch stops the stream instead of failing the read" + ); + let second_at = u64::try_from(second_at).unwrap(); + assert_eq!( + reader.stop_reason(), + Some(StopReason::Corruption { offset: second_at }), + "a mismatch after the first record is damage, not a format gap" + ); + } + #[test] fn checked_range_end_rejects_overflow_and_short_ranges() { assert!(checked_range_end(usize::MAX, 1, usize::MAX).is_err()); diff --git a/nodedb-wal/src/mmap_reader/replay.rs b/nodedb-wal/src/mmap_reader/replay.rs index 7c4e172fb..4d82e16c2 100644 --- a/nodedb-wal/src/mmap_reader/replay.rs +++ b/nodedb-wal/src/mmap_reader/replay.rs @@ -103,7 +103,10 @@ fn scan_segment( let mut reader = MmapWalReader::open(&segment.path, keys)?; let mut records = Vec::new(); let mut last_lsn = 0u64; - while let Some(record) = reader.next_record()? { + while let Some(record) = reader + .next_record() + .map_err(|error| error.with_segment_path(&segment.path))? + { last_lsn = record.header.lsn; if record.header.lsn >= from_lsn { records.push(record); @@ -211,7 +214,10 @@ pub fn replay_segments_mmap_limit( continuity.check(seg)?; let mut reader = MmapWalReader::open(&seg.path, keys)?; let mut last_lsn = 0u64; - while let Some(record) = reader.next_record()? { + while let Some(record) = reader + .next_record() + .map_err(|error| error.with_segment_path(&seg.path))? + { last_lsn = record.header.lsn; if record.header.lsn >= from_lsn { records.push(record); diff --git a/nodedb-wal/src/reader.rs b/nodedb-wal/src/reader.rs index ce46141ab..2f4df7553 100644 --- a/nodedb-wal/src/reader.rs +++ b/nodedb-wal/src/reader.rs @@ -29,7 +29,10 @@ use std::path::Path; use crate::crypto::KeyRing; use crate::error::{Result, WalError}; use crate::preamble::{SegmentPreamble, WAL_PREAMBLE_MAGIC, read_leading_preamble}; -use crate::record::{HEADER_SIZE, RecordHeader, RecordType, WalRecord}; +use crate::record::foreign_version::read_declared_payload; +use crate::record::{ + HEADER_SIZE, RecordHeader, RecordType, WalRecord, is_format_gap, may_be_format_gap, +}; use crate::segment::SegmentDecryptor; fn checked_offset_add(offset: u64, len: usize) -> Result { @@ -58,8 +61,9 @@ pub enum StopReason { /// The segment ended cleanly on a record boundary. Eof, - /// A record at this offset failed structural validation (bad magic, - /// unsupported version, short read, or checksum mismatch). + /// A record at this offset failed structural validation (bad magic, a + /// zeroed format version, a short read, a checksum mismatch, or a format + /// version this build cannot read found after the segment's first record). Corruption { offset: u64 }, } @@ -67,6 +71,10 @@ pub enum StopReason { pub struct WalReader { file: File, offset: u64, + /// Offset of the segment's first record, past any preamble. The format + /// version is judged here and nowhere else: one writer never mixes formats + /// inside one segment. + records_start: u64, /// Preamble read from offset 0 of this segment (present when encryption /// is active). The epoch is used as part of the AAD for decryption. segment_preamble: Option, @@ -133,6 +141,7 @@ impl WalReader { Ok(Self { file, offset: start_offset, + records_start: start_offset, segment_preamble, double_write, stop_reason: None, @@ -185,7 +194,10 @@ impl WalReader { /// Read the next record from the WAL. /// /// Returns `None` at EOF (clean end) or at the first corruption point. - /// Returns `Err` only for I/O errors or unknown required record types. + /// Returns `Err` for I/O errors, unknown required record types, an + /// oversized payload declaration, and a format version this build cannot + /// read at the segment's first record. A reader that returns `Err` for a + /// header has consumed that header, so a caller must stop at the `Err`. pub fn next_record(&mut self) -> Result> { loop { // Read header. @@ -230,6 +242,20 @@ impl WalReader { match header.validate(header_offset) { Ok(()) => {} Err(error @ WalError::PayloadTooLarge { .. }) => return Err(error), + // Another build's format travels out so the refusal can name + // the version. A damaged header in this build's format is a + // stop for `torn_tail` to classify, like any damage. + Err(error @ WalError::UnsupportedVersion { .. }) + if may_be_format_gap(&header, header_offset == self.records_start) => + { + let payload = read_declared_payload(&mut self.file, &header, self.offset)?; + if is_format_gap(&header, payload.as_deref()) { + return Err(error); + } + return self.stop(StopReason::Corruption { + offset: header_offset, + }); + } Err(_) => { return self.stop(StopReason::Corruption { offset: header_offset, @@ -525,4 +551,114 @@ mod tests { assert_eq!(records.len(), 1); assert_eq!(records[0].payload, b"keep-me"); } + + /// An unknown format version is a typed error, not a silent stop, and a + /// zeroed version is a stop rather than an error. + /// + /// A store written by another build is intact, so its format reaches the + /// caller as a version and never as damage. Version zero is the exception: + /// no build writes it, so it stays a corruption stop, which `torn_tail` + /// already bounds. + #[test] + fn an_unknown_format_version_is_an_error_and_a_zeroed_one_is_a_stop() { + let dir = tempfile::tempdir().unwrap(); + let framed = dir.path().join("framed.wal"); + { + let mut writer = WalWriter::open_without_direct_io(&framed).unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"x") + .unwrap(); + writer.sync().unwrap(); + } + + let mut bytes = std::fs::read(&framed).unwrap(); + crate::record::foreign_version::reseal_with_version(&mut bytes, 0, 1); + let unknown = dir.path().join("unknown.wal"); + std::fs::write(&unknown, &bytes).unwrap(); + + let reader = WalReader::open_raw(&unknown).unwrap(); + match reader.records().collect::>>() { + Err(WalError::UnsupportedVersion { version, supported }) => { + assert_eq!(version, 1); + assert_eq!(supported, crate::record::WAL_FORMAT_VERSION); + } + other => panic!("expected UnsupportedVersion, got {other:?}"), + } + + // The version field is bytes 4..6 of the record header. + bytes[4..6].copy_from_slice(&0u16.to_le_bytes()); + let zeroed = dir.path().join("zeroed.wal"); + std::fs::write(&zeroed, &bytes).unwrap(); + + let mut reader = WalReader::open_raw(&zeroed).unwrap(); + let mut records = Vec::new(); + while let Some(record) = reader + .next_record() + .expect("a zeroed version stops the stream, it does not fail it") + { + records.push(record); + } + assert!( + records.is_empty(), + "nothing before the stop is a readable record" + ); + // An empty stream is also what a silent skip or a clean end would + // return, so the reason has to be asserted for this to prove a stop. + assert!( + matches!(reader.stop_reason(), Some(StopReason::Corruption { .. })), + "a zeroed version must be a corruption stop, not a quiet end: {:?}", + reader.stop_reason() + ); + } + + /// A version this build cannot read, found after the segment's first + /// record, is damage rather than a format gap. + /// + /// One writer never mixes formats inside one segment, so a mismatch there + /// can only be a broken header. The stream stops and `torn_tail` classifies + /// where, instead of the reader declaring a format gap the store does not + /// have. + #[test] + fn a_version_mismatch_after_the_first_record_is_damage() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("two-records.wal"); + { + let mut writer = WalWriter::open_without_direct_io(&path).unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"first") + .unwrap(); + writer + .append(RecordType::Put as u32, 1, 0, 0, b"second") + .unwrap(); + writer.sync().unwrap(); + } + + let mut reader = WalReader::open_raw(&path).unwrap(); + let first = reader + .next_record() + .unwrap() + .expect("the first record reads"); + assert_eq!(first.header.lsn, 1, "records before the damage still read"); + let second_at = reader.committed_end(); + + let mut bytes = std::fs::read(&path).unwrap(); + let version_at = usize::try_from(second_at).unwrap() + 4; + bytes[version_at..version_at + 2].copy_from_slice(&1u16.to_le_bytes()); + std::fs::write(&path, &bytes).unwrap(); + + let mut reader = WalReader::open_raw(&path).unwrap(); + assert!( + reader.next_record().unwrap().is_some(), + "the first record is still readable" + ); + assert!( + reader.next_record().unwrap().is_none(), + "the mismatch stops the stream instead of failing the read" + ); + assert!( + matches!(reader.stop_reason(), Some(StopReason::Corruption { .. })), + "a mismatch after the first record is damage, not a format gap: {:?}", + reader.stop_reason() + ); + } } diff --git a/nodedb-wal/src/record/foreign_version.rs b/nodedb-wal/src/record/foreign_version.rs new file mode 100644 index 000000000..bb56cd4de --- /dev/null +++ b/nodedb-wal/src/record/foreign_version.rs @@ -0,0 +1,146 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! Judging a record header whose format version this build does not read. +//! +//! A segment written by another build holds intact records in another format. +//! The open must refuse that segment and name both versions. A damaged header +//! in this build's format is not a format gap: the reader stops there, as at +//! any other damaged header. +//! +//! One writer never mixes formats inside one segment, so the format is judged +//! at the segment's first record only. A mismatch later in the segment is +//! damage. No build writes format 0, so a zeroed version is a torn write. +//! +//! A flipped bit in the version field of this build's record leaves every +//! other field and the checksum intact. The checksum covers the version, so +//! recomputing it with this build's version restores a match. Another build's +//! header puts its fields at other offsets, so its checksum does not match. + +use super::header::{MAX_WAL_PAYLOAD_SIZE, RecordHeader, WAL_FORMAT_VERSION}; + +/// Whether `header`, whose version this build does not read, can be another +/// build's format. Only a non-zero version at the segment's first record +/// can. Every other mismatch is damage. +pub fn may_be_format_gap(header: &RecordHeader, at_first_record: bool) -> bool { + at_first_record && header.format_version != 0 +} + +/// Whether a header that [`may_be_format_gap`] accepted is another build's +/// format. +/// +/// `payload` is the bytes the header declares, or `None` when the segment +/// does not hold that many. Without them the header is judged a format gap: +/// nothing proves it is this build's record. +pub fn is_format_gap(header: &RecordHeader, payload: Option<&[u8]>) -> bool { + let Some(payload) = payload else { + return true; + }; + let restored = RecordHeader { + format_version: WAL_FORMAT_VERSION, + ..*header + }; + restored.compute_checksum(payload) != header.crc32c +} + +/// Read the payload `header` declares from `file`, whose cursor sits at +/// `payload_start`. Returns `None` when the declared length is over the +/// payload limit or past the end of the file. +#[cfg(not(target_arch = "wasm32"))] +pub(crate) fn read_declared_payload( + file: &mut std::fs::File, + header: &RecordHeader, + payload_start: u64, +) -> crate::error::Result>> { + use std::io::Read; + + let Ok(len) = usize::try_from(header.payload_len) else { + return Ok(None); + }; + if len > MAX_WAL_PAYLOAD_SIZE { + return Ok(None); + } + let file_len = file.metadata()?.len(); + match payload_start.checked_add(u64::from(header.payload_len)) { + Some(end) if end <= file_len => {} + _ => return Ok(None), + } + let mut payload = vec![0u8; len]; + file.read_exact(&mut payload)?; + Ok(Some(payload)) +} + +/// Rewrite the record at `start` of `bytes` the way another build seals one: +/// declare `version`, then recompute the checksum. +#[cfg(test)] +pub(crate) fn reseal_with_version(bytes: &mut [u8], start: usize, version: u16) { + use super::header::HEADER_SIZE; + + let head: [u8; HEADER_SIZE] = bytes[start..start + HEADER_SIZE] + .try_into() + .expect("a whole header"); + let mut header = RecordHeader::from_bytes(&head); + header.format_version = version; + let payload_start = start + HEADER_SIZE; + let payload_end = payload_start + header.payload_len as usize; + header.crc32c = header.compute_checksum(&bytes[payload_start..payload_end]); + bytes[start..payload_start].copy_from_slice(&header.to_bytes()); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::record::{NO_EVENT_SOURCE, WAL_MAGIC}; + + fn sealed(payload: &[u8]) -> RecordHeader { + let mut header = RecordHeader { + magic: WAL_MAGIC, + format_version: WAL_FORMAT_VERSION, + record_type: 1, + lsn: 7, + tenant_id: 1, + vshard_id: 0, + payload_len: u32::try_from(payload.len()).unwrap(), + database_id: 0, + apply_key: 0, + event_source: NO_EVENT_SOURCE, + commit_hlc: 0, + crc32c: 0, + }; + header.crc32c = header.compute_checksum(payload); + header + } + + #[test] + fn only_a_non_zero_version_at_the_first_record_may_be_a_gap() { + let mut header = sealed(b"x"); + header.format_version = 1; + assert!(may_be_format_gap(&header, true)); + assert!(!may_be_format_gap(&header, false)); + header.format_version = 0; + assert!(!may_be_format_gap(&header, true)); + } + + #[test] + fn a_flipped_version_bit_in_this_builds_record_is_damage() { + let payload = b"payload"; + let mut header = sealed(payload); + header.format_version = WAL_FORMAT_VERSION ^ 0x0100; + assert!(!is_format_gap(&header, Some(payload))); + } + + #[test] + fn a_record_sealed_in_another_version_is_a_gap() { + let payload = b"payload"; + let mut header = sealed(payload); + header.format_version = 1; + header.crc32c = header.compute_checksum(payload); + assert!(is_format_gap(&header, Some(payload))); + } + + #[test] + fn a_header_without_its_payload_is_a_gap() { + let mut header = sealed(b"payload"); + header.format_version = 1; + assert!(is_format_gap(&header, None)); + } +} diff --git a/nodedb-wal/src/record/mod.rs b/nodedb-wal/src/record/mod.rs index e2a182ce8..bb73c6e51 100644 --- a/nodedb-wal/src/record/mod.rs +++ b/nodedb-wal/src/record/mod.rs @@ -4,6 +4,7 @@ pub mod aborted; #[cfg(not(target_arch = "wasm32"))] pub mod anchor; +pub mod foreign_version; #[cfg(not(target_arch = "wasm32"))] pub mod fts_spatial; pub mod header; @@ -26,6 +27,7 @@ pub mod wal_record; pub use aborted::{WRITE_ABORTED_PAYLOAD_SIZE, WriteAbortedPayload}; #[cfg(not(target_arch = "wasm32"))] pub use anchor::{TIME_ANCHOR_PAYLOAD_SIZE, TimeAnchorPayload}; +pub use foreign_version::{is_format_gap, may_be_format_gap}; #[cfg(not(target_arch = "wasm32"))] pub use fts_spatial::{FtsDeletePayload, FtsIndexPayload, SpatialDeletePayload, SpatialPutPayload}; pub use header::{ diff --git a/nodedb-wal/src/recovery.rs b/nodedb-wal/src/recovery.rs index 59ffa96df..b85f572bb 100644 --- a/nodedb-wal/src/recovery.rs +++ b/nodedb-wal/src/recovery.rs @@ -22,7 +22,7 @@ use std::path::Path; -use crate::error::{Result, WalError}; +use crate::error::Result; use crate::reader::WalReader; /// Result of scanning a WAL file for recovery. @@ -76,11 +76,11 @@ pub fn recover(path: &Path) -> Result { // End of committed prefix (EOF or corruption). break; } - Err(e @ WalError::UnknownRequiredRecordType { .. }) => { - // Cannot proceed past unknown required records. - return Err(e); - } - Err(e) => return Err(e), + // Nothing past an unknown required record can be read, and a version + // gap is the one error that needs the segment it was found in: the + // reader sees one header at a time and holds no path. Every other + // error passes through unchanged. + Err(error) => return Err(error.with_segment_path(path)), } } @@ -101,6 +101,7 @@ pub fn recover(path: &Path) -> Result { #[cfg(test)] mod tests { use super::*; + use crate::error::WalError; use crate::record::RecordType; use crate::writer::WalWriter; @@ -188,4 +189,76 @@ mod tests { assert_eq!(info.record_count, 2); assert_eq!(info.next_lsn(), 3); } + + /// Where each of `count` records starts, and the file's bytes. + fn segment_with_records(path: &Path, count: u8) -> (Vec, Vec) { + { + let mut writer = WalWriter::open_without_direct_io(path).unwrap(); + for i in 0..count { + writer + .append(RecordType::Put as u32, 1, 0, 0, &[i; 8]) + .unwrap(); + } + writer.sync().unwrap(); + } + let mut reader = WalReader::open_raw(path).unwrap(); + let mut starts = Vec::new(); + loop { + let start = reader.committed_end(); + if reader.next_record().unwrap().is_none() { + break; + } + starts.push(start); + } + (starts, std::fs::read(path).unwrap()) + } + + #[test] + fn a_lone_first_record_in_another_format_is_a_format_gap() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("lone.wal"); + let (starts, mut bytes) = segment_with_records(&path, 1); + let at = usize::try_from(starts[0]).unwrap(); + crate::record::foreign_version::reseal_with_version(&mut bytes, at, 1); + std::fs::write(&path, &bytes).unwrap(); + + match recover(&path) { + Err(WalError::SegmentFormatVersion { + path: named, + version, + supported, + }) => { + assert_eq!(named, path.display().to_string()); + assert_eq!(version, 1); + assert_eq!(supported, crate::record::WAL_FORMAT_VERSION); + } + other => panic!("expected SegmentFormatVersion, got {other:?}"), + } + } + + /// A flipped bit in the first record's version field is damage to this + /// build's record, not another build's format. The intact records behind + /// it make the damage a hole, so recovery refuses it as mid-file + /// corruption instead of reporting a format gap. + #[test] + fn a_flipped_first_version_bit_is_damage_not_a_format_gap() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("flipped.wal"); + let (starts, mut bytes) = segment_with_records(&path, 3); + // The version field is bytes 4..6 of a record header. + let at = usize::try_from(starts[0]).unwrap() + 4; + let flipped = crate::record::WAL_FORMAT_VERSION ^ 0x0100; + bytes[at..at + 2].copy_from_slice(&flipped.to_le_bytes()); + std::fs::write(&path, &bytes).unwrap(); + + match recover(&path) { + Err(WalError::MidFileCorruption { + offset, resync_lsn, .. + }) => { + assert_eq!(offset, starts[0]); + assert_eq!(resync_lsn, 2); + } + other => panic!("expected MidFileCorruption, got {other:?}"), + } + } } diff --git a/nodedb-wal/src/segmented.rs b/nodedb-wal/src/segmented.rs index f4754c63b..3ffd18def 100644 --- a/nodedb-wal/src/segmented.rs +++ b/nodedb-wal/src/segmented.rs @@ -463,7 +463,10 @@ pub fn replay_from_limit_dir( // the `from_lsn` filter — continuity is a property of the log, not of // the caller's window into it. let mut last_lsn = 0u64; - while let Some(record) = reader.next_record()? { + while let Some(record) = reader + .next_record() + .map_err(|error| error.with_segment_path(&seg.path))? + { last_lsn = record.header.lsn; if record.header.lsn >= from_lsn { records.push(record); diff --git a/nodedb/src/bootstrap/wal_init.rs b/nodedb/src/bootstrap/wal_init.rs index 7c1d79dbf..0a17aa35b 100644 --- a/nodedb/src/bootstrap/wal_init.rs +++ b/nodedb/src/bootstrap/wal_init.rs @@ -22,6 +22,7 @@ pub fn init_wal( )> { let wal_segment_target = config.checkpoint.wal_segment_target_bytes(); let wal_dir = config.wal_dir(); + let wal = { let mut mgr = WalManager::open_with_tuning(&wal_dir, wal_segment_target, &config.tuning.wal) @@ -37,9 +38,12 @@ pub fn init_wal( info!(next_lsn = %wal.next_lsn(), "WAL ready"); if let Err(e) = wal.validate_for_startup() { + // Neutral on purpose: this wraps every validation error. The wrapped + // error names the cause, such as a WAL format version this build + // cannot read. tracing::error!( error = %e, - "StartupError: WAL validation failed — cannot start with corrupted WAL segments" + "StartupError: WAL validation failed: cannot start with these WAL segments" ); std::process::exit(1); } @@ -104,6 +108,11 @@ pub fn init_wal( /// `O_DIRECT` by design, the server will not downgrade itself to buffered /// writes to get past this, and the operator is the only one who can decide /// between the two ways out. +/// +/// A store whose records are in a format this build cannot read gets one too. +/// The open reaches it first, because it recovers the newest segment, and the +/// WAL layer reports the segment and both versions; what it cannot know is which +/// actions exist, and those belong to the server. fn wal_open_error(wal_dir: &std::path::Path, error: crate::Error) -> anyhow::Error { if let crate::Error::Wal(nodedb_wal::WalError::DirectIoUnsupported { .. }) = error { return anyhow::Error::new(nodedb_types::NodeDbError::wal_at( @@ -111,6 +120,16 @@ fn wal_open_error(wal_dir: &std::path::Path, error: crate::Error) -> anyhow::Err direct_io_unsupported_message(wal_dir), )); } + if let crate::Error::Wal(nodedb_wal::WalError::SegmentFormatVersion { + path, + version, + supported, + }) = &error + { + return anyhow::Error::new(crate::Error::VersionCompat { + detail: crate::wal::manager::replay::version_gap_detail(path, *version, *supported), + }); + } anyhow::Error::new(error) } diff --git a/nodedb/src/ctl/restore/cut_rule.rs b/nodedb/src/ctl/restore/cut_rule.rs index 93eb83683..6dd27ba24 100644 --- a/nodedb/src/ctl/restore/cut_rule.rs +++ b/nodedb/src/ctl/restore/cut_rule.rs @@ -83,10 +83,10 @@ pub fn scan_cut( list_through: u64, marks_after: u64, ) -> Result { - let (_, frames) = Frames::open(bytes)?; + let (_, mut frames) = Frames::open(bytes)?; let mut scan = CutScan::default(); let mut last = None; - for frame in frames { + for frame in frames.by_ref() { if frame.is_padding() { continue; } @@ -114,6 +114,7 @@ pub fn scan_cut( } } } + frames.finish(key)?; Ok(scan) } @@ -242,4 +243,31 @@ mod tests { assert_eq!(scan.max_kept, Some(1)); assert!(scan.dropped.is_empty()); } + + /// A segment whose first record is in another WAL format is refused. Read + /// as empty, it would report that no record of the restore point exists. + #[test] + fn a_segment_in_another_format_is_refused_not_read_as_empty() { + let mut bytes = segment(&[(RecordType::Put, 7, 50, b"kept".to_vec())]); + let head: &[u8; nodedb_wal::record::HEADER_SIZE] = + bytes[..nodedb_wal::record::HEADER_SIZE].try_into().unwrap(); + let mut header = nodedb_wal::record::RecordHeader::from_bytes(head); + header.format_version = 1; + let payload_end = nodedb_wal::record::HEADER_SIZE + header.payload_len as usize; + header.crc32c = + header.compute_checksum(&bytes[nodedb_wal::record::HEADER_SIZE..payload_end]); + bytes[..nodedb_wal::record::HEADER_SIZE].copy_from_slice(&header.to_bytes()); + + match scan_cut("seg", &bytes, &rule(), 9, 0) { + Err(RestoreError::Wal(nodedb_wal::WalError::SegmentFormatVersion { + path, + version, + .. + })) => { + assert_eq!(path, "seg"); + assert_eq!(version, 1); + } + other => panic!("expected a format gap, got {other:?}"), + } + } } diff --git a/nodedb/src/ctl/restore/segment.rs b/nodedb/src/ctl/restore/segment.rs index 7cb69e233..537283680 100644 --- a/nodedb/src/ctl/restore/segment.rs +++ b/nodedb/src/ctl/restore/segment.rs @@ -9,13 +9,15 @@ //! alignment padding and carry no LSN. use nodedb_types::temporal::LsnTimeAnchor; +use nodedb_wal::WalError; use nodedb_wal::crypto::KeyRing; use nodedb_wal::preamble::{ PREAMBLE_SIZE, SegmentPreamble, WAL_PREAMBLE_MAGIC, parse_leading_preamble, }; use nodedb_wal::record::{ - HEADER_SIZE, RecordHeader, RecordType, RestorePointPayload, SurrogateAllocPayload, - TimeAnchorPayload, WalRecord, WalRecordArgs, WriteAbortedPayload, padding_record, padding_span, + HEADER_SIZE, MAX_WAL_PAYLOAD_SIZE, RecordHeader, RecordType, RestorePointPayload, + SurrogateAllocPayload, TimeAnchorPayload, WalRecord, WalRecordArgs, WriteAbortedPayload, + is_format_gap, may_be_format_gap, padding_record, padding_span, }; use nodedb_wal::segment::SegmentDecryptor; @@ -78,7 +80,13 @@ impl Frame<'_> { pub(super) struct Frames<'a> { bytes: &'a [u8], offset: usize, + /// Where the segment's first record starts. The WAL format version is + /// judged here and nowhere else, as the segment reader does. + records_start: usize, done: bool, + /// Version and supported version of an unreadable first record. A walk + /// that ends here holds no records, and [`Frames::finish`] reports why. + refused: Option<(u16, u16)>, } impl<'a> Frames<'a> { @@ -90,11 +98,29 @@ impl<'a> Frames<'a> { Self { bytes, offset, + records_start: offset, done: false, + refused: None, }, )) } + /// Report a first record in a WAL format version this build cannot read. + /// + /// Call it after the walk. An archive from another build holds intact + /// records, so reading it as an empty segment would drop all of them + /// without an error. `key` names the segment in the error. + pub(super) fn finish(&self, key: &str) -> Result<(), RestoreError> { + match self.refused { + Some((version, supported)) => Err(RestoreError::Wal(WalError::SegmentFormatVersion { + path: key.to_string(), + version, + supported, + })), + None => Ok(()), + } + } + /// Byte offset where records begin: past the preamble, if any. fn records_start(preamble: Option<&SegmentPreamble>) -> usize { if preamble.is_some() { PREAMBLE_SIZE } else { 0 } @@ -118,17 +144,37 @@ impl<'a> Iterator for Frames<'a> { } impl<'a> Frames<'a> { - fn read_at(&self, start: usize) -> Option> { - let head: &[u8; HEADER_SIZE] = self - .bytes + fn read_at(&mut self, start: usize) -> Option> { + let bytes: &'a [u8] = self.bytes; + let head: &[u8; HEADER_SIZE] = bytes .get(start..start.checked_add(HEADER_SIZE)?)? .try_into() .ok()?; let header = RecordHeader::from_bytes(head); - header.validate(start as u64).ok()?; + match header.validate(start as u64) { + Ok(()) => {} + Err(WalError::UnsupportedVersion { version, supported }) => { + // Another build's format refuses the segment. Damage, in this + // build's format, ends the walk without an error. + if may_be_format_gap(&header, start == self.records_start) { + let payload = usize::try_from(header.payload_len) + .ok() + .filter(|len| *len <= MAX_WAL_PAYLOAD_SIZE) + .and_then(|len| { + let payload_start = start.checked_add(HEADER_SIZE)?; + bytes.get(payload_start..payload_start.checked_add(len)?) + }); + if is_format_gap(&header, payload) { + self.refused = Some((version, supported)); + } + } + return None; + } + Err(_) => return None, + } let payload_start = start + HEADER_SIZE; let end = payload_start.checked_add(usize::try_from(header.payload_len).ok()?)?; - let payload = self.bytes.get(payload_start..end)?; + let payload = bytes.get(payload_start..end)?; (header.compute_checksum(payload) == header.crc32c).then_some(Frame { header, payload, @@ -144,10 +190,10 @@ pub fn scan_segment( bytes: &[u8], ring: Option<&KeyRing>, ) -> Result { - let (preamble, frames) = Frames::open(bytes)?; + let (preamble, mut frames) = Frames::open(bytes)?; let decryptor = SegmentDecryptor::new(preamble.as_ref(), ring); let mut scan = SegmentScan::default(); - for frame in frames { + for frame in frames.by_ref() { if frame.is_padding() { continue; } @@ -188,6 +234,7 @@ pub fn scan_segment( _ => {} } } + frames.finish(key)?; Ok(scan) } @@ -215,12 +262,12 @@ pub fn cut_segment( if !alignment.is_power_of_two() { return Err(RestoreError::BadAlignment { alignment }); } - let (preamble, frames) = Frames::open(bytes)?; + let (preamble, mut frames) = Frames::open(bytes)?; let mut keep_end = Frames::records_start(preamble.as_ref()); let mut last_kept: Option = None; let mut last_seen: Option = None; let mut dropped_records = 0u64; - for frame in frames { + for frame in frames.by_ref() { if frame.is_padding() { continue; } @@ -233,6 +280,7 @@ pub fn cut_segment( dropped_records += 1; } } + frames.finish(key)?; let mut out = bytes .get(..keep_end) @@ -486,4 +534,100 @@ mod tests { assert_eq!(scan.last_lsn, Some(5)); assert_eq!(scan_segment("seg", &[], None).unwrap().last_lsn, None); } + + /// Where each record of `bytes` starts. + fn record_starts(bytes: &[u8]) -> Vec { + let (preamble, frames) = Frames::open(bytes).unwrap(); + let mut starts = vec![Frames::records_start(preamble.as_ref())]; + starts.extend(frames.map(|frame| frame.end)); + starts.pop(); + starts + } + + /// Declare `version` in the header of the record that starts at `start`, + /// leaving its checksum as it was: a damaged header. + fn set_format_version(bytes: &mut [u8], start: usize, version: u16) { + let head: &[u8; HEADER_SIZE] = bytes[start..start + HEADER_SIZE].try_into().unwrap(); + let mut header = RecordHeader::from_bytes(head); + header.format_version = version; + bytes[start..start + HEADER_SIZE].copy_from_slice(&header.to_bytes()); + } + + /// Seal the record that starts at `start` in format `version`, the way + /// another build writes one: declare it and recompute the checksum. + fn reseal_with_version(bytes: &mut [u8], start: usize, version: u16) { + let head: &[u8; HEADER_SIZE] = bytes[start..start + HEADER_SIZE].try_into().unwrap(); + let mut header = RecordHeader::from_bytes(head); + header.format_version = version; + let payload_start = start + HEADER_SIZE; + let payload = &bytes[payload_start..payload_start + header.payload_len as usize]; + header.crc32c = header.compute_checksum(payload); + bytes[start..payload_start].copy_from_slice(&header.to_bytes()); + } + + fn assert_format_gap(result: Result, version: u16) { + match result { + Err(RestoreError::Wal(WalError::SegmentFormatVersion { + path, + version: found, + supported, + })) => { + assert_eq!(path, "seg"); + assert_eq!(found, version); + assert_eq!(supported, nodedb_wal::record::WAL_FORMAT_VERSION); + } + other => panic!("expected a format gap, got {other:?}"), + } + } + + /// An archive from another build holds intact records. Reading it as an + /// empty segment would drop all of them without an error. + #[test] + fn an_archive_in_another_format_is_refused_not_read_as_empty() { + let mut bytes = segment(3, None); + let starts = record_starts(&bytes); + for &start in &starts { + reseal_with_version(&mut bytes, start, 1); + } + + assert_format_gap(scan_segment("seg", &bytes, None), 1); + assert_format_gap(cut_segment("seg", &bytes, 3, &[], None, ALIGN), 1); + } + + /// A flipped bit in the first record's version field is damage to this + /// build's record. The walk ends there without refusing the segment as + /// another build's. + #[test] + fn a_flipped_first_version_bit_stops_the_walk() { + let mut bytes = segment(3, None); + let starts = record_starts(&bytes); + set_format_version( + &mut bytes, + starts[0], + nodedb_wal::record::WAL_FORMAT_VERSION ^ 0x0100, + ); + + let scan = scan_segment("seg", &bytes, None).unwrap(); + assert_eq!(scan.last_lsn, None); + } + + #[test] + fn a_version_mismatch_after_the_first_record_stops_the_walk() { + let mut bytes = segment(3, None); + let starts = record_starts(&bytes); + set_format_version(&mut bytes, starts[1], 1); + + let scan = scan_segment("seg", &bytes, None).unwrap(); + assert_eq!(scan.last_lsn, Some(1)); + } + + #[test] + fn a_zeroed_first_version_stops_the_walk() { + let mut bytes = segment(3, None); + let starts = record_starts(&bytes); + set_format_version(&mut bytes, starts[0], 0); + + let scan = scan_segment("seg", &bytes, None).unwrap(); + assert_eq!(scan.last_lsn, None); + } } diff --git a/nodedb/src/wal/manager/replay.rs b/nodedb/src/wal/manager/replay.rs index dac5cc7c8..e7100c173 100644 --- a/nodedb/src/wal/manager/replay.rs +++ b/nodedb/src/wal/manager/replay.rs @@ -17,6 +17,11 @@ impl WalManager { /// does not parse as WAL records is treated as fatal corruption, not as an /// empty WAL. The WAL replay path is lenient (stops at the first invalid /// record) — this method is the complementary hard check run at startup. + /// + /// A segment whose records are in a WAL format this build cannot read is + /// reported as a version gap rather than as corruption. Its answer is to + /// start the store with the build that wrote the segment, or to re-create + /// the data directory; corruption calls for repair. pub fn validate_for_startup(&self) -> crate::Result<()> { let segments = nodedb_wal::segment::discover_segments(&self.wal_dir).map_err(crate::Error::Wal)?; @@ -47,7 +52,16 @@ impl WalManager { continue; } - let info = nodedb_wal::recovery::recover(&seg.path).map_err(crate::Error::Wal)?; + let info = nodedb_wal::recovery::recover(&seg.path).map_err(|error| match error { + nodedb_wal::error::WalError::SegmentFormatVersion { + path, + version, + supported, + } => crate::Error::VersionCompat { + detail: version_gap_detail(&path, version, supported), + }, + other => crate::Error::Wal(other), + })?; if info.end_offset == 0 { return Err(crate::Error::SegmentCorrupted { @@ -207,11 +221,28 @@ impl WalManager { } } +/// The operator-facing text for a store whose records are in a WAL format this +/// build cannot read. +/// +/// One text, because the same store is refused from two places: the open fails +/// on the newest segment, and startup validation finds the gap in any segment +/// behind it. It names the segment, both versions, and the actions that exist. +/// No WAL format migration tool exists, so naming one would send an operator +/// looking for something they cannot find. +pub(crate) fn version_gap_detail(path: &str, version: u16, supported: u16) -> String { + format!( + "WAL segment '{path}' holds records in WAL format version {version}; this build reads \ + version {supported}. Start the store with the build that wrote that segment, or re-create \ + the data directory and load the data again" + ) +} + #[cfg(test)] mod tests { use super::*; use crate::types::{DatabaseId, TenantId, VShardId}; use crate::wal::manager::NO_APPLY_KEY; + use nodedb_wal::record::{RecordType, WAL_FORMAT_VERSION, WAL_MAGIC, WalRecordArgs}; fn put(wal: &WalManager, payload: &[u8]) -> Lsn { wal.appender(NO_APPLY_KEY) @@ -261,4 +292,268 @@ mod tests { ); assert!(wal.time_anchors().lsn_at_or_before(first_ns - 1).is_err()); } + + /// One record sealed in format `version`: header, then payload, with the + /// checksum computed over the declared version. + /// + /// The framing is this build's, which a version-1 writer never produced, + /// because its record type field was 16 bits wide and this one is 32. The + /// checksum matches the declared version, as it does for a record another + /// build wrote, so the reader judges a format gap and not damage. + fn record_with_format_version(lsn: u64, version: u16) -> Vec { + let mut record = WalRecord::new(WalRecordArgs { + record_type: RecordType::Put as u32, + lsn, + tenant_id: 1, + vshard_id: 0, + database_id: 0, + payload: vec![7u8; 16], + encryption_key: None, + preamble_bytes: None, + }) + .expect("build a record"); + record.header.format_version = version; + record.header.crc32c = record.header.compute_checksum(&record.payload); + let mut bytes = record.header.to_bytes().to_vec(); + bytes.extend_from_slice(&record.payload); + bytes + } + + /// A segment whose first record declares `version`. + fn write_segment_with_format_version( + path: &std::path::Path, + first_lsn: u64, + version: u16, + ) -> std::path::PathBuf { + let bytes = record_with_format_version(first_lsn, version); + let segment = nodedb_wal::segment::segment_path(path, first_lsn); + std::fs::write(&segment, &bytes).expect("write the segment"); + segment + } + + /// An encrypted store keeps its first record behind the preamble, so the + /// version has to be read from behind it too. + #[test] + fn a_store_with_a_preamble_says_which_version() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + { + let wal = WalManager::open_for_testing(&path).expect("open wal"); + put(&wal, b"a"); + wal.sync().expect("sync"); + } + let wal = WalManager::open_for_testing(&path).expect("reopen wal"); + let newest = nodedb_wal::segment::discover_segments(&path) + .expect("discover segments") + .last() + .expect("the boot's own segment") + .first_lsn; + + // A real preamble, not a zeroed one: the reader validates its own + // version field and would reject a malformed preamble for the wrong + // reason. + let mut bytes = nodedb_wal::preamble::SegmentPreamble::new_wal([9, 9, 9, 9]) + .to_bytes() + .to_vec(); + bytes.extend_from_slice(&record_with_format_version(newest + 1_000, 1)); + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 1_000), + &bytes, + ) + .expect("write an encrypted-format segment"); + // A newer segment, so the one above is not the tail the gate tolerates. + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 2_000), + vec![0u8; 64 * 1024], + ) + .expect("write a newer segment"); + + let error = wal + .validate_for_startup() + .expect_err("an older format behind a preamble must be refused"); + let text = format!("{error}"); + assert!( + text.contains("version 1"), + "the version sits behind the preamble and the message must still name it: {text}" + ); + assert!( + text.contains(&format!("reads version {WAL_FORMAT_VERSION}")), + "the message must name the version this build reads: {text}" + ); + } + + /// A torn write that persists the magic and leaves the version zeroed is + /// not a format gap, and must not turn a bootable store into a refused one. + #[test] + fn a_zeroed_version_is_not_a_format_gap() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + { + let wal = WalManager::open_for_testing(&path).expect("open wal"); + put(&wal, b"a"); + wal.sync().expect("sync"); + } + let wal = WalManager::open_for_testing(&path).expect("reopen wal"); + let newest = nodedb_wal::segment::discover_segments(&path) + .expect("discover segments") + .last() + .expect("the boot's own segment") + .first_lsn; + + // The magic of a real record, the version bytes never written. + let mut bytes = vec![0u8; 64 * 1024]; + bytes[0..4].copy_from_slice(&WAL_MAGIC.to_le_bytes()); + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 1_000), + &bytes, + ) + .expect("write a torn newest segment"); + + assert!( + wal.validate_for_startup().is_ok(), + "a zeroed version is uninitialised bytes, not a format this build cannot read" + ); + } + + /// A valid store, then a newer segment written in an older format. + fn store_with_an_older_newest_segment( + path: &std::path::Path, + version: u16, + ) -> (WalManager, std::path::PathBuf) { + { + let wal = WalManager::open_for_testing(path).expect("open wal"); + put(&wal, b"a"); + wal.sync().expect("sync"); + } + let wal = WalManager::open_for_testing(path).expect("reopen wal"); + let newest = nodedb_wal::segment::discover_segments(path) + .expect("discover segments") + .last() + .expect("the boot's own segment") + .first_lsn; + let segment = write_segment_with_format_version(path, newest + 1_000, version); + (wal, segment) + } + + /// The newest segment is the one the open recovers, so a store in another + /// format fails there, before validation runs. The refusal names the + /// segment and both versions: that is what tells a store written by another + /// build apart from a damaged one. + #[test] + fn the_open_refuses_a_newest_segment_in_another_format() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + let (_wal, segment) = store_with_an_older_newest_segment(&path, 1); + + let error = match WalManager::open_for_testing(&path) { + Ok(_) => panic!("a store written in another format must not open"), + Err(error) => error, + }; + let text = format!("{error}"); + + assert!( + text.contains(&segment.display().to_string()), + "the refusal must name the segment: {text}" + ); + assert!( + text.contains("version 1"), + "the refusal must name the version it found: {text}" + ); + assert!( + text.contains(&format!("reads version {WAL_FORMAT_VERSION}")), + "the refusal must name the version this build reads: {text}" + ); + assert!( + !text.contains("corrupted"), + "a format gap is not corruption: {text}" + ); + } + + /// The same gap behind the newest segment is out of the open's reach, so + /// startup validation is what finds it. That message also names the actions + /// that exist, which the WAL layer cannot know. + #[test] + fn validation_names_a_gap_behind_the_newest_segment() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + let (wal, segment) = store_with_an_older_newest_segment(&path, 1); + let newest = nodedb_wal::segment::discover_segments(&path) + .expect("discover segments") + .last() + .expect("a segment") + .first_lsn; + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 1_000), + vec![0u8; 64 * 1024], + ) + .expect("write a newer segment"); + + let error = wal + .validate_for_startup() + .expect_err("a gap behind the tail is still a gap"); + let text = format!("{error}"); + + assert!( + text.contains(&segment.display().to_string()), + "the message must name the segment: {text}" + ); + assert!( + text.contains("version 1"), + "the message must name the version it found, and it named none: {text}" + ); + assert!( + text.contains(&format!("reads version {WAL_FORMAT_VERSION}")), + "the message must name the version this build reads: {text}" + ); + assert!( + !text.contains("corrupted"), + "a format gap is reported as a version gap, never as corruption: {text}" + ); + assert!( + text.contains("re-create the data directory"), + "the message must name an action that exists: {text}" + ); + } + + /// The other side of the same branch: a segment that is not a WAL at all is + /// still corruption, and still says so. + #[test] + fn a_segment_that_is_not_a_wal_still_says_corrupted() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + { + let wal = WalManager::open_for_testing(&path).expect("open wal"); + put(&wal, b"a"); + wal.sync().expect("sync"); + } + let wal = WalManager::open_for_testing(&path).expect("reopen wal"); + let newest = nodedb_wal::segment::discover_segments(&path) + .expect("discover segments") + .last() + .expect("a segment") + .first_lsn; + // No magic at all, which is what a truncated or overwritten file looks + // like, as opposed to a file another version wrote. A newer segment + // follows it so this one is not the tail, because the tail is the one + // case the gate tolerates. + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 1_000), + vec![0u8; 64 * 1024], + ) + .expect("write a segment with no framing"); + std::fs::write( + nodedb_wal::segment::segment_path(&path, newest + 2_000), + vec![0u8; 64 * 1024], + ) + .expect("write a newer zero-filled segment"); + + let error = wal + .validate_for_startup() + .expect_err("a segment with no framing must be refused"); + let text = format!("{error}"); + assert!( + text.contains("corrupted"), + "a segment that is not a WAL is corruption: {text}" + ); + } } From 1dc0333459ed9d73afea67db9042a493b0ad6e45 Mon Sep 17 00:00:00 2001 From: EnRaiha <15997552+EnRaiha@users.noreply.github.com> Date: Thu, 15 Oct 2026 01:36:50 +0800 Subject: [PATCH 3/5] fix(wal): accept a newest segment that holds only crash residue A crash right after a segment rollover leaves the newest segment with zeros or a torn first record. The next boot resumes that file, but startup validation refused it as corrupt, so every later boot failed too. holds_only_crash_residue recognises that state, and validate_for_startup exempts the newest segment when it matches. Any other content, and any older segment without a record, is still corruption. --- nodedb-wal/src/lib.rs | 2 +- nodedb-wal/src/segmented.rs | 35 ++++++++++ nodedb-wal/src/torn_tail.rs | 99 +++++++++++++++++++++++++++ nodedb/src/wal/manager/replay.rs | 113 ++++++++++++++++++++++++++++++- 4 files changed, 246 insertions(+), 3 deletions(-) diff --git a/nodedb-wal/src/lib.rs b/nodedb-wal/src/lib.rs index 40c321eba..c2722bb1f 100644 --- a/nodedb-wal/src/lib.rs +++ b/nodedb-wal/src/lib.rs @@ -108,6 +108,6 @@ pub use time_anchors::TimeAnchors; #[cfg(not(target_arch = "wasm32"))] pub use tombstone::{CollectionTombstonePayload, MAX_COLLECTION_NAME_LEN}; #[cfg(not(target_arch = "wasm32"))] -pub use torn_tail::{TailVerdict, verify_committed_prefix}; +pub use torn_tail::{TailVerdict, holds_only_crash_residue, verify_committed_prefix}; #[cfg(not(target_arch = "wasm32"))] pub use writer::WalWriter; diff --git a/nodedb-wal/src/segmented.rs b/nodedb-wal/src/segmented.rs index 3ffd18def..3449bc3b7 100644 --- a/nodedb-wal/src/segmented.rs +++ b/nodedb-wal/src/segmented.rs @@ -1172,4 +1172,39 @@ mod tests { Ok(_) => panic!("opened a WAL whose last segment reissues LSNs"), } } + + /// A newest segment whose first record has the record magic and a zeroed + /// version is a torn write, not a format this build cannot read. Opening + /// over it must succeed, and the writer must keep working, because the + /// writer's open is the first thing a boot does. + #[test] + fn open_accepts_a_newest_segment_with_a_zeroed_version() { + let dir = tempfile::tempdir().unwrap(); + let next = { + let mut wal = SegmentedWal::open(test_config(dir.path())).unwrap(); + wal.append(RecordType::Put as u32, 1, 0, 0, b"a").unwrap(); + wal.sync().unwrap(); + wal.next_lsn() + }; + + // A rollover that crashed before its first record: the new segment is + // named for the next LSN and holds the magic with nothing behind it. + // Written after the close, so the open has to read this state back. + let mut bytes = vec![0u8; 64 * 1024]; + bytes[0..4].copy_from_slice(&crate::record::WAL_MAGIC.to_le_bytes()); + std::fs::write(segment_path(dir.path(), next), &bytes).unwrap(); + + let mut wal = SegmentedWal::open(test_config(dir.path())) + .expect("a torn newest segment is not a format this build cannot read"); + let lsn = wal.append(RecordType::Put as u32, 1, 0, 0, b"b").unwrap(); + assert_eq!( + lsn, next, + "the writer has to keep working over the torn tail, not just open" + ); + wal.sync().unwrap(); + drop(wal); + let (records, _) = replay_from_limit_dir(dir.path(), 0, usize::MAX, None) + .expect("the resumed log replays without a gap"); + assert_eq!(records.len(), 2, "both records replay"); + } } diff --git a/nodedb-wal/src/torn_tail.rs b/nodedb-wal/src/torn_tail.rs index 39d5e9773..3f332a3d9 100644 --- a/nodedb-wal/src/torn_tail.rs +++ b/nodedb-wal/src/torn_tail.rs @@ -139,6 +139,10 @@ pub fn classify(path: &Path, corruption_offset: u64, last_lsn: u64) -> Result Result> { let header_end = match offset.checked_add(HEADER_SIZE as u64) { Some(end) if end <= file_len => end, @@ -178,6 +182,31 @@ fn intact_record_lsn(file: &mut File, offset: u64, file_len: u64) -> Result Result { + let mut file = File::open(path)?; + let mut window = Vec::with_capacity(SCAN_CHUNK); + let mut read = read_up_to(&mut file, SCAN_CHUNK, &mut window)?; + if window.get(..4) == Some(WAL_MAGIC.to_le_bytes().as_slice()) { + return Ok(true); + } + loop { + if window.iter().any(|&byte| byte != 0) { + return Ok(false); + } + if read < SCAN_CHUNK { + return Ok(true); + } + read = read_up_to(&mut file, SCAN_CHUNK, &mut window)?; + } +} + /// Replace `buf` with up to `len` bytes from the file's current position, /// returning the number of bytes read. Fewer than `len` means EOF. /// @@ -227,6 +256,32 @@ mod tests { file.sync_all().unwrap(); } + #[test] + fn crash_residue_is_zeros_or_a_torn_first_record() { + let dir = tempfile::tempdir().unwrap(); + + let zeros = dir.path().join("zeros.wal"); + std::fs::write(&zeros, vec![0u8; SCAN_CHUNK + 4096]).unwrap(); + assert!(holds_only_crash_residue(&zeros).unwrap()); + + let torn = dir.path().join("torn.wal"); + let mut bytes = vec![0u8; 4096]; + bytes[..4].copy_from_slice(&WAL_MAGIC.to_le_bytes()); + bytes[4..10].copy_from_slice(&[0xA5; 6]); + std::fs::write(&torn, &bytes).unwrap(); + assert!(holds_only_crash_residue(&torn).unwrap()); + + let junk = dir.path().join("junk.wal"); + std::fs::write(&junk, b"NDBS-not-a-wal-record").unwrap(); + assert!(!holds_only_crash_residue(&junk).unwrap()); + + let late_junk = dir.path().join("late_junk.wal"); + let mut bytes = vec![0u8; SCAN_CHUNK + 4096]; + bytes[SCAN_CHUNK + 7] = 1; + std::fs::write(&late_junk, &bytes).unwrap(); + assert!(!holds_only_crash_residue(&late_junk).unwrap()); + } + #[test] fn truncated_tail_is_a_torn_write() { let dir = tempfile::tempdir().unwrap(); @@ -303,6 +358,50 @@ mod tests { ); } + /// A record whose format version this build cannot read is damage, not a + /// resync point: the scan passes over it to the intact record behind it. + #[test] + fn a_version_mismatch_past_the_stop_is_not_a_record() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("version_mismatch.wal"); + write_segment(&path, 4); + + // Where each record starts: the reader's committed end after the one + // before it. + let mut starts = Vec::new(); + let mut reader = crate::reader::WalReader::open_raw(&path).unwrap(); + loop { + let start = reader.committed_end(); + if reader.next_record().unwrap().is_none() { + break; + } + starts.push(start); + } + + // Damage the second record, and give the third a version this build + // cannot read. The checksum covers the version, so it is recomputed: + // the third record is intact in every way except its version. The + // fourth stays intact. + smash(&path, starts[1], HEADER_SIZE); + let mut bytes = std::fs::read(&path).unwrap(); + let third = usize::try_from(starts[2]).unwrap(); + let head: &[u8; HEADER_SIZE] = bytes[third..third + HEADER_SIZE].try_into().unwrap(); + let mut header = RecordHeader::from_bytes(head); + let payload_end = third + HEADER_SIZE + header.payload_len as usize; + header.format_version = 1; + header.crc32c = header.compute_checksum(&bytes[third + HEADER_SIZE..payload_end]); + bytes[third..third + HEADER_SIZE].copy_from_slice(&header.to_bytes()); + std::fs::write(&path, &bytes).unwrap(); + + assert_eq!( + classify(&path, starts[1], 1).unwrap(), + TailVerdict::MidFileCorruption { + resync_offset: starts[3], + resync_lsn: 4, + } + ); + } + #[test] fn clean_eof_needs_no_scan() { let dir = tempfile::tempdir().unwrap(); diff --git a/nodedb/src/wal/manager/replay.rs b/nodedb/src/wal/manager/replay.rs index e7100c173..0b5c1eae5 100644 --- a/nodedb/src/wal/manager/replay.rs +++ b/nodedb/src/wal/manager/replay.rs @@ -11,7 +11,9 @@ impl WalManager { /// /// Returns `Err` if any non-empty segment contains no valid WAL records — /// a reliable signal that the segment was corrupted (wrong magic, truncated - /// header, etc.) rather than simply rolled over empty. + /// header, etc.) rather than simply rolled over empty. The newest segment + /// passes when it holds only what a crash leaves behind: zeros, or a torn + /// first record. /// /// This check is intentionally strict: a segment file with content that /// does not parse as WAL records is treated as fatal corruption, not as an @@ -26,7 +28,7 @@ impl WalManager { let segments = nodedb_wal::segment::discover_segments(&self.wal_dir).map_err(crate::Error::Wal)?; - for seg in &segments { + for (idx, seg) in segments.iter().enumerate() { let file_len = std::fs::metadata(&seg.path).map(|m| m.len()).unwrap_or(0); if file_len == 0 { @@ -64,6 +66,20 @@ impl WalManager { })?; if info.end_offset == 0 { + // This check covers only segments without a preamble. A + // segment that opens with one reports its end at the end of the + // preamble, never at 0, even when no record follows it. + // + // A crash can leave the newest segment holding only zeros or a + // torn first record, and the next boot resumes that file. + // Refusing it would refuse every later boot too. Any other + // content, or any older segment without a record, is + // corruption: replay would skip whatever it held. + if idx + 1 == segments.len() + && nodedb_wal::holds_only_crash_residue(&seg.path).map_err(crate::Error::Wal)? + { + continue; + } return Err(crate::Error::SegmentCorrupted { detail: format!( "WAL segment '{}' is non-empty ({file_len} bytes) but contains no valid \ @@ -331,6 +347,99 @@ mod tests { segment } + /// A valid store, then a newest segment holding no parseable record: the + /// zero-filled pages a crash can leave behind, which the next boot resumes. + fn wal_with_a_recordless_newest_segment(path: &std::path::Path) -> WalManager { + { + let wal = WalManager::open_for_testing(path).expect("open wal"); + put(&wal, b"a"); + wal.sync().expect("sync"); + } + + let wal = WalManager::open_for_testing(path).expect("reopen wal"); + let newest = nodedb_wal::segment::discover_segments(path) + .expect("discover segments") + .last() + .expect("the boot's own segment") + .first_lsn; + // Written after open on purpose: open resumes the newest segment and may + // rewrite its preamble, and the state under test is the one the gate + // then reads back. + std::fs::write( + nodedb_wal::segment::segment_path(path, newest + 1_000), + vec![0u8; 64 * 1024], + ) + .expect("write a zero-filled segment"); + wal + } + + #[test] + fn a_recordless_newest_segment_does_not_block_startup() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + let wal = wal_with_a_recordless_newest_segment(&path); + + assert!( + wal.validate_for_startup().is_ok(), + "the newest segment holds no committed record, which torn_tail already warns \ + about, and the next boot resumes it. Refusing it here makes the store \ + permanently unbootable" + ); + } + + /// A newest segment holding bytes no crash leaves behind is corruption, + /// not a torn tail. + #[test] + fn a_newest_segment_of_junk_blocks_startup() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + let wal = wal_with_a_recordless_newest_segment(&path); + let newest = nodedb_wal::segment::discover_segments(&path) + .expect("discover segments") + .last() + .expect("the zero-filled segment") + .path + .clone(); + std::fs::write(&newest, b"NDBS-not-a-wal-record").expect("write junk"); + + let error = wal + .validate_for_startup() + .expect_err("junk in the newest segment must be refused"); + assert!( + matches!(error, crate::Error::SegmentCorrupted { .. }), + "junk is corruption: {error}" + ); + } + + #[test] + fn a_segment_with_no_records_behind_the_newest_still_blocks_startup() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("wal"); + let wal = wal_with_a_recordless_newest_segment(&path); + let segments = nodedb_wal::segment::discover_segments(&path).expect("discover segments"); + let recordless = segments.last().expect("the zero-filled segment"); + // A newer zero-filled segment becomes the newest, so the first one sits + // behind it: nothing resumes that file, and the gate refuses it. + std::fs::write( + nodedb_wal::segment::segment_path(&path, recordless.first_lsn + 2_000), + vec![0u8; 64 * 1024], + ) + .expect("write a newer zero-filled segment"); + + let error = wal + .validate_for_startup() + .expect_err("a record-less segment behind the newest must be refused"); + assert!( + matches!(error, crate::Error::SegmentCorrupted { .. }), + "a record-less segment behind the newest is corruption: {error}" + ); + let text = format!("{error}"); + assert!( + text.contains(&recordless.path.display().to_string()), + "the refusal must name the record-less segment: {text}" + ); + } + /// An encrypted store keeps its first record behind the preamble, so the /// version has to be read from behind it too. #[test] From 038ecbe4699fdd1123091c7946ace0b7c87064d9 Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Thu, 15 Oct 2026 02:40:44 +0800 Subject: [PATCH 4/5] chore(deps): update rustls to 0.23.46 --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 0d9526127..10c298f56 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6289,9 +6289,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.45" +version = "0.23.46" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d41d731c7d2f962d1ccc364cec258de3c0e93b38c2fb3ba97ac74513048d634" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" dependencies = [ "aws-lc-rs", "log", From faea3682cea9137d3fc9139e21683a2c299eba9f Mon Sep 17 00:00:00 2001 From: Farhan Syah Date: Thu, 15 Oct 2026 02:44:10 +0800 Subject: [PATCH 5/5] ci(fuzz): pin the nightly toolchain and cache fuzz builds A cold ASan build uses most of the job time limit. Pin the nightly so the cache key stays stable across days, and cache the fuzz target directory per matrix target, keyed on the toolchain and fuzz/Cargo.lock. --- .github/workflows/fuzz.yml | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/.github/workflows/fuzz.yml b/.github/workflows/fuzz.yml index 0edab882d..7501202c1 100644 --- a/.github/workflows/fuzz.yml +++ b/.github/workflows/fuzz.yml @@ -51,9 +51,18 @@ jobs: dictionary: fuzz/dictionaries/sql.dict steps: - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + # A pinned nightly keeps the build cache key stable across days. Move + # the pin forward on purpose, not by the calendar. - uses: dtolnay/rust-toolchain@4fd1da8b0805d2d2e936788875a7d65dbd677dc2 with: - toolchain: nightly + toolchain: nightly-2026-10-01 + # A cold ASan build takes most of the job's time limit. One cache per + # target, keyed on the toolchain and fuzz/Cargo.lock. The weekly run on + # main warms it for every pull request. + - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + with: + workspaces: fuzz -> target + key: ${{ matrix.target.name }} # All six matrix jobs install the same pinned binary; build it once. - name: Cache cargo-fuzz id: cache-cargo-fuzz