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 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", 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-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/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/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..3449bc3b7 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); @@ -1169,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/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/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/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/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/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?; diff --git a/nodedb/src/wal/manager/replay.rs b/nodedb/src/wal/manager/replay.rs index dac5cc7c8..0b5c1eae5 100644 --- a/nodedb/src/wal/manager/replay.rs +++ b/nodedb/src/wal/manager/replay.rs @@ -11,17 +11,24 @@ 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 /// 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)?; - 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 { @@ -47,9 +54,32 @@ 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 { + // 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 \ @@ -207,11 +237,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 +308,361 @@ 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 + } + + /// 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] + 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}" + ); + } }