diff --git a/Cargo.lock b/Cargo.lock index 126a1710..bc74fccc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4801,6 +4801,7 @@ dependencies = [ "silver_network", "silver_peer", "silver_storage", + "tempfile", "tracing", ] @@ -4982,6 +4983,7 @@ dependencies = [ "flux-disk", "fxhash", "raft", + "rand 0.8.6", "silver_chain_spec", "silver_columns", "silver_common", diff --git a/crates/bin/Cargo.toml b/crates/bin/Cargo.toml index 0d62c979..8cea185f 100644 --- a/crates/bin/Cargo.toml +++ b/crates/bin/Cargo.toml @@ -32,6 +32,9 @@ tracing.workspace = true [build-dependencies] build-info-build.workspace = true +[dev-dependencies] +tempfile = "3" + [lints] workspace = true diff --git a/crates/bin/src/cluster.rs b/crates/bin/src/cluster.rs new file mode 100644 index 00000000..bcd7efcc --- /dev/null +++ b/crates/bin/src/cluster.rs @@ -0,0 +1,73 @@ +use std::{ + error::Error, + fs::{File, OpenOptions}, + io, + path::Path, +}; + +use silver_common::Enr; +use silver_config::ClusterConfig; +use silver_control::cluster::{AttestationClusterConfig, ClusterStorageConfig}; + +pub struct ClusterStartup { + pub config: AttestationClusterConfig, + _journal_lock: File, +} + +impl ClusterStartup { + pub fn new( + config: &ClusterConfig, + local: &Enr, + directory: &Path, + ) -> Result> { + let node_id = config + .nodes + .iter() + .find(|(_, enr)| enr.node_id() == local.node_id()) + .map(|(id, _)| *id) + .ok_or("no local node configured in cluster config")?; + let journal_lock = Self::lock(directory)?; + let path = directory.join("attestation-cluster.wal"); + let storage = if config.bootstrap { + ClusterStorageConfig::Create(path) + } else { + ClusterStorageConfig::Open(path) + }; + Ok(Self { + config: AttestationClusterConfig::new( + node_id, + config.nodes.keys().copied().collect(), + storage, + ), + _journal_lock: journal_lock, + }) + } + + fn lock(directory: &Path) -> io::Result { + // Retain the lock file: removing it could let processes lock different inodes. + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(directory.join("attestation-cluster.lock"))?; + file.try_lock().map_err(|error| { + io::Error::other(format!("cannot lock attestation cluster journal: {error}")) + })?; + Ok(file) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn journal_lock_is_exclusive_until_the_process_guard_is_dropped() { + let directory = tempfile::tempdir().unwrap(); + let first = ClusterStartup::lock(directory.path()).unwrap(); + assert!(ClusterStartup::lock(directory.path()).is_err()); + drop(first); + let _second = ClusterStartup::lock(directory.path()).unwrap(); + } +} diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index 5d1d07b1..64c85d5c 100644 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -2,6 +2,7 @@ use std::{ error::Error, io, net::IpAddr, + path::Path, str::FromStr, sync::Arc, time::{Duration, Instant, SystemTime, UNIX_EPOCH}, @@ -21,14 +22,14 @@ use silver_columns::tile::DataColumnsTile; #[cfg(feature = "alloc-profile")] use silver_common::metrics::CountingAllocator; use silver_common::{ - APP_NAME, Enr, GossipTopic, ProtoIdentify, SilverSpine, TCache, TCacheId, TCacheProducer, - TCacheReader, TCacheTable, + APP_NAME, Enr, GossipTopic, MAX_CLUSTER_MESSAGE_BYTES, ProtoIdentify, SilverSpine, TCache, + TCacheId, TCacheProducer, TCacheReader, TCacheTable, cell_store::{CellStoreConfig, GOSSIP_DELIVERY_RETENTION}, profiler::enable_profiler, tracing::initialise_tracing_log, }; use silver_config::Config; -use silver_control::{Controller, cluster::AttestationClusterConfig, sync_engine::SyncEngine}; +use silver_control::{Controller, sync_engine::SyncEngine}; use silver_discovery::{DiscV5, Discovery}; use silver_gossip::GossipHandler; use silver_httpcore::Bind; @@ -36,6 +37,10 @@ use silver_network::{ClusterNodes, Context, NetworkTile, P2p}; use silver_peer::PeerManager; use silver_storage::{latest_local_checkpoint, tile::StorageTile}; +use crate::cluster::ClusterStartup; + +mod cluster; + #[cfg(not(feature = "alloc-profile"))] #[global_allocator] static GLOBAL: MiMalloc = MiMalloc; @@ -99,10 +104,14 @@ fn main() -> Result<(), Box> { TCacheId::BoundaryProcessing, config.engine_config().incoming_engine_resp_tcache_size, ); - let cluster_inbound_producer = - TCache::producer(TCacheId::ClusterInbound, CLUSTER_MESSAGE_TCACHE_SIZE); + let cluster_cache_bytes = if config.cluster_config().is_some() { + 4 * MAX_CLUSTER_MESSAGE_BYTES + } else { + CLUSTER_MESSAGE_TCACHE_SIZE + }; + let cluster_inbound_producer = TCache::producer(TCacheId::ClusterInbound, cluster_cache_bytes); let cluster_outbound_producer = - TCache::producer(TCacheId::ClusterOutbound, CLUSTER_MESSAGE_TCACHE_SIZE); + TCache::producer(TCacheId::ClusterOutbound, cluster_cache_bytes); let control_rpc_producer = TCache::producer(TCacheId::ControlRpc, CONTROL_RPC_TCACHE_SIZE); let storage_delivery_producer = TCache::producer(TCacheId::StorageDelivery, config.outgoing_rpc_tcache_size()); @@ -146,19 +155,13 @@ fn main() -> Result<(), Box> { local_enr.set_syncnets(long_lived_syncnets, keypair.secret_key())?; // Cluster configuration - let (cluster_config, cluster_nodes) = config + let cluster_startup = config .cluster_config() - .map(|c| { - let voters = c.nodes.clone(); - let node_id = match voters.iter().find(|(_, enr)| enr.node_id() == local_enr.node_id()) - { - Some((id, _)) => *id, - None => return Err("no local node configured in cluster config"), - }; - Ok((AttestationClusterConfig::new(node_id, voters.keys().copied().collect()), voters)) + .map(|cluster| { + ClusterStartup::new(cluster, &local_enr, Path::new(config.data_storage_dir())) }) - .transpose()? - .unzip(); + .transpose()?; + let cluster_nodes = config.cluster_config().map(|cluster| cluster.nodes.clone()); let discv5_addr = config.discovery_bind_addr().expect("no discovery port"); let p2p_addr = config.p2p_bind_addr().expect("no p2p port"); @@ -279,7 +282,7 @@ fn main() -> Result<(), Box> { control_rpc_producer, tcaches, cluster_outbound_producer, - cluster_config, + cluster_startup.as_ref().map(|cluster| cluster.config.clone()), SyncEngine::new( config.syncing_config(), booting_from_local_checkpoint, diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index ed63e286..4156e419 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -83,3 +83,5 @@ pub use flux::timing::{IngestionTime, Nanos}; pub use generated::{Identify as ProtoIdentify, IdentifyView as ProtoIdentifyView}; pub const APP_NAME: &str = "silver"; + +pub const MAX_CLUSTER_MESSAGE_BYTES: usize = 16 * 1024 * 1024; diff --git a/crates/common/src/spine/stream_protocol.rs b/crates/common/src/spine/stream_protocol.rs index 0030d36b..916cfc18 100644 --- a/crates/common/src/spine/stream_protocol.rs +++ b/crates/common/src/spine/stream_protocol.rs @@ -171,7 +171,7 @@ impl StreamProtocol { StreamProtocol::ExecutionPayloadEnvelopesByRoot => { b"\x48/eth2/beacon_chain/req/execution_payload_envelopes_by_root/1/ssz_snappy\n" } - StreamProtocol::Cluster => b"\x0f/cluster/1.0.0\n", + StreamProtocol::Cluster => b"\x0f/cluster/2.0.0\n", } } diff --git a/crates/common/src/spine/tcache.rs b/crates/common/src/spine/tcache.rs index 7a061c6f..4133affb 100644 --- a/crates/common/src/spine/tcache.rs +++ b/crates/common/src/spine/tcache.rs @@ -190,6 +190,11 @@ impl TCache { self.len as usize } + #[inline] + pub fn max_payload_len(&self) -> usize { + self.capacity() - size_of::() + } + /// Attach to a named shmem segment as a producer, creating it if needed. /// Either side (producer or consumer) may start first. `n` must be /// identical on both sides. diff --git a/crates/common/src/spine/tcache/producer.rs b/crates/common/src/spine/tcache/producer.rs index a2c33e22..eb05d455 100644 --- a/crates/common/src/spine/tcache/producer.rs +++ b/crates/common/src/spine/tcache/producer.rs @@ -209,7 +209,7 @@ impl AllocationState { /// not depend on when `read()` is called. #[inline] fn reserve(&mut self, cache: TCacheRef, len: usize, auto_commit: bool) -> Option { - if len > cache.capacity() - size_of::() { + if len > cache.max_payload_len() { return None; } let reservation_len = cache.reserve_len(self.seq, len); diff --git a/crates/config/src/cluster_config.rs b/crates/config/src/cluster_config.rs index fe6706d8..a0c7cc97 100644 --- a/crates/config/src/cluster_config.rs +++ b/crates/config/src/cluster_config.rs @@ -8,4 +8,24 @@ pub struct ClusterConfig { /// All cluster nodes, including the local node, keyed by cluster node id. #[serde(default)] pub nodes: HashMap, + /// Create a new journal exclusively. Disable on restart; recovery never + /// recreates missing journals. Never use this to replace a lost voter + /// journal. The journal lives at + /// `data_storage_dir/attestation-cluster.wal`; its parent directory must + /// exist. + #[serde(default)] + pub bootstrap: bool, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn journal_creation_requires_explicit_bootstrap() { + let config: ClusterConfig = serde_yml::from_str("nodes: {}").unwrap(); + assert!(!config.bootstrap); + let config: ClusterConfig = serde_yml::from_str("nodes: {}\nbootstrap: true").unwrap(); + assert!(config.bootstrap); + } } diff --git a/crates/control/Cargo.toml b/crates/control/Cargo.toml index 258d861d..bf8bd985 100644 --- a/crates/control/Cargo.toml +++ b/crates/control/Cargo.toml @@ -10,6 +10,7 @@ buffa.workspace = true bytes.workspace = true flux.workspace = true raft.workspace = true +rand.workspace = true silver_chain_spec.workspace = true silver_common.workspace = true silver_gossip.workspace = true diff --git a/crates/control/src/cluster/lock_store.rs b/crates/control/src/cluster/lock_store.rs index 2d367ffc..095ea84b 100644 --- a/crates/control/src/cluster/lock_store.rs +++ b/crates/control/src/cluster/lock_store.rs @@ -1,13 +1,15 @@ -use std::collections::hash_map::Entry; +use std::{collections::hash_map::Entry, io}; use fxhash::FxHashMap; -use silver_common::{SLOTS_PER_EPOCH, merkle}; +use silver_common::{MAX_CLUSTER_MESSAGE_BYTES, SLOTS_PER_EPOCH, merkle}; use super::command::AttestationLockCommand; /// Admission accepts `[wall - SLOTS_PER_EPOCH, wall]`, which spans at most two /// epochs. const LOCK_RING_SIZE: usize = 2; +const SNAPSHOT_MAGIC: &[u8; 8] = b"SLVLOCK\x01"; +const MAX_SNAPSHOT_BYTES: usize = MAX_CLUSTER_MESSAGE_BYTES - 1024; /// Result of applying a committed attestation selection command. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -28,7 +30,7 @@ pub enum LockResult { } /// Anti-equivocation state shared by standalone and replicated admission. -#[derive(Default)] +#[derive(Debug, Default)] pub(crate) struct AttestationLockStore { /// Commands below this slot are rejected even if proposed by a node with a /// stale wall clock. Replicated stores advance this through committed @@ -37,7 +39,7 @@ pub(crate) struct AttestationLockStore { locks: [EpochLocks; LOCK_RING_SIZE], } -#[derive(Default)] +#[derive(Debug, Default)] struct EpochLocks { /// The absolute epoch occupying this modulo bucket. The tag prevents a /// late older command from clearing locks for a newer colliding epoch. @@ -83,6 +85,84 @@ impl AttestationLockStore { self.minimum_slot } + pub(super) fn encode_snapshot(&self) -> io::Result> { + let retained = |bucket: &EpochLocks| { + bucket.epoch.is_some_and(|epoch| epoch >= self.minimum_slot / SLOTS_PER_EPOCH) + }; + let count: usize = self + .locks + .iter() + .filter(|bucket| retained(bucket)) + .map(|bucket| bucket.attestations.len()) + .sum(); + let length = count + .checked_mul(40) + .and_then(|n| n.checked_add(16 + LOCK_RING_SIZE * 13)) + .filter(|n| *n <= MAX_SNAPSHOT_BYTES) + .ok_or_else(|| invalid_snapshot("attestation lock snapshot exceeds transport limit"))?; + let mut bytes = Vec::with_capacity(length); + bytes.extend_from_slice(SNAPSHOT_MAGIC); + bytes.extend_from_slice(&self.minimum_slot.to_le_bytes()); + for bucket in &self.locks { + let Some(epoch) = bucket.epoch else { + bytes.push(0); + continue; + }; + // Keep the epoch tag even when its locks expired: it prevents ring reuse by + // older commands. + bytes.push(1); + bytes.extend_from_slice(&epoch.to_le_bytes()); + let count = if retained(bucket) { bucket.attestations.len() } else { 0 }; + bytes.extend_from_slice(&(count as u32).to_le_bytes()); + if count != 0 { + for (validator, hash) in &bucket.attestations { + bytes.extend_from_slice(&validator.to_le_bytes()); + bytes.extend_from_slice(hash); + } + } + } + Ok(bytes) + } + + pub(super) fn decode_snapshot(bytes: &[u8]) -> io::Result { + if bytes.len() > MAX_SNAPSHOT_BYTES { + return Err(invalid_snapshot("attestation lock snapshot exceeds transport limit")); + } + let mut cursor = SnapshotCursor(bytes); + if &cursor.take::<8>()? != SNAPSHOT_MAGIC { + return Err(invalid_snapshot("unsupported attestation lock snapshot")); + } + let mut store = + Self { minimum_slot: u64::from_le_bytes(cursor.take()?), ..Self::default() }; + for (index, bucket) in store.locks.iter_mut().enumerate() { + match cursor.take::<1>()?[0] { + 0 => continue, + 1 => {} + _ => return Err(invalid_snapshot("invalid snapshot epoch tag")), + } + let epoch = u64::from_le_bytes(cursor.take()?); + if epoch as usize % LOCK_RING_SIZE != index { + return Err(invalid_snapshot("snapshot epoch is in the wrong ring bucket")); + } + bucket.epoch = Some(epoch); + let count = u32::from_le_bytes(cursor.take()?) as usize; + if count > cursor.0.len() / 40 { + return Err(invalid_snapshot("invalid snapshot lock count")); + } + bucket.attestations.reserve(count); + for _ in 0..count { + let validator = u64::from_le_bytes(cursor.take()?); + if bucket.attestations.insert(validator, cursor.take()?).is_some() { + return Err(invalid_snapshot("duplicate validator in snapshot epoch")); + } + } + } + if !cursor.0.is_empty() { + return Err(invalid_snapshot("trailing attestation snapshot bytes")); + } + Ok(store) + } + #[cfg(test)] pub(super) fn len(&self) -> usize { self.locks @@ -105,6 +185,23 @@ impl AttestationLockStore { } } +struct SnapshotCursor<'a>(&'a [u8]); + +impl SnapshotCursor<'_> { + fn take(&mut self) -> io::Result<[u8; N]> { + let (bytes, rest) = self + .0 + .split_at_checked(N) + .ok_or_else(|| invalid_snapshot("truncated attestation snapshot"))?; + self.0 = rest; + bytes.try_into().map_err(|_| invalid_snapshot("invalid attestation snapshot field")) + } +} + +fn invalid_snapshot(message: &'static str) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, message) +} + #[cfg(test)] mod tests { use super::*; @@ -124,6 +221,55 @@ mod tests { } } + #[test] + fn snapshot_preserves_epoch_locks_floor_and_ring_reuse_protection() { + let mut store = AttestationLockStore::default(); + store.apply(&command_for(10, 7, 1)); + store.apply(&command_for(40, 8, 2)); + store.advance_minimum_slot(20); + let mut restored = + AttestationLockStore::decode_snapshot(&store.encode_snapshot().unwrap()).unwrap(); + assert_eq!(restored.minimum_slot(), 20); + assert_eq!(restored.apply(&command_for(10, 7, 1)), LockResult::TooOld); + assert_eq!(restored.apply(&command_for(21, 7, 3)), LockResult::ConflictingAttestation); + assert_eq!(restored.apply(&command_for(40, 8, 2)), LockResult::AlreadyAcceptedSame); + + store.apply(&command_for(100, 9, 4)); + let mut restored = + AttestationLockStore::decode_snapshot(&store.encode_snapshot().unwrap()).unwrap(); + assert_eq!(restored.apply(&command_for(40, 8, 2)), LockResult::TooOld); + assert_eq!(restored.apply(&command_for(100, 9, 4)), LockResult::AlreadyAcceptedSame); + store.advance_minimum_slot(128); + let bytes = store.encode_snapshot().unwrap(); + assert!(bytes.len() < 64, "expired hashes should not be serialized"); + let restored = AttestationLockStore::decode_snapshot(&bytes).unwrap(); + assert_eq!(restored.minimum_slot(), 128); + } + + #[test] + fn malformed_lock_snapshots_are_rejected() { + let mut store = AttestationLockStore::default(); + store.apply(&command_for(10, 7, 1)); + let bytes = store.encode_snapshot().unwrap(); + for len in 0..bytes.len() { + assert!(AttestationLockStore::decode_snapshot(&bytes[..len]).is_err()); + } + let mut invalid = bytes.clone(); + invalid.push(0); + assert!(AttestationLockStore::decode_snapshot(&invalid).is_err()); + let mut invalid = bytes.clone(); + invalid[16] = 2; + assert!(AttestationLockStore::decode_snapshot(&invalid).is_err()); + let mut invalid = bytes.clone(); + invalid[17] = 1; + assert!(AttestationLockStore::decode_snapshot(&invalid).is_err()); + let mut duplicate = bytes; + duplicate[25..29].copy_from_slice(&2u32.to_le_bytes()); + let record = duplicate[29..69].to_vec(); + duplicate.splice(69..69, record); + assert!(AttestationLockStore::decode_snapshot(&duplicate).is_err()); + } + #[test] fn selection_results_distinguish_same_and_conflicting_attestations() { let mut store = AttestationLockStore::default(); diff --git a/crates/control/src/cluster/mod.rs b/crates/control/src/cluster/mod.rs index c9366da0..2ac15112 100644 --- a/crates/control/src/cluster/mod.rs +++ b/crates/control/src/cluster/mod.rs @@ -6,6 +6,9 @@ mod command; mod generated; mod lock_store; mod node; +mod persistence; +mod raft_storage; +mod snapshot_transfers; #[cfg(target_os = "linux")] mod storage; mod wire; @@ -19,6 +22,7 @@ pub use node::{ AttestationCluster, AttestationClusterConfig, AttestationDecision, ClusterError, ClusterEvent, ProposalId, ProposeError, }; +pub use persistence::{ClusterStorageConfig, RecoveredStorage}; #[cfg(target_os = "linux")] -pub use storage::{ClusterStorage, ClusterStorageEvent, RecoveredStorage, StorageIdentity}; +pub use storage::{ClusterStorage, ClusterStorageEvent, StorageIdentity}; pub(crate) use wire::{decode_message, encode_message}; diff --git a/crates/control/src/cluster/node.rs b/crates/control/src/cluster/node.rs index f368947c..4dfccf40 100644 --- a/crates/control/src/cluster/node.rs +++ b/crates/control/src/cluster/node.rs @@ -1,20 +1,24 @@ use std::{ collections::VecDeque, error::Error, - fmt, + fmt, io, time::{Duration, Instant}, }; use raft::{ - Config, RawNode, StateRole, + Config, RawNode, SnapshotStatus, StateRole, eraftpb::{Entry, EntryType, Message}, - storage::MemStorage, }; +#[cfg(any(target_os = "linux", test))] +use super::persistence::RecoveredStorage; use super::{ admission::{AdmissionError, AttestationAdmission}, command::{AttestationLockCommand, CommandDecodeError, ReplicatedCommand}, lock_store::{AttestationLockStore, LockResult}, + persistence::{ClusterStorageConfig, PersistedReady, Persistence, PersistenceEvent}, + raft_storage::{RaftStorage, RestoredSnapshot}, + snapshot_transfers::SnapshotTransfers, }; const DEFAULT_TICK_INTERVAL: Duration = Duration::from_millis(100); @@ -22,30 +26,34 @@ const DEFAULT_HEARTBEAT_TICKS: usize = 2; const DEFAULT_ELECTION_TICKS: usize = 20; const DEFAULT_PROPOSAL_TIMEOUT: Duration = Duration::from_millis(100); const MAX_ELAPSED_TICKS_PER_SPIN: usize = 32; -const PROPOSAL_CONTEXT_LEN: usize = 16; +const PROPOSAL_CONTEXT_LEN: usize = 32; /// Static Raft membership and timing configuration. #[derive(Debug, Clone)] pub struct AttestationClusterConfig { pub node_id: u64, pub voters: Vec, + pub storage: ClusterStorageConfig, pub tick_interval: Duration, pub heartbeat_ticks: usize, pub election_ticks: usize, /// Maximum time between accepting a local proposal and observing its /// commit on this node. pub proposal_timeout: Duration, + pub snapshot_interval: u64, } impl AttestationClusterConfig { - pub fn new(node_id: u64, voters: Vec) -> Self { + pub fn new(node_id: u64, voters: Vec, storage: ClusterStorageConfig) -> Self { Self { node_id, voters, + storage, tick_interval: DEFAULT_TICK_INTERVAL, heartbeat_ticks: DEFAULT_HEARTBEAT_TICKS, election_ticks: DEFAULT_ELECTION_TICKS, proposal_timeout: DEFAULT_PROPOSAL_TIMEOUT, + snapshot_interval: 4096, } } @@ -59,6 +67,9 @@ impl AttestationClusterConfig { if self.proposal_timeout.is_zero() { return Err(ClusterError::InvalidConfig("Raft proposal timeout must be non-zero")); } + if self.snapshot_interval == 0 { + return Err(ClusterError::InvalidConfig("Raft snapshot interval must be non-zero")); + } if self.voters.is_empty() { return Err(ClusterError::InvalidConfig("Raft voter set must not be empty")); } @@ -97,6 +108,7 @@ impl AttestationClusterConfig { pub struct ProposalId { pub origin_node_id: u64, pub sequence: u64, + pub incarnation: [u8; 16], } impl ProposalId { @@ -104,17 +116,24 @@ impl ProposalId { let mut encoded = Vec::with_capacity(PROPOSAL_CONTEXT_LEN); encoded.extend_from_slice(&self.origin_node_id.to_le_bytes()); encoded.extend_from_slice(&self.sequence.to_le_bytes()); + encoded.extend_from_slice(&self.incarnation); encoded } fn decode(encoded: &[u8]) -> Result { - if encoded.len() != PROPOSAL_CONTEXT_LEN { + if encoded.len() != PROPOSAL_CONTEXT_LEN && encoded.len() != 16 { return Err(encoded.len()); } Ok(Self { - origin_node_id: u64::from_le_bytes(encoded[..8].try_into().expect("slice is 8 bytes")), - sequence: u64::from_le_bytes(encoded[8..].try_into().expect("slice is 8 bytes")), + origin_node_id: u64::from_le_bytes(encoded[..8].try_into().map_err(|_| encoded.len())?), + sequence: u64::from_le_bytes(encoded[8..16].try_into().map_err(|_| encoded.len())?), + // Old contexts can still apply, but cannot match this process's requests. + incarnation: if encoded.len() == 16 { + [0; 16] + } else { + encoded[16..].try_into().map_err(|_| encoded.len())? + }, }) } } @@ -156,6 +175,8 @@ pub enum ClusterEvent { #[derive(Debug)] pub enum ProposeError { Admission(AdmissionError), + NotReady, + Failed, SequenceExhausted, DeadlineOverflow, Raft(raft::Error), @@ -165,6 +186,8 @@ impl fmt::Display for ProposeError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { Self::Admission(error) => error.fmt(f), + Self::NotReady => f.write_str("Raft storage recovery is not complete"), + Self::Failed => f.write_str("Raft processing has stopped after a failure"), Self::SequenceExhausted => f.write_str("Raft proposal sequence exhausted"), Self::DeadlineOverflow => f.write_str("Raft proposal deadline overflowed"), Self::Raft(error) => write!(f, "Raft rejected proposal: {error}"), @@ -177,7 +200,9 @@ impl Error for ProposeError { match self { Self::Admission(error) => Some(error), Self::Raft(error) => Some(error), - Self::SequenceExhausted | Self::DeadlineOverflow => None, + Self::NotReady | Self::Failed | Self::SequenceExhausted | Self::DeadlineOverflow => { + None + } } } } @@ -189,7 +214,10 @@ pub enum ClusterError { Command(CommandDecodeError), InvalidProposalContextLength(usize), UnsupportedEntry(EntryType), - SnapshotsUnsupported, + Snapshot(io::Error), + Storage(io::Error), + NotReady, + Failed, } impl fmt::Display for ClusterError { @@ -199,17 +227,18 @@ impl fmt::Display for ClusterError { write!(f, "invalid attestation cluster config: {message}") } Self::Raft(error) => error.fmt(f), + Self::Storage(error) => write!(f, "Raft storage failed: {error}"), + Self::NotReady => f.write_str("Raft storage recovery is not complete"), + Self::Failed => f.write_str("Raft processing has stopped after a failure"), Self::Command(error) => error.fmt(f), Self::InvalidProposalContextLength(actual) => write!( f, - "Raft proposal context has length {actual}, expected {PROPOSAL_CONTEXT_LEN}" + "Raft proposal context has length {actual}, expected 16 or {PROPOSAL_CONTEXT_LEN}" ), Self::UnsupportedEntry(entry_type) => { write!(f, "unsupported committed Raft entry type {entry_type:?}") } - Self::SnapshotsUnsupported => { - f.write_str("attestation Raft snapshots are not supported") - } + Self::Snapshot(error) => write!(f, "invalid attestation Raft snapshot: {error}"), } } } @@ -218,10 +247,13 @@ impl Error for ClusterError { fn source(&self) -> Option<&(dyn Error + 'static)> { match self { Self::Raft(error) => Some(error), + Self::Storage(error) => Some(error), + Self::Snapshot(error) => Some(error), Self::Command(error) => Some(error), Self::InvalidConfig(_) | + Self::NotReady | + Self::Failed | Self::UnsupportedEntry(_) | - Self::SnapshotsUnsupported | Self::InvalidProposalContextLength(_) => None, } } @@ -237,19 +269,19 @@ impl From for ClusterError { /// /// `spin` is a pump: callers invoke it once per Control tile loop. It executes /// only work that is ready at that point and always returns without waiting. -/// The initial implementation uses `MemStorage`; replacing it with durable -/// storage is required before enabling the cluster in production. pub struct AttestationCluster { - node: RawNode, + node: Option>, + config: AttestationClusterConfig, + persistence: Persistence, + failed: bool, state: AttestationLockStore, admission: AttestationAdmission, - node_id: u64, + incarnation: [u8; 16], next_proposal_sequence: u64, - tick_interval: Duration, next_tick: Instant, - proposal_timeout: Duration, pending_proposals: VecDeque, pending_minimum_slot: Option, + snapshot_transfers: SnapshotTransfers, } #[derive(Debug, Clone, Copy)] @@ -260,31 +292,64 @@ struct PendingProposal { impl AttestationCluster { /// Construct the Raft node with local attestation admission disabled. - /// Call [`Self::set_startup_wall_slot`] once this node first catches up to - /// its wall slot; Raft message processing and elections can run before it. + /// Recovery must complete before Raft participation. Local requests also + /// require a synced startup floor. pub fn new(config: AttestationClusterConfig, now: Instant) -> Result { config.validate()?; + let persistence = Persistence::new(&config).map_err(ClusterError::Storage)?; + Ok(Self::with_persistence(config, persistence, now)) + } - let storage = MemStorage::new_with_conf_state((config.voters.clone(), Vec::::new())); - let node = RawNode::with_default_logger(&config.raft_config(), storage)?; + fn with_persistence( + config: AttestationClusterConfig, + persistence: Persistence, + now: Instant, + ) -> Self { let next_tick = now.checked_add(config.tick_interval).unwrap_or(now); - - Ok(Self { - node, + let mut incarnation: [u8; 16] = rand::random(); + incarnation[0] |= 1; // Zero is reserved for legacy proposal contexts. + Self { + node: None, + config, + persistence, + failed: false, state: AttestationLockStore::default(), admission: AttestationAdmission::new(), - node_id: config.node_id, + incarnation, next_proposal_sequence: 1, - tick_interval: config.tick_interval, next_tick, - proposal_timeout: config.proposal_timeout, pending_proposals: VecDeque::new(), pending_minimum_slot: None, - }) + snapshot_transfers: SnapshotTransfers::default(), + } + } + + #[cfg(test)] + pub(crate) fn in_memory( + config: AttestationClusterConfig, + now: Instant, + ) -> Result { + config.validate()?; + let mut cluster = Self::with_persistence(config, Persistence::memory(), now); + cluster.recover(RecoveredStorage::default(), now)?; + Ok(cluster) + } + + #[cfg(test)] + pub(crate) fn fail_persistence(&mut self) { + self.persistence.fail = true; + } + + pub fn is_ready(&self) -> bool { + self.node.is_some() && !self.failed + } + + pub fn is_failed(&self) -> bool { + self.failed } pub fn node_id(&self) -> u64 { - self.node_id + self.config.node_id } /// Enable local attestation admission using the first wall slot at which @@ -294,12 +359,15 @@ impl AttestationCluster { } pub fn leader_id(&self) -> Option { - let leader_id = self.node.raft.leader_id; + if self.failed { + return None; + } + let leader_id = self.node.as_ref()?.raft.leader_id; (leader_id != 0).then_some(leader_id) } pub fn is_leader(&self) -> bool { - self.node.raft.state == StateRole::Leader + !self.failed && self.node.as_ref().is_some_and(|node| node.raft.state == StateRole::Leader) } #[cfg(test)] @@ -315,17 +383,28 @@ impl AttestationCluster { /// Explicitly start an election, primarily for controlled startup and /// tests. pub fn campaign(&mut self) -> Result<(), ClusterError> { - self.node.campaign().map_err(ClusterError::Raft) + if self.failed { + return Err(ClusterError::Failed); + } + self.node.as_mut().ok_or(ClusterError::NotReady)?.campaign().map_err(ClusterError::Raft) } /// Deliver one decoded message received from another member of this Raft /// group. pub fn step(&mut self, message: Message) -> Result<(), ClusterError> { - self.node.step(message).map_err(ClusterError::Raft) + if self.failed { + return Err(ClusterError::Failed); + } + self.node.as_mut().ok_or(ClusterError::NotReady)?.step(message).map_err(ClusterError::Raft) } pub fn report_unreachable(&mut self, node_id: u64) { - self.node.report_unreachable(node_id); + if !self.failed && + let Some(node) = self.node.as_mut() + { + node.report_unreachable(node_id); + node.report_snapshot(node_id, SnapshotStatus::Failure); + } } /// Submit a locally-originated signed attestation for ordering before @@ -341,6 +420,10 @@ impl AttestationCluster { wall_slot: u64, now: Instant, ) -> Result { + if self.failed { + return Err(ProposeError::Failed); + } + let node = self.node.as_mut().ok_or(ProposeError::NotReady)?; self.admission.validate(command.key.slot, wall_slot).map_err(ProposeError::Admission)?; if command.key.slot < self.state.minimum_slot() { @@ -353,17 +436,20 @@ impl AttestationCluster { let sequence = self.next_proposal_sequence; self.next_proposal_sequence = sequence.checked_add(1).ok_or(ProposeError::SequenceExhausted)?; - let proposal_id = ProposalId { origin_node_id: self.node_id, sequence }; + let proposal_id = ProposalId { + origin_node_id: self.config.node_id, + sequence, + incarnation: self.incarnation, + }; let deadline = - now.checked_add(self.proposal_timeout).ok_or(ProposeError::DeadlineOverflow)?; + now.checked_add(self.config.proposal_timeout).ok_or(ProposeError::DeadlineOverflow)?; debug_assert!( self.pending_proposals.back().is_none_or(|pending| pending.deadline <= deadline), "proposal timestamps must be monotonic" ); - self.node - .propose(proposal_id.encode(), ReplicatedCommand::Lock(command).encode()) + node.propose(proposal_id.encode(), ReplicatedCommand::Lock(command).encode()) .map_err(ProposeError::Raft)?; self.pending_proposals.push_back(PendingProposal { id: proposal_id, deadline }); @@ -381,29 +467,79 @@ impl AttestationCluster { wall_slot: u64, mut emit: impl FnMut(ClusterEvent), ) -> Result<(), ClusterError> { + if self.failed { + return Ok(()); + } self.expire_proposals(now, &mut emit); + let result = self.drive(now, wall_slot, &mut emit); + if result.is_err() { + self.failed = true; + self.pending_proposals.clear(); + self.pending_minimum_slot = None; + } + result + } + + fn drive( + &mut self, + now: Instant, + wall_slot: u64, + emit: &mut impl FnMut(ClusterEvent), + ) -> Result<(), ClusterError> { + self.poll_persistence(now, wall_slot, emit)?; + if self.node.is_none() { + return Ok(()); + } + if let Some(node) = self.node.as_mut() { + self.snapshot_transfers.expire(node, now); + } self.tick_elapsed(now); self.maybe_propose_minimum_slot(wall_slot)?; - while self.node.has_ready() { - self.process_ready(now, wall_slot, &mut emit)?; + while !self.persistence.is_pending() && self.node.as_ref().is_some_and(RawNode::has_ready) { + self.process_ready(now, emit)?; + self.poll_persistence(now, wall_slot, emit)?; } - Ok(()) + self.maybe_compact() + } + + fn maybe_compact(&mut self) -> Result<(), ClusterError> { + if self.persistence.is_pending() { + return Ok(()); + } + let Some(node) = &self.node else { + return Ok(()); + }; + let applied = node.raft.raft_log.applied; + if applied == 0 || + (applied.saturating_sub(node.store().snapshot_index()) < + self.config.snapshot_interval && + !self.persistence.compaction_due()) + { + return Ok(()); + } + let data = self.state.encode_snapshot().map_err(ClusterError::Snapshot)?; + let (snapshot, hard_state) = node.store().checkpoint(applied, data)?; + let entries = node.store().suffix(applied)?; + self.persistence.compact(snapshot, &entries, &hard_state).map_err(ClusterError::Storage) } fn tick_elapsed(&mut self, now: Instant) { + let Some(node) = self.node.as_mut() else { + return; + }; let mut ticks = 0; while now >= self.next_tick && ticks < MAX_ELAPSED_TICKS_PER_SPIN { - self.node.tick(); + node.tick(); ticks += 1; - self.next_tick = self.next_tick.checked_add(self.tick_interval).unwrap_or(now); + self.next_tick = self.next_tick.checked_add(self.config.tick_interval).unwrap_or(now); } if now >= self.next_tick { // A long-stalled tile does not need to replay an unbounded number of // obsolete heartbeat intervals in one invocation. - self.next_tick = now.checked_add(self.tick_interval).unwrap_or(now); + self.next_tick = now.checked_add(self.config.tick_interval).unwrap_or(now); } } @@ -422,7 +558,10 @@ impl AttestationCluster { return Ok(()); } - self.node.propose(Vec::new(), ReplicatedCommand::AdvanceMinimumSlot(desired).encode())?; + self.node + .as_mut() + .ok_or(ClusterError::NotReady)? + .propose(Vec::new(), ReplicatedCommand::AdvanceMinimumSlot(desired).encode())?; self.pending_minimum_slot = Some(desired); Ok(()) } @@ -430,52 +569,130 @@ impl AttestationCluster { fn process_ready( &mut self, now: Instant, - wall_slot: u64, emit: &mut impl FnMut(ClusterEvent), ) -> Result<(), ClusterError> { - let mut ready = self.node.ready(); - - for message in ready.take_messages() { - emit(ClusterEvent::SendRaftMessage(message)); - } - + let node = self.node.as_mut().ok_or(ClusterError::NotReady)?; + let mut ready = node.ready(); + + let snapshot = if ready.snapshot().is_empty() { + None + } else { + let restored = RestoredSnapshot::decode(ready.snapshot(), &self.config.voters) + .map_err(ClusterError::Snapshot)?; + if restored.locks.minimum_slot() < self.state.minimum_slot() { + return Err(ClusterError::Snapshot(io::Error::new( + io::ErrorKind::InvalidData, + "snapshot retention floor regressed", + ))); + } + Some(restored) + }; + self.persistence.submit(&mut ready, snapshot).map_err(ClusterError::Storage)?; if !ready.snapshot().is_empty() { - // State-machine snapshot encoding will be added with durable - // storage. Refuse an incomplete snapshot rather than silently - // losing attestation locks. - return Err(ClusterError::SnapshotsUnsupported); + node.mut_store().install_snapshot(ready.snapshot().clone())?; } - { - let mut storage = self.node.mut_store().wl(); + let mut storage = node.mut_store().memory.wl(); storage.append(ready.entries())?; if let Some(hard_state) = ready.hs() { storage.set_hardstate(hard_state.clone()); } } - for message in ready.take_persisted_messages() { + for message in ready.take_messages() { + self.snapshot_transfers.sent(&message, now); emit(ClusterEvent::SendRaftMessage(message)); } + // Release Ready immediately so step/propose/tick remain legal during disk I/O. + // Durability is acknowledged separately through on_persist_ready. + node.advance_append_async(ready); + Ok(()) + } - self.apply_entries(ready.take_committed_entries(), now, wall_slot, emit)?; + fn poll_persistence( + &mut self, + now: Instant, + wall_slot: u64, + emit: &mut impl FnMut(ClusterEvent), + ) -> Result<(), ClusterError> { + match self.persistence.poll().map_err(ClusterError::Storage)? { + #[cfg(target_os = "linux")] + Some(PersistenceEvent::Recovered(recovered)) => self.recover(recovered, now), + Some(PersistenceEvent::Persisted(ready)) => { + self.complete_ready(ready, now, wall_slot, emit) + } + Some(PersistenceEvent::Compacted(snapshot)) => { + let index = snapshot.get_metadata().index; + self.node.as_mut().ok_or(ClusterError::NotReady)?.mut_store().compact(snapshot)?; + tracing::debug!( + node_id = self.config.node_id, + index, + "attestation Raft journal compacted" + ); + Ok(()) + } + None => Ok(()), + } + } + + #[cfg(any(target_os = "linux", test))] + fn recover(&mut self, recovered: RecoveredStorage, now: Instant) -> Result<(), ClusterError> { + if self.node.is_some() { + return Err(ClusterError::InvalidConfig("Raft already recovered")); + } + let mut storage = RaftStorage::new(self.config.voters.clone()); + if !recovered.snapshot.is_empty() { + self.state = RestoredSnapshot::decode(&recovered.snapshot, &self.config.voters) + .map_err(ClusterError::Snapshot)? + .locks; + storage.install_snapshot(recovered.snapshot)?; + } + storage.memory.wl().append(&recovered.entries)?; + let commit = recovered.hard_state.commit; + storage.memory.wl().set_hardstate(recovered.hard_state); + self.apply_entries( + recovered.entries.into_iter().take_while(|entry| entry.index <= commit), + now, + 0, + &mut |_| {}, + )?; + let mut config = self.config.raft_config(); + config.applied = commit; + self.node = Some(RawNode::with_default_logger(&config, storage)?); + self.next_tick = now.checked_add(self.config.tick_interval).unwrap_or(now); + tracing::info!(node_id = self.config.node_id, commit, "attestation Raft storage recovered"); + Ok(()) + } - let mut light_ready = self.node.advance(ready); - if let Some(commit_index) = light_ready.commit_index() { - self.node.mut_store().wl().mut_hard_state().set_commit(commit_index); + fn complete_ready( + &mut self, + ready: PersistedReady, + now: Instant, + wall_slot: u64, + emit: &mut impl FnMut(ClusterEvent), + ) -> Result<(), ClusterError> { + let node = self.node.as_mut().ok_or(ClusterError::NotReady)?; + node.on_persist_ready(ready.number); + let snapshot_index = ready.snapshot.as_ref().map(|snapshot| snapshot.index); + if let Some(snapshot) = ready.snapshot { + self.state = snapshot.locks; + self.pending_minimum_slot = None; } - for message in light_ready.take_messages() { + for message in ready.messages { + self.snapshot_transfers.sent(&message, now); emit(ClusterEvent::SendRaftMessage(message)); } - self.apply_entries(light_ready.take_committed_entries(), now, wall_slot, emit)?; - self.node.advance_apply(); - + let applied = ready.committed_entries.last().map(|entry| entry.index).or(snapshot_index); + self.apply_entries(ready.committed_entries, now, wall_slot, emit)?; + if let Some(applied) = applied { + self.node.as_mut().ok_or(ClusterError::NotReady)?.advance_apply_to(applied); + } Ok(()) } fn apply_entries( &mut self, - entries: Vec, + entries: impl IntoIterator, now: Instant, wall_slot: u64, emit: &mut impl FnMut(ClusterEvent), @@ -493,7 +710,8 @@ impl AttestationCluster { let proposal_id = ProposalId::decode(&entry.context) .map_err(ClusterError::InvalidProposalContextLength)?; let result = self.state.apply(&command); - if proposal_id.origin_node_id == self.node_id && + if proposal_id.origin_node_id == self.config.node_id && + proposal_id.incarnation == self.incarnation && let Some(pending) = self.take_pending_proposal(proposal_id) { if now >= pending.deadline { @@ -533,6 +751,14 @@ impl AttestationCluster { } } +#[cfg(test)] +#[path = "node/persistence_tests.rs"] +mod persistence_tests; + +#[cfg(test)] +#[path = "node/snapshot_tests.rs"] +mod snapshot_tests; + #[cfg(test)] mod tests { use super::*; @@ -549,7 +775,11 @@ mod tests { } fn test_config(node_id: u64, voters: Vec) -> AttestationClusterConfig { - let mut config = AttestationClusterConfig::new(node_id, voters); + let mut config = AttestationClusterConfig::new( + node_id, + voters, + ClusterStorageConfig::Create("unused-test-journal".into()), + ); config.tick_interval = Duration::from_millis(1); config.heartbeat_ticks = 1; config.election_ticks = 5; @@ -561,7 +791,7 @@ mod tests { startup_wall_slot: u64, now: Instant, ) -> AttestationCluster { - let mut cluster = AttestationCluster::new(config, now).unwrap(); + let mut cluster = AttestationCluster::in_memory(config, now).unwrap(); assert!(cluster.set_startup_wall_slot(startup_wall_slot)); cluster } @@ -629,7 +859,7 @@ mod tests { fn proposal_times_out_at_one_hundred_milliseconds_and_remains_locked() { let now = Instant::now(); let mut cluster = initialized_cluster(test_config(1, vec![1]), 9, now); - assert_eq!(cluster.proposal_timeout, Duration::from_millis(100)); + assert_eq!(cluster.config.proposal_timeout, Duration::from_millis(100)); cluster.campaign().unwrap(); let proposal_id = cluster.propose_attestation(command(10, 1), 10, now).unwrap(); @@ -682,7 +912,7 @@ mod tests { #[test] fn proposal_checks_local_admission_before_raft() { let now = Instant::now(); - let mut cluster = AttestationCluster::new(test_config(1, vec![1]), now).unwrap(); + let mut cluster = AttestationCluster::in_memory(test_config(1, vec![1]), now).unwrap(); cluster.campaign().unwrap(); assert!(matches!( diff --git a/crates/control/src/cluster/node/persistence_tests.rs b/crates/control/src/cluster/node/persistence_tests.rs new file mode 100644 index 00000000..ebca0823 --- /dev/null +++ b/crates/control/src/cluster/node/persistence_tests.rs @@ -0,0 +1,403 @@ +use raft::eraftpb::MessageType; +use silver_common::ssz_view::SINGLE_ATT_SIZE; + +use super::*; +use crate::cluster::AttestationKey; + +fn config(node_id: u64, voters: Vec) -> AttestationClusterConfig { + AttestationClusterConfig::new( + node_id, + voters, + ClusterStorageConfig::Create("unused-test-journal".into()), + ) +} + +fn command(slot: u64, root: u8) -> AttestationLockCommand { + let mut ssz = [0; SINGLE_ATT_SIZE]; + ssz[0] = root; + AttestationLockCommand { key: AttestationKey { attester_index: 7, slot }, subnet: 0, ssz } +} + +fn memory(now: Instant, voters: Vec) -> AttestationCluster { + let mut cluster = AttestationCluster::in_memory(config(1, voters), now).unwrap(); + cluster.set_startup_wall_slot(9); + cluster +} + +fn leader(now: Instant) -> AttestationCluster { + let mut cluster = memory(now, vec![1]); + cluster.campaign().unwrap(); + cluster.spin(now, 10, |_| {}).unwrap(); + assert!(cluster.is_leader()); + cluster +} + +fn events(cluster: &mut AttestationCluster, now: Instant) -> Vec { + let mut events = Vec::new(); + cluster.spin(now, 10, |event| events.push(event)).unwrap(); + events +} + +#[test] +fn vote_responses_wait_for_persistence_and_steps_remain_legal_while_waiting() { + let now = Instant::now(); + let mut cluster = memory(now, vec![1, 2, 3]); + cluster.persistence.paused = true; + cluster + .step(Message { + msg_type: MessageType::MsgRequestVote, + from: 2, + to: 1, + term: 1, + ..Message::default() + }) + .unwrap(); + assert!(events(&mut cluster, now).is_empty()); + assert!(cluster.persistence.is_pending()); + + cluster + .step(Message { + msg_type: MessageType::MsgHeartbeat, + from: 3, + to: 1, + term: 2, + ..Message::default() + }) + .unwrap(); + assert!(events(&mut cluster, now).is_empty()); + assert_eq!(cluster.node.as_ref().unwrap().raft.term, 2); + + cluster.persistence.paused = false; + let messages = events(&mut cluster, now); + assert!(messages.iter().any(|event| matches!(event, + ClusterEvent::SendRaftMessage(message) if message.msg_type == MessageType::MsgRequestVoteResponse && message.to == 2 + ))); + assert!(messages.iter().any(|event| matches!(event, + ClusterEvent::SendRaftMessage(message) if message.msg_type == MessageType::MsgHeartbeatResponse && message.to == 3 + ))); +} + +#[test] +fn pending_disk_io_does_not_stop_proposal_timeouts_or_accept_late_decisions() { + let now = Instant::now(); + let mut cluster = leader(now); + cluster.persistence.paused = true; + let first = cluster.propose_attestation(command(10, 1), 10, now).unwrap(); + assert!(events(&mut cluster, now).is_empty()); + assert_eq!(cluster.state.len(), 0); + let mut other = command(10, 2); + other.key.attester_index = 8; + let second = cluster.propose_attestation(other, 10, now).unwrap(); + let later = now + Duration::from_millis(100); + let expired = events(&mut cluster, later); + assert!(matches!(&expired[..], [ + ClusterEvent::AttestationProposalTimedOut(a), ClusterEvent::AttestationProposalTimedOut(b) + ] if *a == first && *b == second)); + assert_eq!(cluster.state.len(), 0); + + cluster.persistence.paused = false; + assert!(events(&mut cluster, later).is_empty()); + assert_eq!(cluster.state.len(), 2); + assert_eq!(cluster.state.apply(&command(10, 3)), LockResult::ConflictingAttestation); +} + +#[test] +fn persistence_failure_stops_votes_proposals_and_committed_decisions() { + let now = Instant::now(); + let mut cluster = leader(now); + cluster.persistence.paused = true; + cluster.propose_attestation(command(10, 1), 10, now).unwrap(); + assert!(events(&mut cluster, now).is_empty()); + cluster.fail_persistence(); + assert!(matches!( + cluster.spin(now, 10, |_| panic!("released failed write")), + Err(ClusterError::Storage(_)) + )); + assert!(cluster.is_failed()); + assert!(!cluster.is_ready()); + assert!(!cluster.is_leader()); + assert_eq!(cluster.pending_proposals(), 0); + assert!(matches!(cluster.campaign(), Err(ClusterError::Failed))); + assert!(matches!(cluster.step(Message::default()), Err(ClusterError::Failed))); + assert!(matches!( + cluster.propose_attestation(command(10, 2), 10, now), + Err(ProposeError::Failed) + )); + assert!(events(&mut cluster, now).is_empty()); + assert_eq!(cluster.state.len(), 0); +} + +#[test] +fn previous_incarnation_cannot_complete_a_new_request_with_the_same_sequence() { + let now = Instant::now(); + let mut old = leader(now); + let previous = old.propose_attestation(command(10, 1), 10, now).unwrap(); + let mut cluster = leader(now); + let current = cluster.propose_attestation(command(10, 2), 10, now).unwrap(); + assert_eq!(previous.sequence, current.sequence); + assert_ne!(previous.incarnation, current.incarnation); + cluster + .apply_entries( + [Entry { + data: ReplicatedCommand::Lock(command(10, 1)).encode().into(), + context: previous.encode().into(), + ..Entry::default() + }], + now, + 10, + &mut |_| panic!("previous incarnation completed new request"), + ) + .unwrap(); + assert_eq!(cluster.pending_proposals(), 1); + assert!(events(&mut cluster, now).iter().any(|event| matches!(event, + ClusterEvent::AttestationCommitted(decision) if decision.proposal_id == current && decision.result == LockResult::ConflictingAttestation + ))); + assert_eq!(cluster.pending_proposals(), 0); +} + +#[test] +fn proposal_contexts_round_trip_and_legacy_contexts_cannot_match_new_requests() { + let now = Instant::now(); + let mut cluster = leader(now); + let proposal = cluster.propose_attestation(command(10, 1), 10, now).unwrap(); + let encoded = proposal.encode(); + assert_eq!(ProposalId::decode(&encoded).unwrap(), proposal); + let legacy = ProposalId::decode(&encoded[..16]).unwrap(); + assert_eq!(legacy.incarnation, [0; 16]); + assert_ne!(legacy, proposal); + for len in [0, 15, 17, 31] { + assert!(ProposalId::decode(&encoded[..len]).is_err()); + } +} + +#[cfg(target_os = "linux")] +mod disk { + use std::{path::Path, thread}; + + use super::*; + + fn stored_config( + path: &Path, + create: bool, + node: u64, + voters: Vec, + ) -> AttestationClusterConfig { + let mut config = config(node, voters); + config.storage = if create { + ClusterStorageConfig::Create(path.into()) + } else { + ClusterStorageConfig::Open(path.into()) + }; + config + } + + fn quiet(cluster: &AttestationCluster) -> bool { + cluster.is_ready() && + !cluster.persistence.is_pending() && + !cluster.node.as_ref().unwrap().has_ready() + } + + fn drain(cluster: &mut AttestationCluster, now: Instant, slot: u64) -> Vec { + let deadline = Instant::now() + Duration::from_secs(5); + let mut events = Vec::new(); + loop { + cluster.spin(now, slot, |event| events.push(event)).unwrap(); + if quiet(cluster) { + return events; + } + assert!(Instant::now() < deadline, "journal did not become idle"); + thread::yield_now(); + } + } + + #[test] + fn recovery_gates_participation_and_restores_epoch_locks_and_retention_floor() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let now = Instant::now(); + let mut cluster = + AttestationCluster::new(stored_config(&path, true, 1, vec![1]), now).unwrap(); + cluster.set_startup_wall_slot(99); + assert!(matches!(cluster.campaign(), Err(ClusterError::NotReady))); + assert!(matches!(cluster.step(Message::default()), Err(ClusterError::NotReady))); + assert!(matches!( + cluster.propose_attestation(command(100, 1), 100, now), + Err(ProposeError::NotReady) + )); + assert!(drain(&mut cluster, now, 100).is_empty()); + cluster.campaign().unwrap(); + drain(&mut cluster, now, 100); + let old_id = cluster.propose_attestation(command(100, 1), 100, now).unwrap(); + assert!(drain(&mut cluster, now, 100).iter().any(|event| matches!(event, + ClusterEvent::AttestationCommitted(decision) if decision.result == LockResult::Accepted + ))); + let node = cluster.node.as_ref().unwrap(); + let hard_state = node.store().memory.rl().hard_state().clone(); + assert_eq!(hard_state.commit, node.raft.raft_log.applied); + drop(cluster); + + let mut cluster = + AttestationCluster::new(stored_config(&path, false, 1, vec![1]), now).unwrap(); + cluster.set_startup_wall_slot(104); + assert!(drain(&mut cluster, now, 105).is_empty()); + assert_eq!(cluster.state.len(), 1); + assert_eq!(cluster.state.minimum_slot(), 68); + assert_eq!(cluster.node.as_ref().unwrap().raft.hard_state(), hard_state); + assert_eq!(cluster.node.as_ref().unwrap().raft.raft_log.applied, hard_state.commit); + cluster.campaign().unwrap(); + drain(&mut cluster, now, 105); + let new_id = cluster.propose_attestation(command(105, 2), 105, now).unwrap(); + assert_ne!(old_id, new_id); + assert!(drain(&mut cluster, now, 105).iter().any(|event| matches!(event, + ClusterEvent::AttestationCommitted(decision) if decision.proposal_id == new_id && decision.result == LockResult::ConflictingAttestation + ))); + } + + #[test] + fn recovered_uncommitted_suffix_is_not_applied_and_can_be_replaced() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("raft.wal"); + let now = Instant::now(); + let mut cluster = + AttestationCluster::new(stored_config(&path, true, 1, vec![1, 2, 3]), now).unwrap(); + drain(&mut cluster, now, 10); + let context = ProposalId { origin_node_id: 2, sequence: 1, incarnation: [1; 16] }.encode(); + cluster + .step(Message { + msg_type: MessageType::MsgAppend, + from: 2, + to: 1, + term: 1, + entries: vec![Entry { + index: 1, + term: 1, + context: context.clone().into(), + data: ReplicatedCommand::Lock(command(10, 1)).encode().into(), + ..Entry::default() + }] + .into(), + ..Message::default() + }) + .unwrap(); + drain(&mut cluster, now, 10); + assert_eq!(cluster.state.len(), 0); + drop(cluster); + + let mut cluster = + AttestationCluster::new(stored_config(&path, false, 1, vec![1, 2, 3]), now).unwrap(); + assert!(drain(&mut cluster, now, 10).is_empty()); + assert_eq!(cluster.state.len(), 0); + assert_eq!(cluster.node.as_ref().unwrap().raft.raft_log.last_index(), 1); + cluster + .step(Message { + msg_type: MessageType::MsgAppend, + from: 2, + to: 1, + term: 2, + commit: 1, + entries: vec![Entry { + index: 1, + term: 2, + context: context.into(), + data: ReplicatedCommand::Lock(command(10, 2)).encode().into(), + ..Entry::default() + }] + .into(), + ..Message::default() + }) + .unwrap(); + drain(&mut cluster, now, 10); + assert_eq!(cluster.state.len(), 1); + assert_eq!(cluster.state.apply(&command(10, 2)), LockResult::AlreadyAcceptedSame); + assert_eq!(cluster.state.apply(&command(10, 1)), LockResult::ConflictingAttestation); + drop(cluster); + + let mut cluster = + AttestationCluster::new(stored_config(&path, false, 1, vec![1, 2, 3]), now).unwrap(); + assert!(drain(&mut cluster, now, 10).is_empty()); + assert_eq!(cluster.state.len(), 1); + assert_eq!(cluster.state.apply(&command(10, 2)), LockResult::AlreadyAcceptedSame); + } + + #[test] + fn missing_journal_disables_cluster_without_bootstrapping_it() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("missing.wal"); + let now = Instant::now(); + let mut cluster = + AttestationCluster::new(stored_config(&path, false, 1, vec![1]), now).unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while !cluster.is_failed() { + let _ = cluster.spin(now, 10, |_| panic!("unrecovered node emitted work")); + assert!(Instant::now() < deadline); + thread::yield_now(); + } + assert!(!path.exists()); + assert!(matches!(cluster.campaign(), Err(ClusterError::Failed))); + } + + #[test] + fn three_durable_nodes_commit_a_follower_proposal_and_recover_the_same_lock() { + let dir = tempfile::tempdir().unwrap(); + let now = Instant::now(); + let voters = vec![1, 2, 3]; + let paths: Vec<_> = voters.iter().map(|id| dir.path().join(format!("{id}.wal"))).collect(); + let mut nodes: Vec<_> = voters + .iter() + .zip(&paths) + .map(|(id, path)| { + let mut node = + AttestationCluster::new(stored_config(path, true, *id, voters.clone()), now) + .unwrap(); + node.set_startup_wall_slot(9); + drain(&mut node, now, 10); + node + }) + .collect(); + let mut messages = Vec::new(); + let mut decisions = Vec::new(); + let mut pump = |nodes: &mut [AttestationCluster]| { + for node in nodes.iter_mut() { + node.spin(now, 10, |event| match event { + ClusterEvent::SendRaftMessage(message) => messages.push(message), + ClusterEvent::AttestationCommitted(decision) => decisions.push(decision), + ClusterEvent::AttestationProposalTimedOut(_) => panic!("unexpected timeout"), + }) + .unwrap(); + } + for message in messages.drain(..) { + nodes + .iter_mut() + .find(|node| node.node_id() == message.to) + .unwrap() + .step(message) + .unwrap(); + } + }; + nodes[0].campaign().unwrap(); + let deadline = Instant::now() + Duration::from_secs(5); + while !nodes[0].is_leader() || !nodes.iter().all(quiet) { + pump(&mut nodes); + assert!(Instant::now() < deadline); + thread::yield_now(); + } + let proposal = nodes[1].propose_attestation(command(10, 4), 10, now).unwrap(); + while !nodes.iter().all(|node| quiet(node) && node.state.len() == 1) { + pump(&mut nodes); + assert!(Instant::now() < deadline); + thread::yield_now(); + } + assert_eq!(decisions.len(), 1); + assert_eq!(decisions[0].proposal_id, proposal); + assert!(decisions[0].may_validate()); + drop(nodes); + for (id, path) in voters.iter().zip(&paths) { + let mut node = + AttestationCluster::new(stored_config(path, false, *id, voters.clone()), now) + .unwrap(); + assert!(drain(&mut node, now, 10).is_empty()); + assert_eq!(node.state.len(), 1); + assert_eq!(node.state.apply(&command(10, 5)), LockResult::ConflictingAttestation); + } + } +} diff --git a/crates/control/src/cluster/node/snapshot_tests.rs b/crates/control/src/cluster/node/snapshot_tests.rs new file mode 100644 index 00000000..5172fdc5 --- /dev/null +++ b/crates/control/src/cluster/node/snapshot_tests.rs @@ -0,0 +1,301 @@ +use raft::{ + Storage, + eraftpb::{HardState, MessageType, Snapshot}, +}; +use silver_common::ssz_view::SINGLE_ATT_SIZE; + +use super::*; +use crate::cluster::AttestationKey; + +fn config(id: u64) -> AttestationClusterConfig { + let mut config = AttestationClusterConfig::new( + id, + vec![1, 2, 3], + ClusterStorageConfig::Create("unused-test-journal".into()), + ); + config.snapshot_interval = 4; + config +} + +fn command(validator: u64, root: u8) -> AttestationLockCommand { + AttestationLockCommand { + key: AttestationKey { attester_index: validator, slot: 100 }, + subnet: 0, + ssz: [root; SINGLE_ATT_SIZE], + } +} + +fn snapshot(index: u64) -> Snapshot { + let mut state = AttestationLockStore::default(); + state.advance_minimum_slot(68); + state.apply(&command(7, 1)); + let mut snapshot = + Snapshot { data: state.encode_snapshot().unwrap().into(), ..Snapshot::default() }; + let metadata = snapshot.mut_metadata(); + metadata.index = index; + metadata.term = 2; + metadata.mut_conf_state().voters = vec![1, 2, 3]; + snapshot +} + +fn snapshot_message(snapshot: Snapshot) -> Message { + let mut message = Message { + from: 2, + to: 1, + term: 2, + msg_type: MessageType::MsgSnapshot, + ..Message::default() + }; + message.set_snapshot(snapshot); + message +} + +fn quiet(node: &AttestationCluster) -> bool { + node.is_ready() && !node.persistence.is_pending() && !node.node.as_ref().unwrap().has_ready() +} + +#[test] +fn snapshot_installation_and_acknowledgement_wait_for_durability() { + let now = Instant::now(); + let mut node = AttestationCluster::in_memory(config(1), now).unwrap(); + node.set_startup_wall_slot(99); + node.persistence.paused = true; + node.step(snapshot_message(snapshot(10))).unwrap(); + node.spin(now, 100, |_| panic!("snapshot acknowledged before sync")).unwrap(); + assert_eq!(node.state.len(), 0); + assert_eq!(node.node.as_ref().unwrap().raft.raft_log.applied, 0); + node.persistence.paused = false; + let mut messages = Vec::new(); + node.spin(now, 100, |event| messages.push(event)).unwrap(); + assert!(messages.iter().any(|event| matches!(event, ClusterEvent::SendRaftMessage(message) + if message.msg_type == MessageType::MsgAppendResponse && message.index == 10))); + assert_eq!(node.node.as_ref().unwrap().raft.raft_log.applied, 10); + assert_eq!(node.state.minimum_slot(), 68); + assert_eq!(node.state.apply(&command(7, 2)), LockResult::ConflictingAttestation); + assert!( + node.admission.validate(99, 100).is_err(), + "snapshot must not replace startup admission" + ); +} + +#[test] +fn invalid_snapshot_payload_or_membership_stops_participation() { + for bad_membership in [false, true] { + let now = Instant::now(); + let mut node = AttestationCluster::in_memory(config(1), now).unwrap(); + let mut snapshot = snapshot(10); + if bad_membership { + snapshot.mut_metadata().mut_conf_state().voters = vec![1, 4]; + } else { + snapshot.data = vec![0; 20].into(); + } + node.step(snapshot_message(snapshot)).unwrap(); + assert!(matches!( + node.spin(now, 100, |_| panic!("released invalid snapshot")), + Err(ClusterError::Snapshot(_)) + )); + assert!(node.is_failed()); + } +} + +#[test] +fn storage_serves_captured_state_not_a_snapshot_at_an_unapplied_commit() { + let mut storage = RaftStorage::new(vec![1, 2, 3]); + storage.install_snapshot(snapshot(10)).unwrap(); + storage.memory.wl().append(&[Entry { index: 11, term: 2, ..Entry::default() }]).unwrap(); + storage.memory.wl().set_hardstate(HardState { term: 2, commit: 11, ..HardState::default() }); + assert_eq!(storage.snapshot(0, 3).unwrap(), snapshot(10)); + assert!(storage.snapshot(11, 3).is_err()); + assert_eq!(storage.term(10).unwrap(), 2); + assert_eq!(storage.first_index().unwrap(), 11); + assert_eq!(storage.last_index().unwrap(), 11); +} + +struct Network { + nodes: Vec, + now: Instant, + isolated: Option, + drop_snapshot: bool, + drop_snapshot_ack: bool, + snapshots: usize, + decisions: Vec, +} + +impl Network { + fn new(nodes: Vec, now: Instant) -> Self { + Self { + nodes, + now, + isolated: None, + drop_snapshot: false, + drop_snapshot_ack: false, + snapshots: 0, + decisions: Vec::new(), + } + } + + fn settle(&mut self) { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let mut messages = Vec::new(); + for node in &mut self.nodes { + node.spin(self.now, 100, |event| match event { + ClusterEvent::SendRaftMessage(message) => messages.push(message), + ClusterEvent::AttestationCommitted(decision) => self.decisions.push(decision), + ClusterEvent::AttestationProposalTimedOut(_) => panic!("proposal timed out"), + }) + .unwrap(); + } + for message in messages { + if self.isolated.is_some_and(|id| message.from == id || message.to == id) { + continue; + } + if message.msg_type == MessageType::MsgSnapshot { + self.snapshots += 1; + if self.drop_snapshot { + self.drop_snapshot = false; + continue; + } + } + if self.drop_snapshot_ack && + self.snapshots != 0 && + message.from == 3 && + message.msg_type == MessageType::MsgAppendResponse + { + self.drop_snapshot_ack = false; + continue; + } + self.nodes + .iter_mut() + .find(|node| node.node_id() == message.to) + .unwrap() + .step(message) + .unwrap(); + } + if self.nodes.iter().all(quiet) { + return; + } + assert!(Instant::now() < deadline, "Raft cluster did not become idle"); + std::thread::yield_now(); + } + } + + fn populate_with_one_follower_offline(&mut self) { + self.settle(); + for node in &mut self.nodes { + node.set_startup_wall_slot(99); + } + self.isolated = Some(3); + self.nodes[0].campaign().unwrap(); + self.settle(); + assert!(self.nodes[0].is_leader()); + for validator in 0..24 { + self.nodes[0].propose_attestation(command(validator, 1), 100, self.now).unwrap(); + self.settle(); + } + assert_eq!(self.decisions.len(), 24); + assert!(self.decisions.iter().all(AttestationDecision::may_validate)); + assert!(self.nodes[0].node.as_ref().unwrap().store().first_index().unwrap() > 1); + assert_eq!(self.nodes[2].state.len(), 0); + } + + fn reconnect(&mut self) { + self.isolated = None; + for _ in 0..70 { + self.now += Duration::from_millis(100); + self.settle(); + if self.nodes[2].state.len() == 24 { + return; + } + } + panic!("lagging follower did not install snapshot"); + } +} + +#[test] +fn a_lost_snapshot_is_retried_and_a_lagging_follower_catches_up() { + let now = Instant::now(); + let nodes = (1..=3).map(|id| AttestationCluster::in_memory(config(id), now).unwrap()).collect(); + let mut network = Network::new(nodes, now); + network.populate_with_one_follower_offline(); + network.drop_snapshot = true; + network.reconnect(); + assert!(network.snapshots >= 2); + for node in &mut network.nodes { + assert_eq!(node.state.apply(&command(7, 2)), LockResult::ConflictingAttestation); + } + let request = network.nodes[2].propose_attestation(command(50, 1), 100, network.now).unwrap(); + network.settle(); + assert!( + network + .decisions + .iter() + .any(|decision| decision.proposal_id == request && decision.may_validate()) + ); +} + +#[test] +fn a_lost_snapshot_ack_does_not_stall_replication() { + let now = Instant::now(); + let nodes = (1..=3).map(|id| AttestationCluster::in_memory(config(id), now).unwrap()).collect(); + let mut network = Network::new(nodes, now); + network.populate_with_one_follower_offline(); + network.drop_snapshot_ack = true; + network.reconnect(); + assert!(!network.drop_snapshot_ack); + for _ in 0..60 { + network.now += Duration::from_millis(100); + network.settle(); + } + assert!(network.snapshots >= 2); + network.nodes[0].propose_attestation(command(50, 1), 100, network.now).unwrap(); + network.settle(); + assert_eq!(network.nodes[2].state.len(), 25); +} + +#[test] +fn compacting_again_at_the_same_applied_index_preserves_the_suffix() { + let mut storage = RaftStorage::new(vec![1, 2, 3]); + storage.install_snapshot(snapshot(10)).unwrap(); + let entry = Entry { index: 11, term: 2, ..Entry::default() }; + storage.memory.wl().append(&[entry.clone()]).unwrap(); + storage.compact(snapshot(10)).unwrap(); + assert_eq!(storage.suffix(10).unwrap(), [entry]); + assert_eq!(storage.snapshot(0, 3).unwrap(), snapshot(10)); +} + +#[cfg(target_os = "linux")] +#[test] +fn durable_snapshot_catch_up_survives_a_full_cluster_restart() { + let directory = tempfile::tempdir().unwrap(); + let now = Instant::now(); + let nodes = (1..=3) + .map(|id| { + let mut config = config(id); + config.storage = + ClusterStorageConfig::Create(directory.path().join(format!("{id}.wal"))); + AttestationCluster::new(config, now).unwrap() + }) + .collect(); + let mut network = Network::new(nodes, now); + network.populate_with_one_follower_offline(); + network.reconnect(); + assert!(network.snapshots > 0); + drop(network); + let nodes = (1..=3) + .map(|id| { + let mut config = config(id); + config.storage = ClusterStorageConfig::Open(directory.path().join(format!("{id}.wal"))); + AttestationCluster::new(config, now).unwrap() + }) + .collect(); + let mut network = Network::new(nodes, now); + network.settle(); + assert!(network.decisions.is_empty()); + for node in &mut network.nodes { + assert_eq!(node.state.len(), 24); + assert_eq!(node.state.minimum_slot(), 68); + assert_eq!(node.state.apply(&command(7, 2)), LockResult::ConflictingAttestation); + assert!(node.node.as_ref().unwrap().store().snapshot_index() > 0); + } +} diff --git a/crates/control/src/cluster/persistence.rs b/crates/control/src/cluster/persistence.rs new file mode 100644 index 00000000..abdbd531 --- /dev/null +++ b/crates/control/src/cluster/persistence.rs @@ -0,0 +1,208 @@ +use std::{io, path::PathBuf}; + +use raft::{ + Ready, + eraftpb::{Entry, HardState, Message, Snapshot}, +}; + +#[cfg(target_os = "linux")] +use super::storage::{ClusterStorage, ClusterStorageEvent, StorageIdentity}; +use super::{node::AttestationClusterConfig, raft_storage::RestoredSnapshot}; + +#[derive(Debug, Clone)] +pub enum ClusterStorageConfig { + Create(PathBuf), + Open(PathBuf), +} + +#[derive(Debug, Default)] +pub struct RecoveredStorage { + pub hard_state: HardState, + pub snapshot: Snapshot, + /// Includes the uncommitted suffix; replay only entries through + /// `hard_state.commit` into the state machine. + pub entries: Vec, +} + +pub(super) struct PersistedReady { + pub number: u64, + pub messages: Vec, + pub committed_entries: Vec, + pub snapshot: Option, + wait_for_disk: bool, +} + +pub(super) enum PersistenceEvent { + #[cfg(target_os = "linux")] + Recovered(RecoveredStorage), + Persisted(PersistedReady), + Compacted(Snapshot), +} + +enum Pending { + Ready(PersistedReady), + Compact(Snapshot), +} + +impl Pending { + fn into_event(self) -> PersistenceEvent { + match self { + Self::Ready(ready) => PersistenceEvent::Persisted(ready), + Self::Compact(snapshot) => PersistenceEvent::Compacted(snapshot), + } + } +} + +pub(super) struct Persistence { + #[cfg(target_os = "linux")] + disk: Option, + pending: Option, + #[cfg(test)] + pub paused: bool, + #[cfg(test)] + pub fail: bool, +} + +impl Persistence { + pub fn new(config: &AttestationClusterConfig) -> io::Result { + #[cfg(target_os = "linux")] + { + let identity = StorageIdentity::new(config.node_id, config.voters.clone())?; + let disk = match &config.storage { + ClusterStorageConfig::Create(path) => ClusterStorage::create(path, identity)?, + ClusterStorageConfig::Open(path) => ClusterStorage::open(path, identity)?, + }; + Ok(Self { + disk: Some(disk), + pending: None, + #[cfg(test)] + paused: false, + #[cfg(test)] + fail: false, + }) + } + #[cfg(not(target_os = "linux"))] + { + let _ = config; + Err(io::Error::new(io::ErrorKind::Unsupported, "Raft persistence requires Linux")) + } + } + + #[cfg(test)] + pub fn memory() -> Self { + Self { + #[cfg(target_os = "linux")] + disk: None, + pending: None, + paused: false, + fail: false, + } + } + + pub fn is_pending(&self) -> bool { + self.pending.is_some() + } + + pub fn compaction_due(&self) -> bool { + #[cfg(target_os = "linux")] + if let Some(disk) = &self.disk { + return disk.appended_bytes() >= 16 * 1024 * 1024; + } + false + } + + pub fn compact( + &mut self, + snapshot: Snapshot, + entries: &[Entry], + hard_state: &HardState, + ) -> io::Result<()> { + if self.is_pending() { + return Err(io::Error::other("Raft persistence already has pending work")); + } + #[cfg(target_os = "linux")] + if let Some(disk) = &mut self.disk { + disk.checkpoint(None, &snapshot, entries, hard_state)?; + } + #[cfg(not(target_os = "linux"))] + let _ = (entries, hard_state); + self.pending = Some(Pending::Compact(snapshot)); + Ok(()) + } + + pub fn submit( + &mut self, + ready: &mut Ready, + snapshot: Option, + ) -> io::Result<()> { + if self.pending.is_some() { + return Err(io::Error::other("Raft persistence already has a pending Ready")); + } + let wait_for_disk = + !ready.entries().is_empty() || ready.hs().is_some() || snapshot.is_some(); + #[cfg(target_os = "linux")] + if wait_for_disk && let Some(disk) = &mut self.disk { + if snapshot.is_some() { + disk.checkpoint( + Some(ready.number()), + ready.snapshot(), + ready.entries(), + ready + .hs() + .ok_or_else(|| io::Error::other("Raft snapshot Ready has no hard state"))?, + )?; + } else { + disk.persist(ready.number(), ready.entries(), ready.hs())?; + } + } + self.pending = Some(Pending::Ready(PersistedReady { + number: ready.number(), + messages: ready.take_persisted_messages(), + committed_entries: ready.take_committed_entries(), + snapshot, + wait_for_disk, + })); + Ok(()) + } + + pub fn poll(&mut self) -> io::Result> { + #[cfg(test)] + { + if self.fail { + return Err(io::Error::other("injected Raft persistence failure")); + } + if self.paused { + return Ok(None); + } + } + + if matches!(&self.pending, Some(Pending::Ready(ready)) if !ready.wait_for_disk) { + return Ok(self.pending.take().map(Pending::into_event)); + } + #[cfg(target_os = "linux")] + if let Some(disk) = &mut self.disk { + return match disk.poll()? { + Some(ClusterStorageEvent::Recovered(recovered)) if self.pending.is_none() => { + Ok(Some(PersistenceEvent::Recovered(recovered))) + } + Some(ClusterStorageEvent::Persisted { ready_number }) => { + match self.pending.take() { + Some(Pending::Ready(ready)) if ready.number == ready_number => { + Ok(Some(PersistenceEvent::Persisted(ready))) + } + _ => Err(io::Error::other("unexpected Raft persistence completion")), + } + } + Some(ClusterStorageEvent::Compacted { index }) => match self.pending.take() { + Some(Pending::Compact(snapshot)) if snapshot.get_metadata().index == index => { + Ok(Some(PersistenceEvent::Compacted(snapshot))) + } + _ => Err(io::Error::other("unexpected Raft compaction completion")), + }, + Some(_) => Err(io::Error::other("unexpected Raft recovery completion")), + None => Ok(None), + }; + } + Ok(self.pending.take().map(Pending::into_event)) + } +} diff --git a/crates/control/src/cluster/raft_storage.rs b/crates/control/src/cluster/raft_storage.rs new file mode 100644 index 00000000..8a572795 --- /dev/null +++ b/crates/control/src/cluster/raft_storage.rs @@ -0,0 +1,144 @@ +use std::io; + +use raft::{ + GetEntriesContext, RaftState, Storage, StorageError, + eraftpb::{Entry, HardState, Snapshot}, + storage::MemStorage, +}; + +use super::lock_store::AttestationLockStore; + +pub(super) struct RestoredSnapshot { + pub index: u64, + pub locks: AttestationLockStore, +} + +impl RestoredSnapshot { + pub fn decode(snapshot: &Snapshot, voters: &[u64]) -> io::Result { + validate_snapshot(snapshot, voters)?; + Ok(Self { + index: snapshot.get_metadata().index, + locks: AttestationLockStore::decode_snapshot(&snapshot.data)?, + }) + } +} + +pub(super) struct RaftStorage { + pub memory: MemStorage, + snapshot: Snapshot, +} + +impl RaftStorage { + #[cfg(any(target_os = "linux", test))] + pub fn new(voters: Vec) -> Self { + Self { + memory: MemStorage::new_with_conf_state((voters, Vec::::new())), + snapshot: Snapshot::default(), + } + } + + pub fn snapshot_index(&self) -> u64 { + self.snapshot.get_metadata().index + } + + pub fn install_snapshot(&mut self, snapshot: Snapshot) -> raft::Result<()> { + self.memory.wl().apply_snapshot(snapshot.clone())?; + self.snapshot = snapshot; + Ok(()) + } + + pub fn suffix(&self, index: u64) -> raft::Result> { + let last = self.last_index()?; + if index >= last { + return Ok(Vec::new()); + } + self.entries(index + 1, last + 1, None, GetEntriesContext::empty(false)) + } + + pub fn compact(&mut self, snapshot: Snapshot) -> raft::Result<()> { + if snapshot.get_metadata().index == self.snapshot_index() { + self.snapshot = snapshot; + return Ok(()); + } + let entries = self.suffix(snapshot.get_metadata().index)?; + let hard_state = self.memory.rl().hard_state().clone(); + self.install_snapshot(snapshot)?; + let mut memory = self.memory.wl(); + memory.append(&entries)?; + memory.set_hardstate(hard_state); + Ok(()) + } + + pub fn checkpoint(&self, index: u64, data: Vec) -> raft::Result<(Snapshot, HardState)> { + let mut snapshot = Snapshot { data: data.into(), ..Snapshot::default() }; + let metadata = snapshot.mut_metadata(); + metadata.index = index; + metadata.term = self.term(index)?; + metadata.set_conf_state(self.initial_state()?.conf_state); + Ok((snapshot, self.memory.rl().hard_state().clone())) + } +} + +// MemStorage synthesizes empty snapshots at its commit index, which can exceed +// the applied index during asynchronous persistence. Serve only our captured +// state. +impl Storage for RaftStorage { + fn initial_state(&self) -> raft::Result { + self.memory.initial_state() + } + + fn entries( + &self, + low: u64, + high: u64, + max_size: impl Into>, + context: GetEntriesContext, + ) -> raft::Result> { + if low == high && low >= self.first_index()? && low <= self.last_index()?.saturating_add(1) + { + return Ok(Vec::new()); + } + self.memory.entries(low, high, max_size, context) + } + + fn term(&self, index: u64) -> raft::Result { + self.memory.term(index) + } + fn first_index(&self) -> raft::Result { + self.memory.first_index() + } + fn last_index(&self) -> raft::Result { + self.memory.last_index() + } + + fn snapshot(&self, request_index: u64, _to: u64) -> raft::Result { + if self.snapshot.is_empty() || request_index > self.snapshot_index() { + return Err(StorageError::SnapshotTemporarilyUnavailable.into()); + } + Ok(self.snapshot.clone()) + } +} + +pub(super) fn validate_snapshot(snapshot: &Snapshot, voters: &[u64]) -> io::Result<()> { + let metadata = snapshot.get_metadata(); + let conf = metadata.get_conf_state(); + let mut actual = conf.voters.clone(); + actual.sort_unstable(); + let mut expected = voters.to_vec(); + expected.sort_unstable(); + if metadata.index == 0 || + metadata.index == u64::MAX || + metadata.term == 0 || + actual != expected || + !conf.learners.is_empty() || + !conf.voters_outgoing.is_empty() || + !conf.learners_next.is_empty() || + conf.auto_leave + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid Raft snapshot metadata or membership", + )); + } + Ok(()) +} diff --git a/crates/control/src/cluster/snapshot_transfers.rs b/crates/control/src/cluster/snapshot_transfers.rs new file mode 100644 index 00000000..6f231f76 --- /dev/null +++ b/crates/control/src/cluster/snapshot_transfers.rs @@ -0,0 +1,53 @@ +use std::time::{Duration, Instant}; + +use raft::{ + ProgressState, RawNode, SnapshotStatus, StateRole, + eraftpb::{Message, MessageType}, +}; + +use super::raft_storage::RaftStorage; + +const TRANSFER_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Default)] +pub(super) struct SnapshotTransfers { + pending: Vec, +} + +struct Transfer { + peer: u64, + started: Instant, +} + +impl SnapshotTransfers { + pub fn sent(&mut self, message: &Message, now: Instant) { + if message.msg_type != MessageType::MsgSnapshot { + return; + } + if let Some(transfer) = self.pending.iter_mut().find(|transfer| transfer.peer == message.to) + { + transfer.started = now; + } else { + self.pending.push(Transfer { peer: message.to, started: now }); + } + } + + pub fn expire(&mut self, node: &mut RawNode, now: Instant) { + self.pending.retain(|transfer| { + if node.raft.state != StateRole::Leader || + node.raft + .prs() + .get(transfer.peer) + .is_none_or(|progress| progress.state != ProgressState::Snapshot) + { + return false; + } + if now.saturating_duration_since(transfer.started) < TRANSFER_TIMEOUT { + return true; + } + node.report_snapshot(transfer.peer, SnapshotStatus::Failure); + tracing::warn!(peer = transfer.peer, "Raft snapshot transfer timed out"); + false + }); + } +} diff --git a/crates/control/src/cluster/storage/compaction_tests.rs b/crates/control/src/cluster/storage/compaction_tests.rs new file mode 100644 index 00000000..7e54dc55 --- /dev/null +++ b/crates/control/src/cluster/storage/compaction_tests.rs @@ -0,0 +1,190 @@ +use std::{ + fs, + time::{Duration, Instant}, +}; + +use flux_disk::FailedOp; + +use super::{ + tests::{entry, event, identity, persisted, recovered, state}, + *, +}; + +fn snapshot(index: u64) -> Snapshot { + let mut snapshot = Snapshot { data: vec![7; 32].into(), ..Snapshot::default() }; + let metadata = snapshot.mut_metadata(); + metadata.index = index; + metadata.term = 2; + metadata.mut_conf_state().voters = vec![1, 2, 3]; + snapshot +} + +fn seeded(path: &Path) -> ClusterStorage { + let mut storage = ClusterStorage::create(path, identity()).unwrap(); + recovered(&mut storage); + let entries: Vec<_> = (1..=20).map(|index| entry(index, 2)).collect(); + storage.persist(1, &entries, Some(&state(2, 18))).unwrap(); + persisted(&mut storage, 1); + storage +} + +fn start_checkpoint(storage: &mut ClusterStorage) -> FileToken { + storage.checkpoint(None, &snapshot(18), &[entry(19, 2), entry(20, 2)], &state(2, 18)).unwrap(); + match &storage.state.phase { + Phase::Replacing(replacement) => replacement.file, + _ => panic!("not replacing"), + } +} + +fn poll_disk_until(storage: &mut ClusterStorage, matched: impl Fn(&DiskEvent<'_>) -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + let mut found = false; + while !found { + storage.disk.poll_with(|event| { + found |= matched(&event); + assert!( + storage.state.on_event(event).unwrap().is_none(), + "checkpoint acknowledged before directory sync" + ); + }); + assert!(Instant::now() < deadline); + std::thread::yield_now(); + } +} + +#[test] +fn compaction_shrinks_the_journal_and_preserves_the_uncommitted_suffix() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let original_len = fs::metadata(&path).unwrap().len(); + start_checkpoint(&mut storage); + assert!(!storage.is_ready()); + assert!(matches!(event(&mut storage).unwrap(), ClusterStorageEvent::Compacted { index: 18 })); + assert!(storage.is_ready()); + assert_eq!(storage.appended_bytes(), 0); + assert!(fs::metadata(&path).unwrap().len() < original_len / 2); + drop(storage); + let mut storage = ClusterStorage::open(&path, identity()).unwrap(); + let restored = recovered(&mut storage); + assert_eq!(restored.snapshot, snapshot(18)); + assert_eq!(restored.entries, [entry(19, 2), entry(20, 2)]); + assert_eq!(restored.hard_state, state(2, 18)); + storage.persist(1, &[entry(19, 3)], Some(&state(3, 19))).unwrap(); + persisted(&mut storage, 1); + drop(storage); + let restored = recovered(&mut ClusterStorage::open(&path, identity()).unwrap()); + assert_eq!(restored.snapshot, snapshot(18)); + assert_eq!(restored.entries, [entry(19, 3)]); + assert_eq!(restored.hard_state, state(3, 19)); +} + +#[test] +fn restart_before_rename_uses_the_old_journal_and_ignores_staging() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let old = fs::read(&path).unwrap(); + let file = start_checkpoint(&mut storage); + poll_disk_until( + &mut storage, + |event| matches!(event, DiskEvent::Synced { file: completed, .. } if *completed == file), + ); + assert_eq!(fs::read(&path).unwrap(), old); + drop(storage); + assert!(dir.path().join("raft.wal.next").exists()); + let mut storage = ClusterStorage::open(&path, identity()).unwrap(); + let restored = recovered(&mut storage); + assert!(restored.snapshot.is_empty()); + assert_eq!(restored.entries.len(), 20); + start_checkpoint(&mut storage); + assert!(matches!(event(&mut storage).unwrap(), ClusterStorageEvent::Compacted { index: 18 })); +} + +#[test] +fn rename_does_not_acknowledge_and_the_new_image_is_self_contained() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let file = start_checkpoint(&mut storage); + poll_disk_until( + &mut storage, + |event| matches!(event, DiskEvent::Synced { file: completed, .. } if *completed == file), + ); + storage.state.advance(&mut storage.disk).unwrap(); + poll_disk_until( + &mut storage, + |event| matches!(event, DiskEvent::Renamed { file: completed, .. } if *completed == file), + ); + assert!(!storage.is_ready()); + drop(storage); + let restored = recovered(&mut ClusterStorage::open(&path, identity()).unwrap()); + assert_eq!(restored.snapshot, snapshot(18)); + assert_eq!(restored.hard_state, state(2, 18)); + assert_eq!(restored.entries, [entry(19, 2), entry(20, 2)]); +} + +#[test] +fn checkpoint_failure_never_releases_a_completion() { + for op in [FailedOp::Write { offset: 0, len: 1 }, FailedOp::Sync, FailedOp::Rename] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let file = start_checkpoint(&mut storage); + assert!( + storage + .state + .on_event(DiskEvent::Failed { + file, + op, + operation_id: None, + error: io::Error::other("injected checkpoint failure"), + }) + .is_err() + ); + assert!(!storage.is_ready()); + assert!(storage.poll().is_err()); + } +} + +#[test] +fn directory_sync_failure_after_rename_does_not_acknowledge_the_checkpoint() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let file = start_checkpoint(&mut storage); + poll_disk_until( + &mut storage, + |event| matches!(event, DiskEvent::Synced { file: completed, .. } if *completed == file), + ); + storage.state.advance(&mut storage.disk).unwrap(); + poll_disk_until( + &mut storage, + |event| matches!(event, DiskEvent::Renamed { file: completed, .. } if *completed == file), + ); + storage.state.advance(&mut storage.disk).unwrap(); + assert!( + storage + .state + .on_event(DiskEvent::Failed { + file: storage.state.directory, + op: FailedOp::Sync, + operation_id: None, + error: io::Error::other("injected directory sync failure"), + }) + .is_err() + ); + assert!(storage.poll().is_err()); + assert!(!storage.is_ready()); +} + +#[test] +fn local_checkpoint_cannot_discard_durable_suffix_or_change_hard_state() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("raft.wal"); + let mut storage = seeded(&path); + let original = fs::read(&path).unwrap(); + assert!(storage.checkpoint(None, &snapshot(18), &[], &state(2, 18)).is_err()); + assert!(storage.checkpoint(None, &snapshot(20), &[], &state(2, 20)).is_err()); + assert_eq!(fs::read(&path).unwrap(), original); +} diff --git a/crates/control/src/cluster/storage/journal.rs b/crates/control/src/cluster/storage/journal.rs index 5bfaaaab..23cbf0c5 100644 --- a/crates/control/src/cluster/storage/journal.rs +++ b/crates/control/src/cluster/storage/journal.rs @@ -1,14 +1,17 @@ use std::io; use buffa::Message as _; -use raft::eraftpb::{Entry, HardState}; +use raft::eraftpb::{Entry, HardState, Snapshot}; use super::super::{ generated, - wire::{from_wire_entry, to_wire_entry}, + persistence::RecoveredStorage, + raft_storage::validate_snapshot, + wire::{from_wire_entry, from_wire_snapshot, to_wire_entry, to_wire_snapshot}, }; -const MAGIC: &[u8; 8] = b"SLVRAFT\x01"; +const MAGIC: &[u8; 8] = b"SLVRAFT\x02"; +const LEGACY_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; @@ -37,6 +40,14 @@ impl StorageIdentity { } pub(super) fn write(&self, output: &mut Vec) { + self.write_header(output, false); + } + + pub(super) fn write_checkpoint(&self, output: &mut Vec) { + self.write_header(output, true); + } + + fn write_header(&self, output: &mut Vec, checkpoint: bool) { let start = Frame::start(output); output.extend_from_slice(MAGIC); output.extend_from_slice(&self.node_id.to_le_bytes()); @@ -44,12 +55,18 @@ impl StorageIdentity { for voter in &self.voters { output.extend_from_slice(&voter.to_le_bytes()); } + output.push(u8::from(checkpoint)); Frame::finish(output, start); } - fn verify(&self, payload: &[u8]) -> io::Result<()> { + pub(super) fn validate_snapshot(&self, snapshot: &Snapshot) -> io::Result<()> { + validate_snapshot(snapshot, &self.voters) + } + + fn verify(&self, payload: &[u8]) -> io::Result { let mut cursor = Cursor(payload); - if cursor.take(MAGIC.len())? != MAGIC { + let magic = cursor.take(MAGIC.len())?; + if magic != MAGIC && magic != LEGACY_MAGIC { return Err(invalid_data("unsupported Raft journal format")); } if cursor.u64()? != self.node_id || cursor.u32()? as usize != self.voters.len() { @@ -60,19 +77,21 @@ impl StorageIdentity { return Err(invalid_data("Raft journal voters do not match configuration")); } } - cursor.finish() + let checkpoint = if magic == LEGACY_MAGIC { + false + } else { + match cursor.take(1)?[0] { + 0 => false, + 1 => true, + _ => return Err(invalid_data("invalid Raft journal header kind")), + } + }; + cursor.finish()?; + Ok(checkpoint) } } -#[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)] +#[derive(Debug, Clone, Default, PartialEq)] pub(super) struct LogState { pub hard_state: HardState, pub last_index: u64, @@ -80,6 +99,21 @@ pub(super) struct LogState { } impl LogState { + pub fn from_snapshot(snapshot: &Snapshot, hard_state: &HardState) -> io::Result { + let metadata = snapshot.get_metadata(); + if metadata.index == 0 || + metadata.index == u64::MAX || + metadata.term == 0 || + metadata.term > hard_state.term || + metadata.index > hard_state.commit + { + return Err(invalid_data("Raft snapshot is inconsistent with hard state")); + } + let mut state = hard_state.clone(); + state.commit = metadata.index; + Ok(Self { hard_state: state, last_index: metadata.index, last_term: metadata.term }) + } + 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 || @@ -130,9 +164,15 @@ impl LogState { pub(super) struct Record<'a> { entries: &'a [Entry], hard_state: Option<&'a HardState>, + snapshot: Option<&'a Snapshot>, payload_len: usize, } +struct DecodedRecord { + hard_state: Option, + snapshot: Option, +} + 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 }; @@ -146,18 +186,44 @@ impl<'a> Record<'a> { return Err(invalid_data("Raft journal record exceeds 64 MiB")); } } - Ok(Self { entries, hard_state, payload_len }) + Ok(Self { entries, hard_state, snapshot: None, payload_len }) + } + + pub fn checkpoint( + snapshot: &'a Snapshot, + entries: &'a [Entry], + hard_state: &'a HardState, + ) -> io::Result { + let mut record = Self::new(entries, Some(hard_state))?; + if snapshot.data.len() > MAX_RECORD_BYTES { + return Err(invalid_data("Raft snapshot exceeds journal limit")); + } + record.payload_len += 4 + to_wire_snapshot(snapshot.clone()).compute_size() as usize; + if record.payload_len > MAX_RECORD_BYTES { + return Err(invalid_data("Raft checkpoint exceeds journal limit")); + } + record.snapshot = Some(snapshot); + Ok(record) + } + + pub fn len(&self) -> u64 { + (FRAME_HEADER_LEN + self.payload_len) as u64 } 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())); + output.push(u8::from(self.hard_state.is_some()) | (u8::from(self.snapshot.is_some()) << 1)); 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()); } + if let Some(snapshot) = self.snapshot { + let snapshot = to_wire_snapshot(snapshot.clone()); + output.extend_from_slice(&snapshot.compute_size().to_le_bytes()); + snapshot.write_to(output); + } output.extend_from_slice(&(self.entries.len() as u32).to_le_bytes()); for entry in self.entries { let entry = to_wire_entry(entry.clone()); @@ -167,11 +233,12 @@ impl<'a> Record<'a> { Frame::finish(output, start); } - fn decode(payload: &[u8], entries: &mut Vec) -> io::Result> { + fn decode(payload: &[u8], entries: &mut Vec) -> io::Result { let mut cursor = Cursor(payload); - let state = match cursor.take(1)?[0] { + let flags = cursor.take(1)?[0]; + let hard_state = match flags { 0 => None, - 1 => Some(HardState { + 1 | 3 => Some(HardState { term: cursor.u64()?, vote: cursor.u64()?, commit: cursor.u64()?, @@ -179,6 +246,14 @@ impl<'a> Record<'a> { }), _ => return Err(invalid_data("invalid Raft hard state tag")), }; + let snapshot = if flags & 2 != 0 { + let len = cursor.u32()? as usize; + let snapshot = generated::Snapshot::decode_from_slice(cursor.take(len)?) + .map_err(|_| invalid_data("invalid Raft snapshot protobuf"))?; + Some(from_wire_snapshot(snapshot).map_err(|_| invalid_data("invalid Raft snapshot"))?) + } else { + None + }; let count = cursor.u32()? as usize; if count > cursor.0.len() / 4 { return Err(invalid_data("invalid Raft journal entry count")); @@ -192,7 +267,7 @@ impl<'a> Record<'a> { .push(from_wire_entry(entry).map_err(|_| invalid_data("invalid Raft entry type"))?); } cursor.finish()?; - Ok(state) + Ok(DecodedRecord { hard_state, snapshot }) } } @@ -200,10 +275,12 @@ pub(super) struct JournalReplay { identity: StorageIdentity, buffer: Vec, header_read: bool, + checkpoint_required: bool, record_entries: Vec, pub recovered: RecoveredStorage, pub log: LogState, pub valid_len: u64, + pub appended_bytes: u64, } impl JournalReplay { @@ -212,10 +289,12 @@ impl JournalReplay { identity, buffer: Vec::with_capacity(READ_CHUNK_BYTES), header_read: false, + checkpoint_required: false, record_entries: Vec::new(), recovered: RecoveredStorage::default(), log: LogState::default(), valid_len: 0, + appended_bytes: 0, } } @@ -224,15 +303,55 @@ impl JournalReplay { let mut consumed = 0; while let Some(frame) = Frame::read(&self.buffer[consumed..])? { if !self.header_read { - self.identity.verify(frame.payload)?; + self.checkpoint_required = 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())?; + let record = Record::decode(frame.payload, &mut self.record_entries)?; + if record.snapshot.is_none() { + self.appended_bytes += (FRAME_HEADER_LEN + frame.payload.len()) as u64; + } + if self.checkpoint_required && + self.recovered.snapshot.is_empty() && + record.snapshot.is_none() + { + return Err(invalid_data("Raft journal is missing its checkpoint")); + } + if let Some(snapshot) = record.snapshot { + if !self.checkpoint_required || + self.log.last_index != 0 || + self.log.hard_state != HardState::default() + { + return Err(invalid_data("Raft checkpoint must precede journal appends")); + } + self.identity.validate_snapshot(&snapshot)?; + self.log = LogState::from_snapshot( + &snapshot, + record + .hard_state + .as_ref() + .ok_or_else(|| invalid_data("checkpoint has no hard state"))?, + )?; + self.recovered.snapshot = snapshot; + } + let next = self.log.next(&self.record_entries, record.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 { + let base = self.recovered.snapshot.get_metadata().index + 1; + let retain = usize::try_from( + first + .index + .checked_sub(base) + .ok_or_else(|| invalid_data("append precedes Raft snapshot"))?, + ) + .map_err(|_| invalid_data("Raft index exceeds address space"))?; + if retain > self.recovered.entries.len() { + return Err(invalid_data("Raft append leaves a gap after snapshot")); + } + let previous_term = if retain == 0 { + self.recovered.snapshot.get_metadata().term + } else { + self.recovered.entries[retain - 1].term + }; + if previous_term > first.term { return Err(invalid_data("Raft entry terms regressed across an append")); } self.recovered.entries.truncate(retain); @@ -252,6 +371,9 @@ impl JournalReplay { if !self.header_read { return Err(invalid_data("Raft journal has no complete identity header")); } + if self.checkpoint_required && self.recovered.snapshot.is_empty() { + return Err(invalid_data("Raft journal has no complete checkpoint")); + } Ok(!self.buffer.is_empty()) } } diff --git a/crates/control/src/cluster/storage/journal/tests.rs b/crates/control/src/cluster/storage/journal/tests.rs index 7b934089..87e331d7 100644 --- a/crates/control/src/cluster/storage/journal/tests.rs +++ b/crates/control/src/cluster/storage/journal/tests.rs @@ -24,6 +24,58 @@ fn header() -> Vec { bytes } +fn snapshot(index: u64, term: u64) -> Snapshot { + let mut snapshot = Snapshot { data: vec![7; 100].into(), ..Snapshot::default() }; + let metadata = snapshot.mut_metadata(); + metadata.index = index; + metadata.term = term; + metadata.mut_conf_state().voters = vec![1, 2, 3]; + snapshot +} + +#[test] +fn checkpoint_replays_a_compacted_base_and_replaces_only_the_uncommitted_suffix() { + let snapshot = snapshot(10, 2); + let mut bytes = Vec::new(); + identity().write_checkpoint(&mut bytes); + Record::checkpoint(&snapshot, &[entry(11, 2), entry(12, 2)], &state(2, 1, 11)) + .unwrap() + .write(&mut bytes); + append(&mut bytes, &[entry(12, 3)], Some(&state(3, 1, 12))); + let recovered = replay(&bytes).unwrap().recovered; + assert_eq!(recovered.snapshot, snapshot); + assert_eq!(recovered.entries, vec![entry(11, 2), entry(12, 3)]); + assert_eq!(recovered.hard_state, state(3, 1, 12)); +} + +#[test] +fn an_incomplete_checkpoint_is_never_recovered_as_an_empty_voter() { + let mut bytes = Vec::new(); + identity().write_checkpoint(&mut bytes); + Record::checkpoint(&snapshot(10, 2), &[], &state(2, 1, 10)).unwrap().write(&mut bytes); + for length in 0..bytes.len() { + assert!(replay(&bytes[..length]).is_err(), "accepted truncated checkpoint at {length}"); + } + assert!(replay(&bytes).is_ok()); + let valid = bytes.len(); + append(&mut bytes, &[entry(11, 2)], Some(&state(2, 1, 10))); + for length in valid..bytes.len() { + let recovered = replay(&bytes[..length]).unwrap().recovered; + assert_eq!(recovered.snapshot.get_metadata().index, 10); + assert!(recovered.entries.is_empty()); + } +} + +#[test] +fn legacy_journal_headers_remain_readable() { + let mut bytes = header(); + bytes[FRAME_HEADER_LEN..FRAME_HEADER_LEN + MAGIC.len()].copy_from_slice(LEGACY_MAGIC); + bytes.pop(); + Frame::finish(&mut bytes, 0); + append(&mut bytes, &[entry(1, 1)], Some(&state(1, 1, 1))); + assert_eq!(replay(&bytes).unwrap().recovered.entries, vec![entry(1, 1)]); +} + fn append(bytes: &mut Vec, entries: &[Entry], hard_state: Option<&HardState>) { Record::new(entries, hard_state).unwrap().write(bytes); } @@ -159,7 +211,7 @@ fn checksummed_records_cannot_overwrite_commits_or_introduce_gaps() { #[test] fn unknown_format_and_malformed_batches_fail_closed() { let mut bytes = header(); - bytes[FRAME_HEADER_LEN + MAGIC.len() - 1] = 2; + bytes[FRAME_HEADER_LEN + MAGIC.len() - 1] = 0xff; Frame::finish(&mut bytes, 0); assert!(replay(&bytes).is_err()); diff --git a/crates/control/src/cluster/storage/mod.rs b/crates/control/src/cluster/storage/mod.rs index 0f2d3a65..ac031508 100644 --- a/crates/control/src/cluster/storage/mod.rs +++ b/crates/control/src/cluster/storage/mod.rs @@ -1,16 +1,24 @@ mod journal; +mod replacement; -use std::{io, path::Path}; +use std::{ + io, mem, + path::{Path, PathBuf}, +}; use flux_disk::{DiskConfig, DiskEvent, DiskIo, FileToken, OpenOptions, OperationId}; +pub use journal::StorageIdentity; use journal::{JournalReplay, LogState, READ_CHUNK_BYTES, Record}; -pub use journal::{RecoveredStorage, StorageIdentity}; -use raft::eraftpb::{Entry, HardState}; +use raft::eraftpb::{Entry, HardState, Snapshot}; +use replacement::Replacement; + +use super::persistence::RecoveredStorage; #[derive(Debug)] pub enum ClusterStorageEvent { Recovered(RecoveredStorage), Persisted { ready_number: u64 }, + Compacted { index: u64 }, } /// Keeps one persistence batch in flight. Completion always means durable, not @@ -41,13 +49,17 @@ impl ClusterStorage { 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) + (Phase::Creating(identity.clone()), None) } else { - (Phase::ReadNext { offset: 0 }, Some(JournalReplay::new(identity))) + (Phase::ReadNext { offset: 0 }, Some(JournalReplay::new(identity.clone()))) }; let mut state = StorageState { file, directory, + path: path.to_path_buf(), + identity, + retired_file: None, + appended_bytes: 0, phase, replay, recovered: None, @@ -62,6 +74,73 @@ impl ClusterStorage { matches!(self.state.phase, Phase::Ready) } + pub fn appended_bytes(&self) -> u64 { + self.state.appended_bytes + } + + pub fn checkpoint( + &mut self, + ready_number: Option, + snapshot: &Snapshot, + entries: &[Entry], + hard_state: &HardState, + ) -> io::Result<()> { + if !self.is_ready() { + return Err(io::Error::new(io::ErrorKind::WouldBlock, "Raft storage is busy")); + } + if ready_number.is_some_and(|number| number <= self.state.last_ready_number) { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "Raft Ready numbers must increase", + )); + } + let previous = &self.state.log.hard_state; + if hard_state.term < previous.term || + hard_state.commit < previous.commit || + (hard_state.term == previous.term && + previous.vote != 0 && + hard_state.vote != previous.vote) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "checkpoint hard state regressed", + )); + } + let next = + LogState::from_snapshot(snapshot, hard_state)?.next(entries, Some(hard_state))?; + if ready_number.is_none() && + (snapshot.get_metadata().index > previous.commit || next != self.state.log) + { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "local checkpoint must preserve the durable log suffix and hard state", + )); + } + if ready_number.is_some() && snapshot.get_metadata().index < previous.commit { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "incoming snapshot precedes durable commit", + )); + } + let replacement = Replacement::new( + &mut self.disk, + &self.state.path, + &self.state.identity, + snapshot, + entries, + hard_state, + ready_number, + ); + match replacement { + Ok(replacement) => self.state.phase = Phase::Replacing(replacement), + Err(error) => { + self.state.phase = Phase::Failed; + return Err(error); + } + } + Ok(()) + } + /// 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. @@ -84,6 +163,7 @@ impl ClusterStorage { } let next = self.state.log.next(entries, hard_state)?; let record = Record::new(entries, hard_state)?; + self.state.appended_bytes = self.state.appended_bytes.saturating_add(record.len()); if !self.disk.write_with(self.state.file, |output| record.write(output)) { self.state.phase = Phase::Failed; return Err(failed()); @@ -130,12 +210,17 @@ enum Phase { SyncingDirectory { sync: OperationId }, Ready, SyncingBatch { sync: OperationId, ready_number: u64, next: LogState }, + Replacing(Replacement), Failed, } struct StorageState { file: FileToken, directory: FileToken, + path: PathBuf, + identity: StorageIdentity, + retired_file: Option, + appended_bytes: u64, phase: Phase, replay: Option, recovered: Option, @@ -145,7 +230,15 @@ struct StorageState { impl StorageState { fn advance(&mut self, disk: &mut DiskIo) -> io::Result<()> { - match &self.phase { + if let Some(file) = self.retired_file.take() && + !disk.close(file) + { + return Err(failed()); + } + match &mut self.phase { + Phase::Replacing(replacement) => { + replacement.advance(disk, &self.path, self.directory)? + } Phase::Creating(identity) => { if !disk.write_with(self.file, |output| identity.write(output)) { return Err(failed()); @@ -186,11 +279,29 @@ impl StorageState { 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}"))); + if let DiskEvent::Failed { op, error, .. } = event { + self.phase = Phase::Failed; + return Err(io::Error::new(error.kind(), format!("Raft journal {op:?}: {error}"))); + } + if let Phase::Replacing(replacement) = &mut self.phase { + if !replacement.on_event(event, self.directory)? { + return Ok(None); } + let Phase::Replacing(replacement) = mem::replace(&mut self.phase, Phase::Ready) else { + return Err(failed()); + }; + self.retired_file = Some(mem::replace(&mut self.file, replacement.file)); + self.log = replacement.log; + self.appended_bytes = 0; + return Ok(Some(match replacement.ready_number { + Some(ready_number) => { + self.last_ready_number = ready_number; + ClusterStorageEvent::Persisted { ready_number } + } + None => ClusterStorageEvent::Compacted { index: replacement.index }, + })); + } + match event { DiskEvent::Read { file, offset, payload, eof } => { if file != self.file || !matches!(self.phase, Phase::Reading { offset: expected } if expected == offset) @@ -203,6 +314,7 @@ impl StorageState { let truncate = replay.finish()?; let replay = self.replay.take().ok_or_else(failed)?; self.log = replay.log; + self.appended_bytes = replay.appended_bytes; self.recovered = Some(replay.recovered); self.phase = Phase::Repair { truncate, valid_len: replay.valid_len }; } else { @@ -247,5 +359,7 @@ fn failed() -> io::Error { ) } +#[cfg(test)] +mod compaction_tests; #[cfg(test)] mod tests; diff --git a/crates/control/src/cluster/storage/replacement.rs b/crates/control/src/cluster/storage/replacement.rs new file mode 100644 index 00000000..27f7481c --- /dev/null +++ b/crates/control/src/cluster/storage/replacement.rs @@ -0,0 +1,105 @@ +use std::{io, path::Path}; + +use flux_disk::{DiskEvent, DiskIo, FileToken, OpenOptions, OperationId}; +use raft::eraftpb::{Entry, HardState, Snapshot}; + +use super::{ + failed, + journal::{LogState, Record, StorageIdentity}, +}; + +pub(super) struct Replacement { + pub file: FileToken, + pub log: LogState, + pub ready_number: Option, + pub index: u64, + phase: Phase, +} + +enum Phase { + SyncFile(OperationId), + Rename, + Renaming(OperationId), + SyncDirectory, + SyncingDirectory(OperationId), +} + +impl Replacement { + pub fn new( + disk: &mut DiskIo, + path: &Path, + identity: &StorageIdentity, + snapshot: &Snapshot, + entries: &[Entry], + hard_state: &HardState, + ready_number: Option, + ) -> io::Result { + identity.validate_snapshot(snapshot)?; + let log = LogState::from_snapshot(snapshot, hard_state)?.next(entries, Some(hard_state))?; + let record = Record::checkpoint(snapshot, entries, hard_state)?; + let mut staging = path.as_os_str().to_os_string(); + staging.push(".next"); + // Only the canonical path is authoritative. An interrupted replacement is + // disposable. + let file = disk.open( + Path::new(&staging), + OpenOptions::new().read(true).write(true).create(true).truncate(true), + )?; + if !disk.write_with(file, |output| { + identity.write_checkpoint(output); + record.write(output); + }) { + return Err(failed()); + } + let sync = disk.sync_all(file).ok_or_else(failed)?; + Ok(Self { + file, + log, + ready_number, + index: snapshot.get_metadata().index, + phase: Phase::SyncFile(sync), + }) + } + + pub fn advance( + &mut self, + disk: &mut DiskIo, + path: &Path, + directory: FileToken, + ) -> io::Result<()> { + match self.phase { + Phase::Rename => { + self.phase = Phase::Renaming(disk.rename(self.file, path)?.ok_or_else(failed)?); + } + Phase::SyncDirectory => { + self.phase = Phase::SyncingDirectory(disk.sync_all(directory).ok_or_else(failed)?); + } + _ => {} + } + Ok(()) + } + + pub fn on_event(&mut self, event: DiskEvent<'_>, directory: FileToken) -> io::Result { + match event { + DiskEvent::Synced { file, operation_id } => match self.phase { + Phase::SyncFile(expected) if file == self.file && operation_id == expected => { + self.phase = Phase::Rename; + } + Phase::SyncingDirectory(expected) + if file == directory && operation_id == expected => + { + return Ok(true) + } + _ => return Err(io::Error::other("unexpected checkpoint sync completion")), + }, + DiskEvent::Renamed { file, operation_id } => match self.phase { + Phase::Renaming(expected) if file == self.file && operation_id == expected => { + self.phase = Phase::SyncDirectory; + } + _ => return Err(io::Error::other("unexpected checkpoint rename completion")), + }, + _ => {} + } + Ok(false) + } +} diff --git a/crates/control/src/cluster/storage/tests.rs b/crates/control/src/cluster/storage/tests.rs index f93b700e..ba4a68e4 100644 --- a/crates/control/src/cluster/storage/tests.rs +++ b/crates/control/src/cluster/storage/tests.rs @@ -12,19 +12,19 @@ use crate::cluster::{ command::ReplicatedCommand, }; -fn identity() -> StorageIdentity { +pub(super) fn identity() -> StorageIdentity { StorageIdentity::new(1, vec![1, 2, 3]).unwrap() } -fn entry(index: u64, term: u64) -> Entry { +pub(super) 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 { +pub(super) fn state(term: u64, commit: u64) -> HardState { HardState { term, vote: 1, commit, ..HardState::default() } } -fn event(storage: &mut ClusterStorage) -> io::Result { +pub(super) fn event(storage: &mut ClusterStorage) -> io::Result { let deadline = Instant::now() + Duration::from_secs(5); loop { if let Some(event) = storage.poll()? { @@ -35,14 +35,14 @@ fn event(storage: &mut ClusterStorage) -> io::Result { } } -fn recovered(storage: &mut ClusterStorage) -> RecoveredStorage { +pub(super) 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) { +pub(super) fn persisted(storage: &mut ClusterStorage, number: u64) { assert!( matches!(event(storage).unwrap(), ClusterStorageEvent::Persisted { ready_number } if ready_number == number) ); diff --git a/crates/control/src/cluster/wire.rs b/crates/control/src/cluster/wire.rs index db7cde52..b819282c 100644 --- a/crates/control/src/cluster/wire.rs +++ b/crates/control/src/cluster/wire.rs @@ -2,7 +2,7 @@ use buffa::Message as _; use raft::eraftpb::{ ConfState, Entry, EntryType, Message, MessageType, Snapshot, SnapshotMetadata, }; -use silver_common::{Error, TCacheProducer, TCacheRead, TProducer}; +use silver_common::{Error, MAX_CLUSTER_MESSAGE_BYTES, TCacheProducer, TCacheRead, TProducer}; use super::generated as wire; @@ -14,6 +14,9 @@ pub(crate) fn encode_message( ) -> Result { let message = to_wire_message(message); let len = message.compute_size() as usize; + if len > MAX_CLUSTER_MESSAGE_BYTES { + return Err(Error::BufferTooSmall); + } let mut reservation = producer.reserve(len, true).ok_or(Error::BufferTooSmall)?; let output = producer.reservation_buffer(&mut reservation)?; let mut cursor: &mut [u8] = &mut output[..len]; @@ -104,7 +107,7 @@ pub(super) fn to_wire_entry(entry: Entry) -> wire::Entry { } } -fn to_wire_snapshot(mut snapshot: Snapshot) -> wire::Snapshot { +pub(super) fn to_wire_snapshot(mut snapshot: Snapshot) -> wire::Snapshot { let metadata = if snapshot.has_metadata() { buffa::MessageField::some(to_wire_snapshot_metadata(snapshot.take_metadata())) } else { @@ -176,7 +179,7 @@ pub(super) fn from_wire_entry(entry: wire::Entry) -> Result Ok(decoded) } -fn from_wire_snapshot(snapshot: wire::Snapshot) -> Result { +pub(super) fn from_wire_snapshot(snapshot: wire::Snapshot) -> Result { let metadata = snapshot.metadata.into_option().map(from_wire_snapshot_metadata).transpose()?; let mut decoded = Snapshot::default(); decoded.set_data(snapshot.data); diff --git a/crates/control/src/tile/attestation_cluster.rs b/crates/control/src/tile/attestation_cluster.rs index 66b823a2..d09b82e6 100644 --- a/crates/control/src/tile/attestation_cluster.rs +++ b/crates/control/src/tile/attestation_cluster.rs @@ -81,7 +81,7 @@ impl AttestationClusterHandler { let initialized = cluster.set_startup_wall_slot(wall_slot); debug_assert!(initialized, "handler and cluster startup floors latch together"); } - tracing::info!(wall_slot, "local attestation admission enabled"); + tracing::info!(wall_slot, "local attestation startup floor latched"); } /// Consume inbound Raft messages and pump all work currently ready in the @@ -189,6 +189,9 @@ impl AttestationClusterHandler { ); return; }; + if !cluster.is_ready() { + return; + } let message = match decode_message(bytes) { Ok(message) => message, Err(error) => { @@ -301,6 +304,13 @@ impl AttestationClusterHandler { }); if let Err(error) = result { tracing::error!(?error, "attestation cluster spin failed"); + for (_, attestation) in self.pending_attestations.drain() { + produce_response( + producers, + attestation.request_id, + Err(LocalGossipFailure::Internal), + ); + } } } } @@ -310,7 +320,10 @@ mod tests { use silver_common::{TCache, TCacheId}; use super::*; - use crate::tile::local_gossip::{VALIDATION_TIMEOUT, tests::Harness}; + use crate::{ + cluster::ClusterStorageConfig, + tile::local_gossip::{VALIDATION_TIMEOUT, tests::Harness}, + }; fn handler(now: Instant) -> AttestationClusterHandler { AttestationClusterHandler::new( @@ -506,6 +519,67 @@ mod tests { handler.on_status(20, 20); assert_eq!(handler.admission.validate(11, 20), Ok(()), "floor must not relatch"); } + + #[test] + fn failed_cluster_rejects_pending_and_new_requests_without_standalone_fallback() { + let now = Instant::now(); + let mut handler = handler(now); + let mut harness = Harness::new(); + handler.cluster = Some( + AttestationCluster::in_memory( + AttestationClusterConfig::new( + 1, + vec![1], + ClusterStorageConfig::Create("unused-test-journal".into()), + ), + now, + ) + .unwrap(), + ); + handler.on_status(10, 10); + handler.on_status(11, 11); + handler.cluster.as_mut().unwrap().campaign().unwrap(); + handler.drive( + now, + &mut harness.validation, + &mut harness.gossip, + &mut harness.adapter.producers, + ); + + let mut ssz = [0; SINGLE_ATT_SIZE]; + ssz[16..24].copy_from_slice(&11u64.to_le_bytes()); + let attestation = PendingAttestation::new(1, 0, ssz); + handler.on_local_attestation( + attestation, + now, + &mut harness.validation, + &mut harness.gossip, + &mut harness.adapter.producers, + ); + assert_eq!(handler.pending_attestations.len(), 1); + assert!(harness.responses().is_empty()); + handler.cluster.as_mut().unwrap().fail_persistence(); + handler.drive( + now, + &mut harness.validation, + &mut harness.gossip, + &mut harness.adapter.producers, + ); + assert_eq!(harness.responses(), [(1, Err(LocalGossipFailure::Internal))]); + assert!(handler.pending_attestations.is_empty()); + assert!(handler.cluster.as_ref().unwrap().is_failed()); + + handler.on_local_attestation( + PendingAttestation { request_id: 2, ..attestation }, + now, + &mut harness.validation, + &mut harness.gossip, + &mut harness.adapter.producers, + ); + assert_eq!(harness.responses(), [(2, Err(LocalGossipFailure::Internal))]); + assert!(harness.gossip.pop_event().is_none()); + assert!(harness.validation.is_empty()); + } } fn decision_response(decision: &AttestationDecision) -> LocalGossipResult { @@ -527,6 +601,8 @@ fn lock_response(result: LockResult) -> LocalGossipResult { fn proposal_failure(error: &ProposeError) -> LocalGossipFailure { match error { ProposeError::Admission(error) => admission_failure(*error), + ProposeError::NotReady | + ProposeError::Failed | ProposeError::SequenceExhausted | ProposeError::DeadlineOverflow | ProposeError::Raft(_) => LocalGossipFailure::Internal, diff --git a/crates/network/src/p2p/quic/peer.rs b/crates/network/src/p2p/quic/peer.rs index 5f30d836..f00c0144 100644 --- a/crates/network/src/p2p/quic/peer.rs +++ b/crates/network/src/p2p/quic/peer.rs @@ -1429,13 +1429,17 @@ mod tests { impl PeerHarness { fn new() -> Self { + Self::with_cluster_size(TCACHE_BYTES) + } + + fn with_cluster_size(cluster_bytes: usize) -> Self { let gossip_in_p = TCache::producer(TCacheId::NetworkIngress, TCACHE_BYTES); let gossip_in_c = gossip_in_p.cache_ref().consumer("peer_gossip_in").unwrap(); let gossip_out_p = TCache::producer(TCacheId::ControlGossip, TCACHE_BYTES); let rpc_in_p = TCache::producer(TCacheId::NetworkProcessing, TCACHE_BYTES); let rpc_out_producer = TCache::producer(TCacheId::StorageDelivery, TCACHE_BYTES); - let cluster_in = TCache::producer(TCacheId::ClusterInbound, TCACHE_BYTES); - let cluster_out_producer = TCache::producer(TCacheId::ClusterOutbound, TCACHE_BYTES); + let cluster_in = TCache::producer(TCacheId::ClusterInbound, cluster_bytes); + let cluster_out_producer = TCache::producer(TCacheId::ClusterOutbound, cluster_bytes); let columns = TCache::producer(TCacheId::ControlSlot, 1 << 16); let tcaches = TCacheTable::from_iter( [&gossip_out_p, &rpc_out_producer, &cluster_out_producer, &columns] @@ -1831,6 +1835,52 @@ mod tests { assert!(!pair.client_peer.streams.contains_key(&stream)); } + #[test] + fn cluster_frames_above_u16_size_round_trip_and_preserve_the_next_frame() { + let mut client_h = PeerHarness::with_cluster_size(1 << 20); + let mut server_h = PeerHarness::with_cluster_size(1 << 20); + let mut pair = PeerPair::new(); + let client_key = Keypair::from_secret(&[2; 32]).unwrap(); + let server_key = Keypair::from_secret(&[1; 32]).unwrap(); + let mut client_nodes = + ClusterNodes::new(HashMap::from([(2, Enr::empty(server_key.secret_key()).unwrap())])); + client_nodes.connected(pair.client_peer.id()); + client_h.context.cluster_nodes = Some(client_nodes); + let mut server_nodes = + ClusterNodes::new(HashMap::from([(1, Enr::empty(client_key.secret_key()).unwrap())])); + server_nodes.connected(pair.server_peer.id()); + server_h.context.cluster_nodes = Some(server_nodes); + + let body = vec![0x79; 128 * 1024]; + assert!(matches!(client_h.send_cluster(&body, &mut pair.client_peer), SendResult::Ok)); + assert!(matches!(client_h.send_cluster(b"next", &mut pair.client_peer), SendResult::Ok)); + let now = Instant::now(); + let mut received = Vec::new(); + for tick in 0..500 { + pair.step( + now + Duration::from_millis(tick * 2), + &mut client_h, + &mut server_h, + &mut |_| {}, + &mut |event| { + if let NetEvent::Cluster { raft_id, msg, .. } = event { + assert_eq!(raft_id, 1); + received.push(msg); + } + }, + ); + if received.len() == 2 { + break; + } + } + assert_eq!(received.len(), 2); + let producer = &server_h.context.cluster_inbound_producer; + assert_eq!(producer.read_buffer(received[0]).unwrap(), body); + assert_eq!(producer.read_buffer(received[1]).unwrap(), b"next"); + assert!(!pair.client_peer.is_closed()); + assert!(!pair.server_peer.is_closed()); + } + #[test] fn stopped_cluster_stream_reopens_and_shutdown_clears_it() { let mut client_h = PeerHarness::new(); diff --git a/crates/network/src/p2p/streams/cluster_in.rs b/crates/network/src/p2p/streams/cluster_in.rs index 420c5c88..b1da8ef5 100644 --- a/crates/network/src/p2p/streams/cluster_in.rs +++ b/crates/network/src/p2p/streams/cluster_in.rs @@ -1,6 +1,8 @@ use std::time::{Duration, Instant}; -use silver_common::{P2pStreamId, TCacheProducer, TProducer, TReservation}; +use silver_common::{ + MAX_CLUSTER_MESSAGE_BYTES, P2pStreamId, TCacheProducer, TProducer, TReservation, +}; use crate::{ NetEvent, @@ -11,10 +13,10 @@ pub(crate) const BODY_STALL_TIMEOUT: Duration = Duration::from_secs(2); #[derive(Debug)] pub(crate) enum ClusterRead { - /// Reading 2-byte length prefix. + /// Reading 4-byte length prefix. ReadingLength { raft_id: u64, - buf: [u8; 2], + buf: [u8; 4], read: usize, }, /// Have read length but buffer needs to be allocated. @@ -40,7 +42,7 @@ enum Spin { impl ClusterRead { pub(crate) fn new(raft_id: u64) -> Self { - Self::ReadingLength { raft_id, buf: [0u8; 2], read: 0 } + Self::ReadingLength { raft_id, buf: [0u8; 4], read: 0 } } pub(crate) fn spin( @@ -95,10 +97,15 @@ impl ClusterRead { } if read == buf.len() { - let length = u16::from_le_bytes(buf) as usize; + let length = u32::from_le_bytes(buf) as usize; if length == 0 { return Err(StreamError::ClusterFrameZeroSize); } + if length > MAX_CLUSTER_MESSAGE_BYTES || + length > tcache.cache_ref().max_payload_len() + { + return Err(StreamError::ClusterFrameTooLarge); + } return Ok(Spin::Next(Self::AllocBody { raft_id, length, fail_count: 0 })); } @@ -133,7 +140,7 @@ impl ClusterRead { assert!(reservation.is_committed()); emit(NetEvent::Cluster { stream: *p2p_id, raft_id, msg: reservation.read() }); // Continue into the next frame. - return Ok(Spin::Next(Self::ReadingLength { raft_id, buf: [0u8; 2], read: 0 })); + return Ok(Spin::Next(Self::ReadingLength { raft_id, buf: [0u8; 4], read: 0 })); } Ok(Spin::Ok(Self::ReadingBody { raft_id, reservation, remaining, last_read })) } diff --git a/crates/network/src/p2p/streams/cluster_out.rs b/crates/network/src/p2p/streams/cluster_out.rs index a1af5762..8c4cb520 100644 --- a/crates/network/src/p2p/streams/cluster_out.rs +++ b/crates/network/src/p2p/streams/cluster_out.rs @@ -1,4 +1,4 @@ -use silver_common::{P2pStreamId, TRead}; +use silver_common::{MAX_CLUSTER_MESSAGE_BYTES, P2pStreamId, TRead}; use super::StreamIo; use crate::p2p::{quic::Leased, streams::StreamError}; @@ -7,7 +7,7 @@ use crate::p2p::{quic::Leased, streams::StreamError}; pub enum ClusterWrite { Idle, WritingLength { - buffer: [u8; 2], + buffer: [u8; 4], written: usize, message: Leased, }, @@ -49,10 +49,10 @@ impl ClusterWrite { Self::Idle => match io.cluster_next() { Some(message) => { let len = message.len()?; - if len > u16::MAX as usize { + if len > MAX_CLUSTER_MESSAGE_BYTES { return Err(StreamError::ClusterFrameTooLarge); } - let len = len as u16; + let len = len as u32; Ok(Spin::Next(Self::WritingLength { buffer: len.to_le_bytes(), written: 0,