diff --git a/Cargo.lock b/Cargo.lock index ba11d3bd9..126a17102 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1689,6 +1689,18 @@ dependencies = [ "tracing", ] +[[package]] +name = "flux-disk" +version = "0.5.0" +source = "git+https://github.com/gattaca-com/flux?rev=8064a8dbefa19659562d9810e2b80bab7f33b200#8064a8dbefa19659562d9810e2b80bab7f33b200" +dependencies = [ + "flux-timing", + "flux-utils", + "io-uring", + "libc", + "tracing", +] + [[package]] name = "flux-profiler" version = "0.5.0" @@ -2630,6 +2642,17 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "io-uring" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed3bd0ecfbb87805f538bb7b32e5239ca0763890c623e349860ecba69469f2bb" +dependencies = [ + "bitflags 2.13.0", + "cfg-if", + "libc", +] + [[package]] name = "ipconfig" version = "0.3.4" @@ -4954,7 +4977,9 @@ dependencies = [ "buffa", "buffa-build", "bytes", + "crc32c", "flux", + "flux-disk", "fxhash", "raft", "silver_chain_spec", diff --git a/Cargo.toml b/Cargo.toml index 6774e4ce6..f00614df4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -90,6 +90,7 @@ silver_peer = {path = "crates/peer" } silver_storage = { path = "crates/storage" } silver_engine_api = { path = "crates/engine_api" } flux = { git = "https://github.com/gattaca-com/flux", rev = "8064a8dbefa19659562d9810e2b80bab7f33b200"} +flux-disk = { git = "https://github.com/gattaca-com/flux", rev = "8064a8dbefa19659562d9810e2b80bab7f33b200"} flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "8064a8dbefa19659562d9810e2b80bab7f33b200", features = ["bytes"]} flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "8064a8dbefa19659562d9810e2b80bab7f33b200"} diff --git a/crates/control/Cargo.toml b/crates/control/Cargo.toml index b77490379..258d861da 100644 --- a/crates/control/Cargo.toml +++ b/crates/control/Cargo.toml @@ -17,6 +17,10 @@ silver_peer.workspace = true fxhash.workspace = true tracing.workspace = true +[target.'cfg(target_os = "linux")'.dependencies] +crc32c.workspace = true +flux-disk.workspace = true + [build-dependencies] buffa-build = "0.2.0" diff --git a/crates/control/src/cluster/mod.rs b/crates/control/src/cluster/mod.rs index 81f1b0be9..c9366da02 100644 --- a/crates/control/src/cluster/mod.rs +++ b/crates/control/src/cluster/mod.rs @@ -6,6 +6,8 @@ mod command; mod generated; mod lock_store; mod node; +#[cfg(target_os = "linux")] +mod storage; mod wire; pub use admission::AdmissionError; @@ -17,4 +19,6 @@ pub use node::{ AttestationCluster, AttestationClusterConfig, AttestationDecision, ClusterError, ClusterEvent, ProposalId, ProposeError, }; +#[cfg(target_os = "linux")] +pub use storage::{ClusterStorage, ClusterStorageEvent, RecoveredStorage, StorageIdentity}; pub(crate) use wire::{decode_message, encode_message}; diff --git a/crates/control/src/cluster/storage/journal.rs b/crates/control/src/cluster/storage/journal.rs new file mode 100644 index 000000000..5bfaaaab1 --- /dev/null +++ b/crates/control/src/cluster/storage/journal.rs @@ -0,0 +1,340 @@ +use std::io; + +use buffa::Message as _; +use raft::eraftpb::{Entry, HardState}; + +use super::super::{ + generated, + wire::{from_wire_entry, to_wire_entry}, +}; + +const MAGIC: &[u8; 8] = b"SLVRAFT\x01"; +const FRAME_HEADER_LEN: usize = 16; +const MAX_RECORD_BYTES: usize = 64 * 1024 * 1024; +pub(super) const READ_CHUNK_BYTES: usize = 64 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StorageIdentity { + node_id: u64, + voters: Vec, +} + +impl StorageIdentity { + pub fn new(node_id: u64, mut voters: Vec) -> io::Result { + voters.sort_unstable(); + if node_id == 0 || + !voters.contains(&node_id) || + voters.contains(&0) || + voters.windows(2).any(|pair| pair[0] == pair[1]) || + voters.len() > (MAX_RECORD_BYTES - 20) / 8 + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "invalid Raft storage identity", + )); + } + Ok(Self { node_id, voters }) + } + + pub(super) fn write(&self, output: &mut Vec) { + let start = Frame::start(output); + output.extend_from_slice(MAGIC); + output.extend_from_slice(&self.node_id.to_le_bytes()); + output.extend_from_slice(&(self.voters.len() as u32).to_le_bytes()); + for voter in &self.voters { + output.extend_from_slice(&voter.to_le_bytes()); + } + Frame::finish(output, start); + } + + fn verify(&self, payload: &[u8]) -> io::Result<()> { + let mut cursor = Cursor(payload); + if cursor.take(MAGIC.len())? != MAGIC { + return Err(invalid_data("unsupported Raft journal format")); + } + if cursor.u64()? != self.node_id || cursor.u32()? as usize != self.voters.len() { + return Err(invalid_data("Raft journal identity does not match configuration")); + } + for voter in &self.voters { + if cursor.u64()? != *voter { + return Err(invalid_data("Raft journal voters do not match configuration")); + } + } + cursor.finish() + } +} + +#[derive(Debug, Default)] +pub struct RecoveredStorage { + pub hard_state: HardState, + /// Includes the uncommitted suffix; replay only entries through + /// `hard_state.commit` into the state machine. + pub entries: Vec, +} + +#[derive(Debug, Clone, Default)] +pub(super) struct LogState { + pub hard_state: HardState, + pub last_index: u64, + last_term: u64, +} + +impl LogState { + pub fn next(&self, entries: &[Entry], hard_state: Option<&HardState>) -> io::Result { + let hard_state = hard_state.unwrap_or(&self.hard_state); + if hard_state.term < self.hard_state.term || + hard_state.commit < self.hard_state.commit || + (hard_state.term == self.hard_state.term && + self.hard_state.vote != 0 && + hard_state.vote != self.hard_state.vote) + { + return Err(invalid_data("Raft hard state regressed")); + } + + let mut next = Self { + hard_state: hard_state.clone(), + last_index: self.last_index, + last_term: self.last_term, + }; + if let Some(first) = entries.first() { + if first.index <= self.hard_state.commit || + self.last_index.checked_add(1).is_none_or(|end| first.index > end) + { + return Err(invalid_data( + "Raft append overwrites committed entries or leaves a gap", + )); + } + let mut index = first.index; + let mut term = if first.index > self.last_index { self.last_term } else { 0 }; + for entry in entries { + if entry.index != index || + entry.term == 0 || + entry.term < term || + entry.term > hard_state.term + { + return Err(invalid_data("invalid Raft entry index or term")); + } + index = index.checked_add(1).ok_or_else(|| invalid_data("Raft index overflow"))?; + term = entry.term; + } + next.last_index = index - 1; + next.last_term = term; + } + if next.hard_state.commit > next.last_index { + return Err(invalid_data("Raft commit exceeds the stored log")); + } + Ok(next) + } +} + +pub(super) struct Record<'a> { + entries: &'a [Entry], + hard_state: Option<&'a HardState>, + payload_len: usize, +} + +impl<'a> Record<'a> { + pub fn new(entries: &'a [Entry], hard_state: Option<&'a HardState>) -> io::Result { + let mut payload_len = 1usize + 4 + if hard_state.is_some() { 24 } else { 0 }; + for entry in entries { + // Reject oversized byte fields before protobuf's u32 size calculation. + if entry.data.len() > MAX_RECORD_BYTES || entry.context.len() > MAX_RECORD_BYTES { + return Err(invalid_data("Raft journal record exceeds 64 MiB")); + } + payload_len += 4 + to_wire_entry(entry.clone()).compute_size() as usize; + if payload_len > MAX_RECORD_BYTES { + return Err(invalid_data("Raft journal record exceeds 64 MiB")); + } + } + Ok(Self { entries, hard_state, payload_len }) + } + + pub fn write(&self, output: &mut Vec) { + output.reserve(FRAME_HEADER_LEN + self.payload_len); + let start = Frame::start(output); + output.push(u8::from(self.hard_state.is_some())); + if let Some(state) = self.hard_state { + output.extend_from_slice(&state.term.to_le_bytes()); + output.extend_from_slice(&state.vote.to_le_bytes()); + output.extend_from_slice(&state.commit.to_le_bytes()); + } + output.extend_from_slice(&(self.entries.len() as u32).to_le_bytes()); + for entry in self.entries { + let entry = to_wire_entry(entry.clone()); + output.extend_from_slice(&entry.compute_size().to_le_bytes()); + entry.write_to(output); + } + Frame::finish(output, start); + } + + fn decode(payload: &[u8], entries: &mut Vec) -> io::Result> { + let mut cursor = Cursor(payload); + let state = match cursor.take(1)?[0] { + 0 => None, + 1 => Some(HardState { + term: cursor.u64()?, + vote: cursor.u64()?, + commit: cursor.u64()?, + ..HardState::default() + }), + _ => return Err(invalid_data("invalid Raft hard state tag")), + }; + let count = cursor.u32()? as usize; + if count > cursor.0.len() / 4 { + return Err(invalid_data("invalid Raft journal entry count")); + } + entries.clear(); + for _ in 0..count { + let len = cursor.u32()? as usize; + let entry = generated::Entry::decode_from_slice(cursor.take(len)?) + .map_err(|_| invalid_data("invalid Raft journal protobuf entry"))?; + entries + .push(from_wire_entry(entry).map_err(|_| invalid_data("invalid Raft entry type"))?); + } + cursor.finish()?; + Ok(state) + } +} + +pub(super) struct JournalReplay { + identity: StorageIdentity, + buffer: Vec, + header_read: bool, + record_entries: Vec, + pub recovered: RecoveredStorage, + pub log: LogState, + pub valid_len: u64, +} + +impl JournalReplay { + pub fn new(identity: StorageIdentity) -> Self { + Self { + identity, + buffer: Vec::with_capacity(READ_CHUNK_BYTES), + header_read: false, + record_entries: Vec::new(), + recovered: RecoveredStorage::default(), + log: LogState::default(), + valid_len: 0, + } + } + + pub fn feed(&mut self, bytes: &[u8]) -> io::Result<()> { + self.buffer.extend_from_slice(bytes); + let mut consumed = 0; + while let Some(frame) = Frame::read(&self.buffer[consumed..])? { + if !self.header_read { + self.identity.verify(frame.payload)?; + self.header_read = true; + } else { + let hard_state = Record::decode(frame.payload, &mut self.record_entries)?; + let next = self.log.next(&self.record_entries, hard_state.as_ref())?; + if let Some(first) = self.record_entries.first() { + let retain = usize::try_from(first.index - 1) + .map_err(|_| invalid_data("Raft index exceeds address space"))?; + if retain > 0 && self.recovered.entries[retain - 1].term > first.term { + return Err(invalid_data("Raft entry terms regressed across an append")); + } + self.recovered.entries.truncate(retain); + self.recovered.entries.append(&mut self.record_entries); + } + self.recovered.hard_state = next.hard_state.clone(); + self.log = next; + } + consumed += FRAME_HEADER_LEN + frame.payload.len(); + } + self.valid_len += consumed as u64; + self.buffer.drain(..consumed); + Ok(()) + } + + pub fn finish(&self) -> io::Result { + if !self.header_read { + return Err(invalid_data("Raft journal has no complete identity header")); + } + Ok(!self.buffer.is_empty()) + } +} + +struct Frame<'a> { + payload: &'a [u8], +} + +impl<'a> Frame<'a> { + fn start(output: &mut Vec) -> usize { + let start = output.len(); + output.resize(start + FRAME_HEADER_LEN, 0); + start + } + + fn finish(output: &mut [u8], start: usize) { + let payload = &output[start + FRAME_HEADER_LEN..]; + let len = payload.len() as u64; + let checksum = crc32c::crc32c(payload); + output[start..start + 8].copy_from_slice(&len.to_le_bytes()); + output[start + 8..start + 12].copy_from_slice(&checksum.to_le_bytes()); + let header_checksum = crc32c::crc32c(&output[start..start + 12]); + output[start + 12..start + FRAME_HEADER_LEN] + .copy_from_slice(&header_checksum.to_le_bytes()); + } + + fn read(bytes: &'a [u8]) -> io::Result> { + if bytes.len() < FRAME_HEADER_LEN { + return Ok(None); + } + let mut cursor = Cursor(bytes); + let len = cursor.u64()?; + let checksum = cursor.u32()?; + let header_checksum = cursor.u32()?; + // Protect the length too: corruption must not masquerade as an incomplete tail. + if crc32c::crc32c(&bytes[..12]) != header_checksum || len > MAX_RECORD_BYTES as u64 { + return Err(invalid_data("invalid Raft journal frame header")); + } + if cursor.0.len() < len as usize { + return Ok(None); + } + let payload = cursor.take(len as usize)?; + if crc32c::crc32c(payload) != checksum { + return Err(invalid_data("Raft journal checksum mismatch")); + } + Ok(Some(Self { payload })) + } +} + +struct Cursor<'a>(&'a [u8]); + +impl<'a> Cursor<'a> { + fn take(&mut self, len: usize) -> io::Result<&'a [u8]> { + let (value, remainder) = self + .0 + .split_at_checked(len) + .ok_or_else(|| invalid_data("truncated Raft journal record"))?; + self.0 = remainder; + Ok(value) + } + + fn u32(&mut self) -> io::Result { + let bytes = self.take(4)?.try_into().map_err(|_| invalid_data("invalid u32"))?; + Ok(u32::from_le_bytes(bytes)) + } + + fn u64(&mut self) -> io::Result { + let bytes = self.take(8)?.try_into().map_err(|_| invalid_data("invalid u64"))?; + Ok(u64::from_le_bytes(bytes)) + } + + fn finish(self) -> io::Result<()> { + if self.0.is_empty() { + Ok(()) + } else { + Err(invalid_data("trailing Raft journal record bytes")) + } + } +} + +fn invalid_data(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + +#[cfg(test)] +mod tests; diff --git a/crates/control/src/cluster/storage/journal/tests.rs b/crates/control/src/cluster/storage/journal/tests.rs new file mode 100644 index 000000000..7b934089f --- /dev/null +++ b/crates/control/src/cluster/storage/journal/tests.rs @@ -0,0 +1,173 @@ +use super::*; + +fn identity() -> StorageIdentity { + StorageIdentity::new(1, vec![3, 1, 2]).unwrap() +} + +fn entry(index: u64, term: u64) -> Entry { + Entry { + index, + term, + data: vec![index as u8; 80].into(), + context: vec![term as u8; 16].into(), + ..Entry::default() + } +} + +fn state(term: u64, vote: u64, commit: u64) -> HardState { + HardState { term, vote, commit, ..HardState::default() } +} + +fn header() -> Vec { + let mut bytes = Vec::new(); + identity().write(&mut bytes); + bytes +} + +fn append(bytes: &mut Vec, entries: &[Entry], hard_state: Option<&HardState>) { + Record::new(entries, hard_state).unwrap().write(bytes); +} + +fn replay(bytes: &[u8]) -> io::Result { + let mut replay = JournalReplay::new(identity()); + replay.feed(bytes)?; + replay.finish()?; + Ok(replay) +} + +#[test] +fn identity_is_order_independent_and_rejects_invalid_membership() { + assert_eq!(identity(), StorageIdentity::new(1, vec![1, 2, 3]).unwrap()); + for (node, voters) in + [(0, vec![0]), (1, vec![]), (1, vec![2]), (1, vec![0, 1]), (1, vec![1, 1])] + { + assert_eq!( + StorageIdentity::new(node, voters).unwrap_err().kind(), + io::ErrorKind::InvalidInput + ); + } +} + +#[test] +fn identity_and_membership_must_match_on_recovery() { + for identity in [ + StorageIdentity::new(2, vec![1, 2, 3]).unwrap(), + StorageIdentity::new(1, vec![1, 2]).unwrap(), + ] { + let mut replay = JournalReplay::new(identity); + assert_eq!(replay.feed(&header()).unwrap_err().kind(), io::ErrorKind::InvalidData); + } +} + +#[test] +fn missing_or_incomplete_identity_is_not_a_fresh_store() { + let bytes = header(); + for len in 0..bytes.len() { + assert!(replay(&bytes[..len]).is_err(), "accepted incomplete identity at {len}"); + } +} + +#[test] +fn records_round_trip_through_fragmented_reads() { + let mut bytes = header(); + let mut first = entry(1, 1); + first.sync_log = true; + first.entry_type = raft::eraftpb::EntryType::EntryConfChange; + let entries = vec![first, entry(2, 2)]; + append(&mut bytes, &entries, Some(&state(2, 1, 1))); + append(&mut bytes, &[], Some(&state(2, 1, 2))); + + for chunk_size in [1, 7, 16, 67, bytes.len()] { + let mut replay = JournalReplay::new(identity()); + for chunk in bytes.chunks(chunk_size) { + replay.feed(chunk).unwrap(); + } + assert!(!replay.finish().unwrap()); + assert_eq!(replay.valid_len, bytes.len() as u64); + assert_eq!(replay.recovered.entries, entries); + assert_eq!(replay.recovered.hard_state, state(2, 1, 2)); + } +} + +#[test] +fn replacing_an_uncommitted_suffix_preserves_the_prefix() { + let mut bytes = header(); + append(&mut bytes, &[entry(1, 1), entry(2, 1), entry(3, 1)], Some(&state(1, 1, 1))); + append(&mut bytes, &[entry(2, 2)], Some(&state(2, 2, 2))); + let recovered = replay(&bytes).unwrap().recovered; + assert_eq!(recovered.entries, vec![entry(1, 1), entry(2, 2)]); + assert_eq!(recovered.hard_state, state(2, 2, 2)); +} + +#[test] +fn every_incomplete_tail_preserves_complete_batches_only() { + let mut bytes = header(); + append(&mut bytes, &[entry(1, 1)], Some(&state(1, 1, 1))); + let valid_len = bytes.len(); + append(&mut bytes, &[entry(2, 2)], Some(&state(2, 2, 2))); + for len in valid_len..bytes.len() { + let replay = replay(&bytes[..len]).unwrap(); + assert_eq!(replay.valid_len, valid_len as u64); + assert_eq!(replay.finish().unwrap(), len != valid_len); + assert_eq!(replay.recovered.hard_state, state(1, 1, 1)); + assert_eq!(replay.recovered.entries, vec![entry(1, 1)]); + } +} + +#[test] +fn complete_corrupt_records_fail_closed_including_length_corruption() { + let mut bytes = header(); + let start = bytes.len(); + append(&mut bytes, &[entry(1, 1)], Some(&state(1, 1, 1))); + for index in start..bytes.len() { + bytes[index] ^= 1; + assert!(replay(&bytes).is_err(), "accepted corruption at {index}"); + bytes[index] ^= 1; + } +} + +#[test] +fn impossible_log_and_hard_state_transitions_are_rejected() { + let initial = LogState::default(); + assert!(initial.next(&[entry(0, 1)], Some(&state(1, 1, 0))).is_err()); + assert!(initial.next(&[entry(2, 1)], Some(&state(1, 1, 0))).is_err()); + assert!(initial.next(&[entry(1, 1), entry(3, 1)], Some(&state(1, 1, 0))).is_err()); + assert!(initial.next(&[entry(1, 1)], Some(&state(1, 1, 2))).is_err()); + assert!(initial.next(&[entry(1, 2)], Some(&state(1, 1, 0))).is_err()); + + let current = initial.next(&[entry(1, 1), entry(2, 2)], Some(&state(2, 1, 1))).unwrap(); + assert!(current.next(&[entry(1, 2)], None).is_err()); + assert!(current.next(&[], Some(&state(1, 0, 1))).is_err()); + assert!(current.next(&[], Some(&state(2, 2, 1))).is_err()); + assert!(current.next(&[], Some(&state(2, 0, 1))).is_err()); + assert!(current.next(&[], Some(&state(2, 1, 0))).is_err()); + assert!(current.next(&[entry(3, 1)], None).is_err()); + assert!(current.next(&[], Some(&state(3, 0, 1))).is_ok()); +} + +#[test] +fn checksummed_records_cannot_overwrite_commits_or_introduce_gaps() { + let mut initial = header(); + append(&mut initial, &[entry(1, 2), entry(2, 2)], Some(&state(2, 1, 1))); + for entries in [vec![entry(1, 3)], vec![entry(4, 3)], vec![entry(2, 1)]] { + let mut bytes = initial.clone(); + append(&mut bytes, &entries, Some(&state(3, 1, 1))); + assert!(replay(&bytes).is_err()); + } +} + +#[test] +fn unknown_format_and_malformed_batches_fail_closed() { + let mut bytes = header(); + bytes[FRAME_HEADER_LEN + MAGIC.len() - 1] = 2; + Frame::finish(&mut bytes, 0); + assert!(replay(&bytes).is_err()); + + for payload in [vec![2], vec![0, 0xff, 0xff, 0xff, 0xff], vec![0, 0, 0, 0, 0, 1]] { + let mut bytes = header(); + let start = Frame::start(&mut bytes); + bytes.extend_from_slice(&payload); + Frame::finish(&mut bytes, start); + assert!(replay(&bytes).is_err()); + } +} diff --git a/crates/control/src/cluster/storage/mod.rs b/crates/control/src/cluster/storage/mod.rs new file mode 100644 index 000000000..0f2d3a65b --- /dev/null +++ b/crates/control/src/cluster/storage/mod.rs @@ -0,0 +1,251 @@ +mod journal; + +use std::{io, path::Path}; + +use flux_disk::{DiskConfig, DiskEvent, DiskIo, FileToken, OpenOptions, OperationId}; +use journal::{JournalReplay, LogState, READ_CHUNK_BYTES, Record}; +pub use journal::{RecoveredStorage, StorageIdentity}; +use raft::eraftpb::{Entry, HardState}; + +#[derive(Debug)] +pub enum ClusterStorageEvent { + Recovered(RecoveredStorage), + Persisted { ready_number: u64 }, +} + +/// Keeps one persistence batch in flight. Completion always means durable, not +/// merely written. The caller must exclusively own the journal path for this +/// instance's lifetime. +pub struct ClusterStorage { + disk: DiskIo, + state: StorageState, +} + +impl ClusterStorage { + pub fn create(path: &Path, identity: StorageIdentity) -> io::Result { + Self::start(path, identity, true) + } + + /// Missing journals are errors; recovery must never silently bootstrap a + /// fresh voter. + pub fn open(path: &Path, identity: StorageIdentity) -> io::Result { + Self::start(path, identity, false) + } + + fn start(path: &Path, identity: StorageIdentity, create: bool) -> io::Result { + let parent = path + .parent() + .filter(|path| !path.as_os_str().is_empty()) + .unwrap_or_else(|| Path::new(".")); + let mut disk = DiskIo::new(DiskConfig::default())?; + let directory = disk.open_directory(parent)?; + let file = disk.open(path, OpenOptions::new().read(true).write(true).create_new(create))?; + let (phase, replay) = if create { + (Phase::Creating(identity), None) + } else { + (Phase::ReadNext { offset: 0 }, Some(JournalReplay::new(identity))) + }; + let mut state = StorageState { + file, + directory, + phase, + replay, + recovered: None, + log: LogState::default(), + last_ready_number: 0, + }; + state.advance(&mut disk)?; + Ok(Self { disk, state }) + } + + pub fn is_ready(&self) -> bool { + matches!(self.state.phase, Phase::Ready) + } + + /// Entries are a contiguous append or an uncommitted suffix replacement, as + /// supplied by Raft's `Ready`. Even commit-only batches are synced in + /// this initial implementation. + pub fn persist( + &mut self, + ready_number: u64, + entries: &[Entry], + hard_state: Option<&HardState>, + ) -> io::Result<()> { + match self.state.phase { + Phase::Ready => {} + Phase::Failed => return Err(failed()), + _ => return Err(io::Error::new(io::ErrorKind::WouldBlock, "Raft storage is busy")), + } + if ready_number <= self.state.last_ready_number { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Raft Ready numbers must increase", + )); + } + let next = self.state.log.next(entries, hard_state)?; + let record = Record::new(entries, hard_state)?; + if !self.disk.write_with(self.state.file, |output| record.write(output)) { + self.state.phase = Phase::Failed; + return Err(failed()); + } + let Some(sync) = self.disk.sync_data(self.state.file) else { + self.state.phase = Phase::Failed; + return Err(failed()); + }; + self.state.phase = Phase::SyncingBatch { sync, ready_number, next }; + Ok(()) + } + + pub fn poll(&mut self) -> io::Result> { + let mut result = Ok(None); + self.disk.poll_with(|event| { + if result.is_err() { + return; + } + match self.state.on_event(event) { + Ok(Some(event)) => result = Ok(Some(event)), + Ok(None) => {} + Err(error) => result = Err(error), + } + }); + if result.is_ok() { + if let Err(error) = self.state.advance(&mut self.disk) { + result = Err(error); + } + } + if result.is_err() { + self.state.phase = Phase::Failed; + } + result + } +} + +enum Phase { + Creating(StorageIdentity), + ReadNext { offset: u64 }, + Reading { offset: u64 }, + Repair { truncate: bool, valid_len: u64 }, + SyncingLog { sync: OperationId }, + SyncDirectory, + SyncingDirectory { sync: OperationId }, + Ready, + SyncingBatch { sync: OperationId, ready_number: u64, next: LogState }, + Failed, +} + +struct StorageState { + file: FileToken, + directory: FileToken, + phase: Phase, + replay: Option, + recovered: Option, + log: LogState, + last_ready_number: u64, +} + +impl StorageState { + fn advance(&mut self, disk: &mut DiskIo) -> io::Result<()> { + match &self.phase { + Phase::Creating(identity) => { + if !disk.write_with(self.file, |output| identity.write(output)) { + return Err(failed()); + } + self.recovered = Some(RecoveredStorage::default()); + let sync = disk.sync_all(self.file).ok_or_else(failed)?; + self.phase = Phase::SyncingLog { sync }; + } + Phase::ReadNext { offset } => { + if !disk.read_at(self.file, *offset, READ_CHUNK_BYTES) { + return Err(failed()); + } + self.phase = Phase::Reading { offset: *offset }; + } + Phase::Repair { truncate, valid_len } => { + if *truncate { + disk.truncate(self.file, *valid_len).ok_or_else(failed)?; + } + if !disk.set_write_cursor(self.file, *valid_len) { + return Err(failed()); + } + let sync = disk.sync_all(self.file).ok_or_else(failed)?; + self.phase = Phase::SyncingLog { sync }; + } + Phase::SyncDirectory => { + // File tokens have independent queues; sync the directory only after the file + // sync completes. + let sync = disk.sync_all(self.directory).ok_or_else(failed)?; + self.phase = Phase::SyncingDirectory { sync }; + } + Phase::Failed => return Err(failed()), + _ => {} + } + Ok(()) + } + + fn on_event(&mut self, event: DiskEvent<'_>) -> io::Result> { + if matches!(self.phase, Phase::Failed) { + return Ok(None); + } + match event { + DiskEvent::Failed { op, error, .. } => { + self.phase = Phase::Failed; + return Err(io::Error::new(error.kind(), format!("Raft journal {op:?}: {error}"))); + } + DiskEvent::Read { file, offset, payload, eof } => { + if file != self.file || + !matches!(self.phase, Phase::Reading { offset: expected } if expected == offset) + { + return Err(io::Error::other("unexpected Raft journal read completion")); + } + let replay = self.replay.as_mut().ok_or_else(failed)?; + replay.feed(payload)?; + if eof { + let truncate = replay.finish()?; + let replay = self.replay.take().ok_or_else(failed)?; + self.log = replay.log; + self.recovered = Some(replay.recovered); + self.phase = Phase::Repair { truncate, valid_len: replay.valid_len }; + } else { + let offset = offset + .checked_add(payload.len() as u64) + .ok_or_else(|| io::Error::other("Raft journal offset overflow"))?; + self.phase = Phase::ReadNext { offset }; + } + } + DiskEvent::Synced { file, operation_id } => match &self.phase { + Phase::SyncingLog { sync } if file == self.file && *sync == operation_id => { + self.phase = Phase::SyncDirectory; + } + Phase::SyncingDirectory { sync } + if file == self.directory && *sync == operation_id => + { + let recovered = self.recovered.take().ok_or_else(failed)?; + self.phase = Phase::Ready; + return Ok(Some(ClusterStorageEvent::Recovered(recovered))); + } + Phase::SyncingBatch { sync, ready_number, next } + if file == self.file && *sync == operation_id => + { + let ready_number = *ready_number; + self.log = next.clone(); + self.last_ready_number = ready_number; + self.phase = Phase::Ready; + return Ok(Some(ClusterStorageEvent::Persisted { ready_number })); + } + _ => return Err(io::Error::other("unexpected Raft journal sync completion")), + }, + _ => {} + } + Ok(None) + } +} + +fn failed() -> io::Error { + io::Error::new( + io::ErrorKind::BrokenPipe, + "Raft storage is failed; restart and recover before proceeding", + ) +} + +#[cfg(test)] +mod tests; diff --git a/crates/control/src/cluster/storage/tests.rs b/crates/control/src/cluster/storage/tests.rs new file mode 100644 index 000000000..f93b700ef --- /dev/null +++ b/crates/control/src/cluster/storage/tests.rs @@ -0,0 +1,287 @@ +use std::{ + fs, + time::{Duration, Instant}, +}; + +use flux_disk::FailedOp; +use raft::{Storage as _, storage::MemStorage}; + +use super::*; +use crate::cluster::{ + AttestationKey, AttestationLockCommand, AttestationLockStore, LockResult, + command::ReplicatedCommand, +}; + +fn identity() -> StorageIdentity { + StorageIdentity::new(1, vec![1, 2, 3]).unwrap() +} + +fn entry(index: u64, term: u64) -> Entry { + Entry { index, term, data: vec![index as u8; 32].into(), ..Entry::default() } +} + +fn state(term: u64, commit: u64) -> HardState { + HardState { term, vote: 1, commit, ..HardState::default() } +} + +fn event(storage: &mut ClusterStorage) -> io::Result { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(event) = storage.poll()? { + return Ok(event); + } + assert!(Instant::now() < deadline, "Raft storage completion timed out"); + std::thread::yield_now(); + } +} + +fn recovered(storage: &mut ClusterStorage) -> RecoveredStorage { + match event(storage).unwrap() { + ClusterStorageEvent::Recovered(recovered) => recovered, + event => panic!("unexpected event: {event:?}"), + } +} + +fn persisted(storage: &mut ClusterStorage, number: u64) { + assert!( + matches!(event(storage).unwrap(), ClusterStorageEvent::Persisted { ready_number } if ready_number == number) + ); + assert!(storage.is_ready()); +} + +#[test] +fn durable_batches_recover_entries_hard_state_and_suffix_replacement() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + assert!(!storage.is_ready()); + assert!(recovered(&mut storage).entries.is_empty()); + + storage.persist(1, &[entry(1, 1), entry(2, 1), entry(3, 1)], Some(&state(1, 1))).unwrap(); + assert!(!storage.is_ready()); + assert_eq!(storage.persist(2, &[], None).unwrap_err().kind(), io::ErrorKind::WouldBlock); + persisted(&mut storage, 1); + assert_eq!(storage.persist(1, &[], None).unwrap_err().kind(), io::ErrorKind::InvalidInput); + storage.persist(2, &[entry(2, 2)], Some(&state(2, 2))).unwrap(); + persisted(&mut storage, 2); + drop(storage); + + let mut storage = ClusterStorage::open(&path, identity()).unwrap(); + let restored = recovered(&mut storage); + assert_eq!(restored.entries, vec![entry(1, 1), entry(2, 2)]); + assert_eq!(restored.hard_state, state(2, 2)); + let memory = MemStorage::new_with_conf_state((vec![1, 2, 3], vec![])); + memory.wl().append(&restored.entries).unwrap(); + memory.wl().set_hardstate(restored.hard_state); + assert_eq!(memory.last_index().unwrap(), 2); + assert_eq!(memory.initial_state().unwrap().hard_state.commit, 2); + + storage.persist(1, &[entry(3, 2)], Some(&state(2, 2))).unwrap(); + persisted(&mut storage, 1); + storage.persist(2, &[], Some(&state(2, 3))).unwrap(); + persisted(&mut storage, 2); + drop(storage); + let restored = recovered(&mut ClusterStorage::open(&path, identity()).unwrap()); + assert_eq!(restored.entries, vec![entry(1, 1), entry(2, 2), entry(3, 2)]); + assert_eq!(restored.hard_state.commit, 3); +} + +#[test] +fn recovery_truncates_incomplete_tail_before_appending() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + recovered(&mut storage); + storage.persist(1, &[entry(1, 1)], Some(&state(1, 1))).unwrap(); + persisted(&mut storage, 1); + drop(storage); + let intact = fs::read(&path).unwrap(); + let mut partial = intact.clone(); + Record::new(&[entry(2, 2)], Some(&state(2, 2))).unwrap().write(&mut partial); + partial.truncate(partial.len() - 3); + fs::write(&path, partial).unwrap(); + + let mut storage = ClusterStorage::open(&path, identity()).unwrap(); + assert_eq!(recovered(&mut storage).entries, vec![entry(1, 1)]); + assert_eq!(fs::read(&path).unwrap(), intact); + storage.persist(1, &[entry(2, 3)], Some(&state(3, 2))).unwrap(); + persisted(&mut storage, 1); + drop(storage); + assert_eq!(recovered(&mut ClusterStorage::open(&path, identity()).unwrap()).entries, vec![ + entry(1, 1), + entry(2, 3) + ]); +} + +#[test] +fn replay_reads_large_records_across_multiple_chunks() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + recovered(&mut storage); + let mut large = entry(1, 1); + large.data = vec![0x71; READ_CHUNK_BYTES * 3].into(); + storage.persist(1, &[large.clone()], Some(&state(1, 1))).unwrap(); + persisted(&mut storage, 1); + drop(storage); + assert_eq!(recovered(&mut ClusterStorage::open(&path, identity()).unwrap()).entries, vec![ + large + ]); +} + +#[test] +fn failed_or_mismatched_recovery_never_initializes_an_empty_voter() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut missing = ClusterStorage::open(&path, identity()).unwrap(); + assert_eq!(event(&mut missing).unwrap_err().kind(), io::ErrorKind::NotFound); + assert_eq!(missing.persist(1, &[], None).unwrap_err().kind(), io::ErrorKind::BrokenPipe); + drop(missing); + + recovered(&mut ClusterStorage::create(&path, identity()).unwrap()); + let original = fs::read(&path).unwrap(); + let mut exists = ClusterStorage::create(&path, identity()).unwrap(); + assert_eq!(event(&mut exists).unwrap_err().kind(), io::ErrorKind::AlreadyExists); + drop(exists); + assert_eq!(fs::read(&path).unwrap(), original); + + let wrong_node = StorageIdentity::new(2, vec![1, 2, 3]).unwrap(); + let mut mismatch = ClusterStorage::open(&path, wrong_node).unwrap(); + assert_eq!(event(&mut mismatch).unwrap_err().kind(), io::ErrorKind::InvalidData); + drop(mismatch); + assert_eq!(fs::read(&path).unwrap(), original); + + let mut corrupt = original.clone(); + corrupt[0] ^= 1; + fs::write(&path, &corrupt).unwrap(); + let mut storage = ClusterStorage::open(&path, identity()).unwrap(); + assert_eq!(event(&mut storage).unwrap_err().kind(), io::ErrorKind::InvalidData); + assert_eq!(fs::read(&path).unwrap(), corrupt); +} + +#[test] +fn writes_do_not_acknowledge_and_failed_writes_cannot_be_hidden_by_a_later_sync() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + recovered(&mut storage); + storage.persist(7, &[entry(1, 1)], Some(&state(1, 1))).unwrap(); + let Phase::SyncingBatch { sync, .. } = storage.state.phase else { panic!("not syncing") }; + let file = storage.state.file; + assert!( + storage.state.on_event(DiskEvent::Written { file, offset: 0, len: 32 }).unwrap().is_none() + ); + assert!(!storage.is_ready()); + assert!( + storage + .state + .on_event(DiskEvent::Failed { + file, + op: FailedOp::Write { offset: 0, len: 32 }, + operation_id: None, + error: io::Error::other("injected write failure"), + }) + .is_err() + ); + assert!( + storage.state.on_event(DiskEvent::Synced { file, operation_id: sync }).unwrap().is_none() + ); + assert_eq!(storage.persist(8, &[], None).unwrap_err().kind(), io::ErrorKind::BrokenPipe); +} + +#[test] +fn new_journal_is_not_ready_before_its_directory_entry_is_durable() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + let Phase::SyncingLog { sync } = storage.state.phase else { panic!("not syncing") }; + let file = storage.state.file; + assert!( + storage.state.on_event(DiskEvent::Synced { file, operation_id: sync }).unwrap().is_none() + ); + assert!(!storage.is_ready()); + storage.state.advance(&mut storage.disk).unwrap(); + let Phase::SyncingDirectory { sync } = storage.state.phase else { + panic!("not syncing directory") + }; + assert!(matches!( + storage + .state + .on_event(DiskEvent::Synced { file: storage.state.directory, operation_id: sync }) + .unwrap(), + Some(ClusterStorageEvent::Recovered(_)) + )); + assert!(storage.is_ready()); +} + +#[test] +fn sync_failure_keeps_storage_failed() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + recovered(&mut storage); + storage.persist(1, &[entry(1, 1)], Some(&state(1, 1))).unwrap(); + let Phase::SyncingBatch { sync, .. } = storage.state.phase else { panic!("not syncing") }; + assert!( + storage + .state + .on_event(DiskEvent::Failed { + file: storage.state.file, + op: FailedOp::Sync, + operation_id: Some(sync), + error: io::Error::other("injected sync failure"), + }) + .is_err() + ); + assert!(!storage.is_ready()); + assert_eq!(storage.persist(2, &[], None).unwrap_err().kind(), io::ErrorKind::BrokenPipe); +} + +#[test] +fn committed_epoch_locks_can_be_rebuilt_without_applying_the_uncommitted_suffix() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let command = AttestationLockCommand { + key: AttestationKey { attester_index: 7, slot: 100 }, + subnet: 0, + ssz: [0; silver_common::ssz_view::SINGLE_ATT_SIZE], + }; + let mut later = command; + later.key.attester_index = 8; + let entries = [ + Entry { + index: 1, + term: 1, + data: ReplicatedCommand::Lock(command).encode().into(), + ..Entry::default() + }, + Entry { + index: 2, + term: 1, + data: ReplicatedCommand::Lock(later).encode().into(), + ..Entry::default() + }, + ]; + let mut storage = ClusterStorage::create(&path, identity()).unwrap(); + recovered(&mut storage); + storage.persist(1, &entries, Some(&state(1, 1))).unwrap(); + persisted(&mut storage, 1); + drop(storage); + let restored = recovered(&mut ClusterStorage::open(&path, identity()).unwrap()); + let mut locks = AttestationLockStore::default(); + for entry in + restored.entries.iter().take_while(|entry| entry.index <= restored.hard_state.commit) + { + let ReplicatedCommand::Lock(command) = ReplicatedCommand::decode(&entry.data).unwrap() + else { + panic!("not a lock") + }; + assert_eq!(locks.apply(&command), LockResult::Accepted); + } + let mut conflict = command; + conflict.key.slot = 105; + conflict.ssz[0] = 1; + assert_eq!(locks.apply(&command), LockResult::AlreadyAcceptedSame); + assert_eq!(locks.apply(&conflict), LockResult::ConflictingAttestation); + assert_eq!(locks.apply(&later), LockResult::Accepted); +} diff --git a/crates/control/src/cluster/wire.rs b/crates/control/src/cluster/wire.rs index 2a3a724c0..db7cde527 100644 --- a/crates/control/src/cluster/wire.rs +++ b/crates/control/src/cluster/wire.rs @@ -92,7 +92,7 @@ fn to_wire_message(mut message: Message) -> wire::Message { } } -fn to_wire_entry(entry: Entry) -> wire::Entry { +pub(super) fn to_wire_entry(entry: Entry) -> wire::Entry { wire::Entry { entry_type: (entry.entry_type as i32).into(), term: entry.term, @@ -165,7 +165,7 @@ fn from_wire_message(message: wire::Message) -> Result { Ok(decoded) } -fn from_wire_entry(entry: wire::Entry) -> Result { +pub(super) fn from_wire_entry(entry: wire::Entry) -> Result { let mut decoded = Entry::default(); decoded.set_entry_type(entry_type(entry.entry_type.to_i32())?); decoded.set_term(entry.term);