diff --git a/Cargo.lock b/Cargo.lock index d56e2063..cf3a784b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2151,7 +2151,7 @@ checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" name = "magic-root-interface" version = "0.1.0" dependencies = [ - "solana-account 4.3.1", + "solana-account", "solana-instruction", "solana-pubkey", "wincode", @@ -2163,7 +2163,7 @@ version = "0.1.0" dependencies = [ "magic-root-interface", "magicblock-engine-nucleus", - "solana-account 4.3.1", + "solana-account", "solana-instruction", "solana-instruction-error", "solana-program-runtime", @@ -2192,13 +2192,50 @@ dependencies = [ "memmap2", "parking_lot", "scc", - "solana-account 4.3.1", + "solana-account", "solana-pubkey", "thiserror 2.0.19", "tracing", "twox-hash", ] +[[package]] +name = "magicblock-engine" +version = "0.1.0" +dependencies = [ + "agave-transaction-view", + "derive_more", + "magic-root-interface", + "magic-root-program", + "magicblock-engine", + "magicblock-engine-nucleus", + "magicblock-keeper", + "magicblock-ledger", + "magicblock-processor", + "num_cpus", + "oneshot", + "solana-account", + "solana-compute-budget-program", + "solana-instruction", + "solana-instruction-error", + "solana-keypair", + "solana-message", + "solana-packet", + "solana-program-runtime", + "solana-pubkey", + "solana-sdk-ids", + "solana-signer", + "solana-system-interface", + "solana-system-program", + "solana-sysvar", + "solana-transaction", + "thiserror 2.0.19", + "tokio", + "tracing", + "v42-calculator-interface", + "wincode", +] + [[package]] name = "magicblock-engine-nucleus" version = "0.1.0" @@ -2249,7 +2286,7 @@ dependencies = [ "scc", "serde", "smallvec", - "solana-account 4.3.1", + "solana-account", "solana-feature-gate-interface", "solana-hash", "solana-instruction", @@ -2281,6 +2318,7 @@ dependencies = [ "flume", "heed", "magicblock-engine-nucleus", + "magicblock-ledger", "memmap2", "num_cpus", "oneshot", @@ -2310,7 +2348,7 @@ dependencies = [ "magicblock-engine-nucleus", "magicblock-keeper", "oneshot", - "solana-account 4.3.1", + "solana-account", "solana-compute-budget-instruction", "solana-hash", "solana-instruction", @@ -2331,6 +2369,29 @@ dependencies = [ "v42-calculator-interface", ] +[[package]] +name = "magicblock-replicator" +version = "0.1.0" +dependencies = [ + "derive_more", + "magicblock-engine", + "magicblock-engine-nucleus", + "magicblock-keeper", + "magicblock-ledger", + "scc", + "snedfile", + "solana-account", + "solana-keypair", + "solana-pubkey", + "solana-signature", + "solana-sysvar", + "thiserror 2.0.19", + "tokio", + "tracing", + "v42-calculator-interface", + "wincode", +] + [[package]] name = "managed" version = "0.8.0" @@ -3254,6 +3315,25 @@ version = "1.15.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" +[[package]] +name = "snedfile" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b49d9659526e7275bf3cd23e43cff8d14e4864ff1eb1e3836a07decbc3689b4" +dependencies = [ + "libc", +] + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + [[package]] name = "solana-account" version = "4.3.1" @@ -3262,7 +3342,7 @@ dependencies = [ "bitflags 2.13.1", "serde", "serde_bytes", - "solana-account 4.3.1", + "solana-account", "solana-account-info", "solana-clock", "solana-instruction-error", @@ -3273,19 +3353,6 @@ dependencies = [ "wincode", ] -[[package]] -name = "solana-account" -version = "4.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "26788ba0d7eb5b250edda3e1d393739bafd5daf94882f2261d14f5ab0d6a25e0" -dependencies = [ - "solana-account-info", - "solana-clock", - "solana-instruction-error", - "solana-pubkey", - "solana-sdk-ids", -] - [[package]] name = "solana-account-info" version = "3.1.1" @@ -3341,6 +3408,17 @@ dependencies = [ "solana-define-syscall 3.0.0", ] +[[package]] +name = "solana-bincode" +version = "3.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "278a1a5bad62cd9da89ac8d4b7ec444e83caa8ae96aa656dfc27684b28d49a5d" +dependencies = [ + "bincode", + "serde_core", + "solana-instruction-error", +] + [[package]] name = "solana-blake3-hasher" version = "3.1.0" @@ -3455,6 +3533,15 @@ dependencies = [ "solana-sdk-ids", ] +[[package]] +name = "solana-compute-budget-program" +version = "4.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c05c42c6d7a20d221ba188d0ab6914081910e8ff4b35c02e6536b94104f575c" +dependencies = [ + "solana-program-runtime", +] + [[package]] name = "solana-cpi" version = "3.1.0" @@ -3552,7 +3639,7 @@ dependencies = [ "bincode", "serde", "serde_derive", - "solana-account 4.3.2", + "solana-account", "solana-account-info", "solana-instruction", "solana-program-error", @@ -3571,6 +3658,7 @@ dependencies = [ "log", "serde", "serde_derive", + "wincode", ] [[package]] @@ -3608,9 +3696,9 @@ dependencies = [ [[package]] name = "solana-hash-512" -version = "1.2.0" +version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ee9c987913768643be3f4fc5f8ec667e8ec19e59e6d131688ade8d1430dafe9" +checksum = "7ce934b02bab639341f2fd62c3e7e3c39dcceb47f0196b7630ed1f82ecb704bd" [[package]] name = "solana-instruction" @@ -3747,6 +3835,34 @@ version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ae8dd4c280dca9d046139eb5b7a5ac9ad10403fbd64964c7d7571214950d758f" +[[package]] +name = "solana-nonce" +version = "3.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4172d5b33a0a38fcdb2af8f84406570dd51567267da96c28ad0f31fc11621c0" +dependencies = [ + "serde", + "serde_derive", + "solana-fee-calculator", + "solana-hash", + "solana-pubkey", + "solana-sha256-hasher", + "wincode", +] + +[[package]] +name = "solana-nonce-account" +version = "4.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b81bf7e2aa8c443724051507c1007d0b6d9c34b70925d3232dc78f5ca485209e" +dependencies = [ + "solana-account", + "solana-hash", + "solana-nonce", + "solana-sdk-ids", + "wincode", +] + [[package]] name = "solana-packet" version = "4.2.0" @@ -3819,7 +3935,7 @@ dependencies = [ "qualifier_attr", "scc", "serde", - "solana-account 4.3.1", + "solana-account", "solana-account-info", "solana-clock", "solana-epoch-rewards", @@ -4120,7 +4236,7 @@ dependencies = [ "qualifier_attr", "rand 0.9.5", "serde", - "solana-account 4.3.1", + "solana-account", "solana-clock", "solana-ed25519-program", "solana-epoch-schedule", @@ -4160,7 +4276,7 @@ version = "4.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3170a3cfc032f3efca975154dee904ecefea5ae11fbd50787c062886196d32c1" dependencies = [ - "solana-account 4.3.2", + "solana-account", "solana-clock", "solana-precompile-error", "solana-pubkey", @@ -4231,7 +4347,7 @@ dependencies = [ "bincode", "libsecp256k1", "num-traits", - "solana-account 4.3.2", + "solana-account", "solana-account-info", "solana-big-mod-exp", "solana-blake3-hasher", @@ -4280,6 +4396,30 @@ dependencies = [ "wincode", ] +[[package]] +name = "solana-system-program" +version = "4.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84eaae98e1576c1b490760d102ca15cb0cfc7ed2b52012dc0d06698a2529e093" +dependencies = [ + "bincode", + "log", + "solana-account", + "solana-bincode", + "solana-fee-calculator", + "solana-instruction", + "solana-nonce", + "solana-nonce-account", + "solana-packet", + "solana-program-runtime", + "solana-pubkey", + "solana-sdk-ids", + "solana-svm-log-collector", + "solana-system-interface", + "solana-sysvar", + "solana-transaction-context", +] + [[package]] name = "solana-sysvar" version = "4.1.0" @@ -4354,7 +4494,7 @@ version = "4.1.1" dependencies = [ "bincode", "serde", - "solana-account 4.3.1", + "solana-account", "solana-account-info", "solana-instruction", "solana-instructions-sysvar", @@ -4501,7 +4641,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys", @@ -4639,10 +4779,12 @@ version = "1.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" dependencies = [ + "bytes", "libc", "mio", "pin-project-lite", "signal-hook-registry", + "socket2", "tokio-macros", "windows-sys", ] diff --git a/Cargo.toml b/Cargo.toml index 56168ddd..c0d6e21a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ members = [ "programs/magic-root-program", "programs/v42-calculator-interface", "programs/v42-calculator-program", + "replicator", "solana/account", "solana/program-runtime", "solana/svm", diff --git a/accountsdb/src/store/index.rs b/accountsdb/src/store/index.rs index 776be137..7b3c95c2 100644 --- a/accountsdb/src/store/index.rs +++ b/accountsdb/src/store/index.rs @@ -15,9 +15,9 @@ use solana_pubkey::Pubkey; use crate::store::kv::{KeyTail, Offset, OwnerAndOffset, PubkeyBytes, U32LE}; /// LMDB map size for the index database. -#[cfg(any(test, feature = "testkit"))] +#[cfg(feature = "testkit")] const INDEX_MAP_SIZE: usize = nucleus::MB; -#[cfg(not(any(test, feature = "testkit")))] +#[cfg(not(feature = "testkit"))] const INDEX_MAP_SIZE: usize = nucleus::GB; /// Subdirectory used for the LMDB index. const INDEX_SUBDIR: &str = "index"; diff --git a/accountsdb/src/store/mmap.rs b/accountsdb/src/store/mmap.rs index e394016d..26ce556b 100644 --- a/accountsdb/src/store/mmap.rs +++ b/accountsdb/src/store/mmap.rs @@ -29,16 +29,16 @@ const DATABASE_META_RESERVATION: usize = 256; /// Filename used for the mapped storage file. pub const STORAGE_FILE: &str = "storage.db"; /// Growth block for the mapped storage file. -#[cfg(any(test, feature = "testkit"))] +#[cfg(feature = "testkit")] pub(crate) const STORAGE_BLOCK: u64 = 16 * MB as u64; -#[cfg(not(any(test, feature = "testkit")))] +#[cfg(not(feature = "testkit"))] pub(crate) const STORAGE_BLOCK: u64 = 256 * MB as u64; /// Initial file size: one storage block plus the metadata reservation. const INIT_STORAGE_SIZE: u64 = STORAGE_BLOCK + DATABASE_META_RESERVATION as u64; /// Maximum mapped storage size. -#[cfg(any(test, feature = "testkit"))] +#[cfg(feature = "testkit")] const MMAP_SIZE: usize = 64 * MB; -#[cfg(not(any(test, feature = "testkit")))] +#[cfg(not(feature = "testkit"))] const MMAP_SIZE: usize = u32::MAX as usize * STORAGE_UNIT + DATABASE_META_RESERVATION; /// One allocation inside the mapped storage. diff --git a/keeper/src/lib.rs b/keeper/src/lib.rs index 522073ce..8e54b129 100644 --- a/keeper/src/lib.rs +++ b/keeper/src/lib.rs @@ -52,7 +52,7 @@ mod metrics; mod subscriptions; mod util; -#[cfg(any(test, feature = "testkit"))] +#[cfg(feature = "testkit")] pub mod testkit; #[cfg(test)] diff --git a/ledger/Cargo.toml b/ledger/Cargo.toml index f69be780..f1077cdd 100644 --- a/ledger/Cargo.toml +++ b/ledger/Cargo.toml @@ -40,6 +40,7 @@ solana-signature = { workspace = true, features = ["wincode"] } solana-transaction-error = { workspace = true, features = ["wincode"] } [dev-dependencies] +ledger = { workspace = true, features = ["testkit"] } nucleus = { workspace = true, features = ["testkit"] } tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 3fad3a9f..247d9a5d 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -20,9 +20,9 @@ use crate::schema::Offset; /// Index directory below each superblock directory. const INDEX_SUBDIR: &str = "index"; /// Maximum LMDB map size for ledger indexes. -#[cfg(any(test, feature = "testkit"))] +#[cfg(feature = "testkit")] const INDEX_MAP_SIZE: usize = 32 * nucleus::MB; -#[cfg(not(any(test, feature = "testkit")))] +#[cfg(not(feature = "testkit"))] const INDEX_MAP_SIZE: usize = 16 * nucleus::GB; /// Number of LMDB named databases in the ledger index. const INDEX_DBS: u32 = 3; diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs index eb1c68df..8af03947 100644 --- a/ledger/src/lib.rs +++ b/ledger/src/lib.rs @@ -78,9 +78,9 @@ impl Ledger { .spawn(|| appender.run(sh))?; let (reader_tx, rx) = flume::bounded(SERVICE_QUEUE_CAPACITY); - #[cfg(not(any(test, feature = "testkit")))] + #[cfg(not(feature = "testkit"))] let readers = num_cpus::get() as u32; - #[cfg(any(test, feature = "testkit"))] + #[cfg(feature = "testkit")] let readers = 1; for id in 0..readers { diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs index 42f0b91d..cb7750e0 100644 --- a/ledger/src/reader.rs +++ b/ledger/src/reader.rs @@ -102,6 +102,7 @@ impl LedgerReader { } } } + // Release ledger ownership before the manager can reopen it. drop(self); shutdown.terminate(ShutdownReason::Signalled); } diff --git a/replicator/Cargo.toml b/replicator/Cargo.toml new file mode 100644 index 00000000..fbd20941 --- /dev/null +++ b/replicator/Cargo.toml @@ -0,0 +1,44 @@ +[package] +name = "magicblock-replicator" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "replicator" + +[dependencies] +engine = { workspace = true } +keeper = { workspace = true } +ledger = { workspace = true } +nucleus = { workspace = true, features = ["ledger", "service"] } + +derive_more = { workspace = true, features = ["from"] } +scc = { workspace = true } +snedfile = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["io-util", "macros", "net", "sync", "time"] } +tracing = { workspace = true } +wincode = { workspace = true } + +solana-keypair = { workspace = true } +solana-pubkey = { workspace = true } +solana-signature = { workspace = true } + +[dev-dependencies] +engine = { workspace = true, features = ["testkit"] } +keeper = { workspace = true, features = ["testkit"] } +nucleus = { workspace = true, features = ["testkit"] } +tokio = { workspace = true, features = ["rt-multi-thread", "time"] } +v42-calculator-interface = { workspace = true, features = ["builder"] } + +solana-account = { workspace = true } +solana-sysvar = { workspace = true } + +[lints] +workspace = true diff --git a/replicator/README.md b/replicator/README.md new file mode 100644 index 00000000..9f6fae63 --- /dev/null +++ b/replicator/README.md @@ -0,0 +1,57 @@ +# `magicblock-replicator` + +Replicator transfers durable execution state over TCP between independent +engine deployments, typically running on different machines. The follower +reports its durable `BlockstorePosition`; the server either resumes the retained +blockstore stream or sends the newest available accountsdb snapshot. + +## Protocol + +Protocol version 1 uses wincode control messages prefixed by a little-endian +`u32` length. Control frames are limited to 65,535 bytes before allocation. +Snapshot archives and blockstore bytes follow the selected response without +additional framing. + +Handshake requests and responses are signed with the sender's local key and +must be within 30 seconds of the receiver's clock. A server accepts only local +follower identities in its allowlist; an empty allowlist denies all followers. +A follower identity may hold only one active transfer at a time; its reservation +is released when that connection's worker exits. Stream workers detect peer +disconnects through writes triggered by durable cursor updates, which are +published at least every time block is produced while the engine is running. +A follower verifies responses against `Engine::authority()`, which must be +configured with the source authority through `nucleus::config::Authority::remote`. + +Every dispatcher must sign with that same canonical authority key. A follower +whose local signer differs from `Engine::authority()` is therefore a terminal +leaf and dispatcher startup rejects it before binding a listener. Any number of +such leaves may follow the source or a relay. Every relay instead holds the +shared private key, so the source and all relays have one compromise and key +rotation boundary. + +The async dispatcher accepts sockets and assigns each connection to a blocking +thread. File and socket operations on that thread use bounded blocking I/O. +Published ledger cursors are transfer boundaries, including sealed tails and +intermediate superblocks. + +## Follower recovery + +Before each handshake, the follower quiesces execution, flushes queued ledger +appends, and reports the resulting cursor. A received snapshot is written to the +successor superblock directory and its seal is appended synchronously. The +seal's cumulative transaction count replaces the follower ledger baseline, +including when a nonempty follower falls behind retention. The client then +reports `RestartRequired`; keeper restores the staged snapshot on the next +startup and engine replay advances it to the ledger tip. + +Externally paced shutdown flushes the cursor before writing +`CURRENT/volatile.db`. Internally paced origins instead append one reset marker +at startup before producing their first new block, so followers clear +chain-mirrored volatile state at the same stream position while retaining +internal system accounts. + +A shared-key follower may also serve downstream followers. It derives and +validates superblock seals from replicated block boundaries and archives its own +snapshots, while downstream clients continue to verify every response against +the original source authority. A distinct-key follower can consume the same +state but cannot relay it. diff --git a/replicator/src/client.rs b/replicator/src/client.rs new file mode 100644 index 00000000..687e26a4 --- /dev/null +++ b/replicator/src/client.rs @@ -0,0 +1,224 @@ +use std::{ + fs::{self, File}, + io::{self, BufReader, Read}, + net::{SocketAddr, TcpStream}, + thread, +}; + +use derive_more::Deref; +use engine::{Engine, EngineError, ReplayError, pacemaker::ExternalBlock}; +use ledger::{ + Superblock, + schema::{Block, OwnedBlockstoreEntry, blockstore}, +}; +use nucleus::{ + KB, + ledger::{ACCOUNTSDB_SNAPSHOT_FILE, BlockstorePosition}, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, +}; +use tokio::{ + runtime, + sync::mpsc::{Receiver, Sender}, + time, +}; +use tracing::{error, info, warn}; + +use crate::{ + IO_TIMEOUT, MAX_RECONNECT_ATTEMPTS, RETRY_DELAY, ReplicationError, Result, + metrics::{self, Operation}, + protocol::{ + self, Handshake, HandshakeRequest, HandshakeResponse, PROTO_VERSION, SnapshotMetadata, + }, +}; + +type ReplicationStream = BufReader; + +/// Pulls a leader blockstore stream into an externally paced follower engine. +#[derive(Deref)] +pub struct ReplicationClient { + /// Engine receiving replicated transactions, boundaries, seals, and resets. + #[deref] + engine: Engine, + /// Leader endpoint reused after transport loss. + addr: SocketAddr, + /// External pacemaker channel used to preserve block-boundary ordering. + pacer: Sender, + /// Locally committed block boundaries used to verify replicated output. + blocks: Receiver, +} + +impl ReplicationClient { + /// Starts the follower worker; connection failures are reported through shutdown management. + pub fn spawn( + addr: SocketAddr, + engine: Engine, + pacer: Sender, + shutdown: &mut ShutdownManager, + ) -> Result<()> { + metrics::init(); + let shutdown = shutdown.handle(Service::ReplicationClient); + let mut blocks = engine.blocks().subscribe(); + // drain the channel from potential leftovers + while blocks.try_recv().is_ok() {} + let client = Self { engine, addr, pacer, blocks }; + let rt = runtime::Builder::new_current_thread().enable_time().build()?; + thread::Builder::new() + .name("replication-client".into()) + .spawn(move || rt.block_on(client.run(shutdown)))?; + Ok(()) + } + + /// Consumes the leader stream and reports why the client stopped. + async fn run(self, mut shutdown: ShutdownHandle) { + let result = self.consume(&shutdown).await; + if shutdown.requested() || result.is_ok() { + shutdown.terminate(ShutdownReason::Signalled); + return; + } + match result { + Err(ReplicationError::RestartRequired(slot)) => { + info!(%slot, "replication client has requested node restart"); + shutdown.terminate(ShutdownReason::RestartRequired); + } + Err(error) => { + shutdown.terminate(ShutdownReason::Error(Box::new(error))); + } + Ok(()) => (), + } + } + + /// Reads blockstore entries from the leader, reconnecting on transport loss, + /// until shutdown is requested or a non-recoverable error occurs. + async fn consume(mut self, shutdown: &ShutdownHandle) -> Result<()> { + let mut stream = self.reconnect(shutdown).await?; + let mut connected = metrics::client_connection(); + loop { + if shutdown.requested() { + return Ok(()); + } + match blockstore::decode(&mut stream) { + Ok(entry) => self.process(entry).await?, + Err(wincode::error::ReadError::Io(error)) => { + warn!(?error, "replication stream disconnected"); + drop(connected); + stream = self.reconnect(shutdown).await?; + connected = metrics::client_connection(); + } + Err(error) => Err(wincode::Error::from(error))?, + } + } + } + + /// Applies one blockstore entry to the follower engine, holding block-boundary + /// ordering through the pacemaker and flagging superblock seal mismatches. + async fn process(&mut self, entry: OwnedBlockstoreEntry) -> Result<()> { + match entry { + OwnedBlockstoreEntry::Block(block) => { + let (external, guard) = ExternalBlock::new(block); + self.pacer.send(external).await.map_err(EngineError::from)?; + let pending = time::timeout(IO_TIMEOUT, self.blocks.recv()); + let observed = pending.await?.ok_or(ReplicationError::StreamClosed)?; + if block != observed { + // Mismatches are diagnostic until recovery policy is implemented. + error!(?block, ?observed, "replication block divergence detected"); + } + guard.await.map_err(EngineError::from)?; + } + OwnedBlockstoreEntry::Superblock(expected) => { + // The preceding boundary finalized local state; this seal only validates it. + let observed = self.superblocks().sealed(); + if observed != expected { + error!(?expected, ?observed, "replication state mismatch detected"); + metrics::client_state_mismatch(); + Err(EngineError::Replay(ReplayError::StateMismatch))?; + } + } + entry => self.engine.replay(entry).await?, + } + Ok(()) + } + + /// Handshakes with the leader at `position`; either stages a snapshot and + /// signals a required restart, or returns the resumed byte stream. + fn connect(&self, position: BlockstorePosition) -> Result { + let _timer = metrics::time(Operation::ClientConnect); + let mut connection = TcpStream::connect_timeout(&self.addr, IO_TIMEOUT)?; + connection.set_read_timeout(Some(IO_TIMEOUT))?; + connection.set_write_timeout(Some(IO_TIMEOUT))?; + let request = HandshakeRequest { version: PROTO_VERSION, position }; + let handshake = Handshake::new(self.signer(), request)?; + protocol::write(&mut connection, &handshake)?; + let handshake = protocol::read::>(&mut connection)?; + handshake.verify()?; + let expected = self.authority(); + if handshake.identity != expected { + let message = format!( + "unexpected replication server identity {}; expected {expected}", + handshake.identity + ); + return Err(ReplicationError::Handshake(message)); + } + + match handshake.payload { + HandshakeResponse::Snapshot(meta) => { + self.stage_snapshot(&mut connection, meta)?; + Err(ReplicationError::RestartRequired(meta.id)) + } + HandshakeResponse::Stream(remote) => { + info!(?position, ?remote, "replication handshake accepted"); + Ok(BufReader::with_capacity(256 * KB, connection)) + } + HandshakeResponse::Err(message) => Err(ReplicationError::Handshake(message)), + } + } + + /// Reconnects from a quiesced local cursor. + async fn reconnect(&self, shutdown: &ShutdownHandle) -> Result { + // Hold quiescence so every retry uses the same flushed position. + let _guard = self.barrier().await?; + self.sync(false)?; + let position = self.superblocks().position(); + for attempt in 1..=MAX_RECONNECT_ATTEMPTS { + if shutdown.requested() { + return Err(ReplicationError::StreamClosed); + } + metrics::client_connection_attempt(); + match self.connect(position) { + Ok(stream) => { + info!(attempt, ?position, "replication stream connected"); + return Ok(stream); + } + Err(ReplicationError::IO(error)) => { + warn!(attempt, ?error, "replication reconnect failed"); + } + Err(error) => return Err(error), + } + let timeout = RETRY_DELAY * attempt as u32; + if time::timeout(timeout, shutdown.signalled()).await.is_ok() { + return Err(ReplicationError::StreamClosed); + } + } + Err(ReplicationError::ReconnectExhausted) + } + + /// Writes the incoming snapshot archive into a fresh superblock directory and + /// records its seal, readying the follower to restart from that state. + fn stage_snapshot(&self, connection: &mut TcpStream, meta: SnapshotMetadata) -> Result<()> { + let _timer = metrics::time(Operation::ClientStageSnapshot); + // Stage in the successor before seal rotation so restart can find it. + let dir = Superblock::init_dir(self.superblocks().directory(), meta.id + 1)?; + let archive = dir.join(ACCOUNTSDB_SNAPSHOT_FILE); + let temporary = dir.join(format!("{ACCOUNTSDB_SNAPSHOT_FILE}.tmp")); + let mut file = File::options().write(true).create(true).truncate(true).open(&temporary)?; + let written = io::copy(&mut connection.take(meta.len), &mut file)?; + if written != meta.len { + return Err(ReplicationError::Snapshot(meta.len, written)); + } + file.sync_all()?; + drop(file); + fs::rename(temporary, archive)?; + self.superblocks().bootstrap(meta.superblock)?; + info!(?meta, "replication snapshot staged"); + Ok(()) + } +} diff --git a/replicator/src/error.rs b/replicator/src/error.rs new file mode 100644 index 00000000..24e48b33 --- /dev/null +++ b/replicator/src/error.rs @@ -0,0 +1,54 @@ +use std::io; + +use engine::EngineError; +use keeper::error::KeeperError; +use ledger::LedgerError; +use nucleus::ledger::BlockstorePosition; +use tokio::time::error::Elapsed; + +/// Failure while negotiating or transferring replicated state. +#[derive(thiserror::Error, Debug)] +pub enum ReplicationError { + /// Socket or replicated-file access failed. + #[error("replication I/O failed: {0}")] + IO(#[from] io::Error), + /// Applying replicated state through the keeper failed. + #[error("failed to apply replicated state: {0}")] + State(#[from] KeeperError), + /// Applying a replicated entry through the execution engine failed. + #[error("replication engine operation failed: {0}")] + Engine(#[from] EngineError), + /// Reading or advancing replicated ledger storage failed. + #[error("replication ledger operation failed: {0}")] + Ledger(#[from] LedgerError), + /// A control message could not be encoded or decoded. + #[error("invalid replication control message: {0}")] + Serde(#[from] wincode::Error), + /// The peer uses a protocol version this crate cannot read. + #[error("replication protocol version mismatch; expected version {0}")] + VersionMismatch(u32), + /// The requested or published blockstore cursor is unavailable locally. + #[error("replication position is unavailable: {0:?}")] + PositionNotFound(BlockstorePosition), + /// The leader rejected the client's handshake. + #[error("replication handshake rejected: {0}")] + Handshake(String), + /// The snapshot connection ended before the advertised byte count arrived. + #[error("incomplete replication snapshot: expected {0} bytes, received {1}")] + Snapshot(u64, u64), + /// No complete retained snapshot can satisfy an unavailable cursor. + #[error("no complete replication snapshot is available")] + SnapshotUnavailable, + /// All bounded attempts to reconnect to the leader failed. + #[error("replication reconnect attempts exhausted")] + ReconnectExhausted, + /// A staged snapshot must be installed by restarting the engine. + #[error("replication snapshot for superblock {0} is staged; restart required")] + RestartRequired(u64), + /// A replication event stream closed before the transfer completed. + #[error("replication event stream closed")] + StreamClosed, + /// Waiting for a locally committed block boundary timed out. + #[error("timed out waiting for a replicated block boundary: {0}")] + Timeout(#[from] Elapsed), +} diff --git a/replicator/src/lib.rs b/replicator/src/lib.rs new file mode 100644 index 00000000..899fd4ea --- /dev/null +++ b/replicator/src/lib.rs @@ -0,0 +1,23 @@ +#![doc = include_str!("../README.md")] + +mod client; +mod error; +mod metrics; +mod protocol; +mod server; + +use std::time::Duration; + +pub use client::ReplicationClient; +pub use error::ReplicationError; +pub use protocol::PROTO_VERSION; +pub use server::ReplicationDispatcher; + +type Result = std::result::Result; + +/// Read/write timeout applied to replication sockets on both sides. +const IO_TIMEOUT: Duration = Duration::from_secs(4); +/// Delay a follower waits between reconnect attempts. +const RETRY_DELAY: Duration = Duration::from_secs(1); +/// Reconnect attempts a follower makes before giving up. +const MAX_RECONNECT_ATTEMPTS: usize = 10; diff --git a/replicator/src/metrics.rs b/replicator/src/metrics.rs new file mode 100644 index 00000000..c7ab18b9 --- /dev/null +++ b/replicator/src/metrics.rs @@ -0,0 +1,136 @@ +//! Prometheus metrics for replication clients and servers. + +use std::sync::OnceLock; + +use nucleus::metrics::{self as metric, OperationTimer}; +use nucleus::metrics::{IntCounter, IntGauge, MetricOperation, MetricSpec, OperationCounters}; + +/// Process-wide replicator metrics registered in the default Prometheus registry. +static METRICS: OnceLock = OnceLock::new(); + +const OPERATION_TIME: MetricSpec = MetricSpec { + name: "replicator_operation_duration_micros", + help: "Replication operation duration distribution in microseconds.", +}; +const CLIENT_STREAM_CONNECTED: MetricSpec = MetricSpec { + name: "replicator_client_stream_connected", + help: "Whether the replication client currently holds a live blockstore stream.", +}; +const SERVER_CONNECTIONS: MetricSpec = MetricSpec { + name: "replicator_server_connections", + help: "Current replication server connection workers.", +}; +const CLIENT_CONNECTION_ATTEMPTS: MetricSpec = MetricSpec { + name: "replicator_client_connection_attempts", + help: "Replication client connection attempts.", +}; +const CLIENT_STATE_MISMATCHES: MetricSpec = MetricSpec { + name: "replicator_client_state_mismatches", + help: "Superblock seal mismatches detected by the replication client.", +}; +const SERVER_CURSOR_UPDATES_SKIPPED: MetricSpec = MetricSpec { + name: "replicator_server_cursor_updates_skipped", + help: "Replication server cursor updates skipped after receiver lag.", +}; + +/// Replication operation used as a fixed low-cardinality label. +#[derive(Clone, Copy)] +pub(crate) enum Operation { + ClientConnect, + ClientStageSnapshot, + ServerHandshake, + ServerAdvance, + ServerSendSnapshot, +} + +impl MetricOperation for Operation { + fn label(self) -> &'static str { + match self { + Self::ClientConnect => "client_connect", + Self::ClientStageSnapshot => "client_stage_snapshot", + Self::ServerHandshake => "server_handshake", + Self::ServerAdvance => "server_advance", + Self::ServerSendSnapshot => "server_send_snapshot", + } + } +} + +/// Registers all replicator metrics once. +pub(crate) fn init() { + METRICS.get_or_init(Default::default); +} + +/// Starts an operation timer that records latency when the returned guard drops. +pub(crate) fn time(op: Operation) -> OperationTimer<'static> { + op.time(METRICS.get().map(|m| &m.operations)) +} + +/// Records a client connection attempt. +pub(crate) fn client_connection_attempt() { + metric::with_metrics(&METRICS, |m| m.client_connection_attempts.inc()); +} + +/// Marks a client blockstore stream live until the returned guard drops. +pub(crate) fn client_connection() -> ClientConnection { + metric::with_metrics(&METRICS, |m| m.client_stream_connected.set(1)); + ClientConnection +} + +/// Records a superblock seal mismatch. +pub(crate) fn client_state_mismatch() { + metric::with_metrics(&METRICS, |m| m.client_state_mismatches.inc()); +} + +/// Counts a server worker until the returned guard drops. +pub(crate) fn server_connection() -> ServerConnection { + metric::with_metrics(&METRICS, |m| m.server_connections.inc()); + ServerConnection +} + +/// Records durable cursor updates skipped by a lagged receiver. +pub(crate) fn server_cursor_updates_skipped(skipped: u64) { + metric::with_metrics(&METRICS, |m| { + m.server_cursor_updates_skipped.inc_by(skipped) + }); +} + +/// Clears the live-client gauge on every stream exit path. +pub(crate) struct ClientConnection; + +impl Drop for ClientConnection { + fn drop(&mut self) { + metric::with_metrics(&METRICS, |m| m.client_stream_connected.set(0)); + } +} + +/// Decrements the active-server-worker gauge on every worker exit path. +pub(crate) struct ServerConnection; + +impl Drop for ServerConnection { + fn drop(&mut self) { + metric::with_metrics(&METRICS, |m| m.server_connections.dec()); + } +} + +/// Owns all Prometheus collectors registered by replicator. +struct Metrics { + operations: OperationCounters, + client_stream_connected: IntGauge, + server_connections: IntGauge, + client_connection_attempts: IntCounter, + client_state_mismatches: IntCounter, + server_cursor_updates_skipped: IntCounter, +} + +impl Default for Metrics { + fn default() -> Self { + Self { + operations: OperationCounters::new(OPERATION_TIME), + client_stream_connected: metric::gauge(CLIENT_STREAM_CONNECTED, 0), + server_connections: metric::gauge(SERVER_CONNECTIONS, 0), + client_connection_attempts: metric::counter(CLIENT_CONNECTION_ATTEMPTS, 0), + client_state_mismatches: metric::counter(CLIENT_STATE_MISMATCHES, 0), + server_cursor_updates_skipped: metric::counter(SERVER_CURSOR_UPDATES_SKIPPED, 0), + } + } +} diff --git a/replicator/src/protocol.rs b/replicator/src/protocol.rs new file mode 100644 index 00000000..846a545a --- /dev/null +++ b/replicator/src/protocol.rs @@ -0,0 +1,156 @@ +use std::{ + io::{Read, Write}, + time::Duration, +}; + +use derive_more::Deref; +use ledger::schema::SuperblockSeal; +use nucleus::{ledger::BlockstorePosition, unix_time}; +use solana_keypair::{Keypair, Signer}; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use wincode::{SchemaRead, SchemaWrite, config::DefaultConfig}; + +use crate::{ReplicationError, Result}; + +/// Wire protocol version accepted by this crate. +pub const PROTO_VERSION: u32 = 1; + +/// Encoded length prefix preceding every control frame. +const HEADER_LENGTH: usize = size_of::(); +/// Largest control frame accepted before allocating its payload. +const MAX_CONTROL_FRAME_LENGTH: u32 = u16::MAX as u32; +/// Maximum clock difference accepted for a signed control message. +const MAX_CLOCK_SKEW: Duration = Duration::from_secs(30); + +/// Signed, freshness-bounded control message exchanged during negotiation. +#[derive(SchemaRead, SchemaWrite)] +pub(crate) struct Handshake

{ + /// Direction-specific handshake payload. + pub(crate) payload: P, + /// Public identity of the signer. + pub(crate) identity: Pubkey, + /// Unix timestamp in microseconds covered by the signature. + pub(crate) timestamp: u64, + /// Signature over the payload and timestamp. + pub(crate) signature: Signature, +} + +/// Initial follower request identifying its last durable blockstore byte. +#[derive(SchemaRead, SchemaWrite)] +pub(crate) struct HandshakeRequest { + /// Wire version understood by the follower. + pub(crate) version: u32, + /// Next durable blockstore byte required by the follower. + pub(crate) position: BlockstorePosition, +} + +/// Leader decision following a valid handshake. +#[derive(SchemaRead, SchemaWrite)] +pub(crate) enum HandshakeResponse { + /// Snapshot that must be staged before replication can resume. + Snapshot(SnapshotMetadata), + /// Leader cursor from which live streaming begins. + Stream(BlockstorePosition), + /// Reason the leader rejected negotiation. + Err(String), +} + +/// Describes the accountsdb snapshot a follower must stage before it can stream. +#[derive(SchemaRead, SchemaWrite, Debug, Clone, Copy, Deref)] +pub(crate) struct SnapshotMetadata { + /// Length of the snapshot archive in bytes. + pub(crate) len: u64, + /// Seal the snapshot restores accountsdb to. + #[deref] + pub(crate) superblock: SuperblockSeal, +} + +impl

Handshake

+where + for<'de> P: SchemaRead<'de, DefaultConfig, Dst = P>, + P: SchemaWrite, +{ + /// Signs the payload with a fresh timestamp and publishes the signer's identity. + pub(crate) fn new(keypair: &Keypair, payload: P) -> Result { + let identity = keypair.pubkey(); + let timestamp = timestamp(); + let data = message(timestamp, &payload)?; + let signature = keypair.sign_message(&data); + Ok(Self { + payload, + identity, + timestamp, + signature, + }) + } + + /// Rejects altered messages and timestamps outside the accepted clock-skew window. + /// + /// Freshness is time-based; duplicate messages inside the window are not tracked. + pub(crate) fn verify(&self) -> Result<()> { + let data = message(self.timestamp, &self.payload)?; + if !self.signature.verify(self.identity.as_ref(), &data) { + let msg = "invalid handshake signature"; + return Err(ReplicationError::Handshake(msg.into())); + } + let skew = timestamp().abs_diff(self.timestamp); + if skew > MAX_CLOCK_SKEW.as_micros() as u64 { + let msg = "handshake timestamp exceeds maximum clock skew"; + return Err(ReplicationError::Handshake(msg.into())); + } + Ok(()) + } +} + +/// Reads and decodes one length-prefixed control message. +pub(crate) fn read(reader: &mut impl Read) -> Result +where + for<'de> T: SchemaRead<'de, DefaultConfig, Dst = T>, +{ + let mut header = [0; HEADER_LENGTH]; + reader.read_exact(&mut header)?; + let len = u32::from_le_bytes(header); + if len > MAX_CONTROL_FRAME_LENGTH { + return Err(ReplicationError::Handshake(format!( + "replication control frame len {len} exceeds max allowed" + ))); + } + + let mut payload = vec![0; len as usize]; + reader.read_exact(&mut payload)?; + wincode::deserialize_exact(&payload) + .map_err(wincode::Error::from) + .map_err(Into::into) +} + +/// Encodes and writes one length-prefixed control message. +pub(crate) fn write(writer: &mut impl Write, message: &T) -> Result<()> +where + T: SchemaWrite + ?Sized, +{ + let payload = wincode::serialize(message).map_err(wincode::Error::from)?; + let len = payload.len() as u32; + if len > MAX_CONTROL_FRAME_LENGTH { + return Err(ReplicationError::Handshake(format!( + "replication control frame len {len} exceeds max allowed", + ))); + } + writer.write_all(&len.to_le_bytes())?; + writer.write_all(&payload)?; + writer.flush().map_err(Into::into) +} + +fn timestamp() -> u64 { + unix_time().as_micros() as u64 +} + +/// Builds the protocol byte string covered by a handshake signature. +fn message

(ts: u64, payload: &P) -> Result> +where + P: SchemaWrite, +{ + let mut data = wincode::serialize(payload).map_err(wincode::Error::from)?; + data.extend_from_slice(&ts.to_le_bytes()); + Ok(data) +} diff --git a/replicator/src/server.rs b/replicator/src/server.rs new file mode 100644 index 00000000..dfb570cf --- /dev/null +++ b/replicator/src/server.rs @@ -0,0 +1,352 @@ +use std::{ + fs::File, + io::Write, + net::{SocketAddr, TcpStream}, + sync::Arc, + thread, +}; + +use derive_more::Deref; +use engine::Engine; +use ledger::schema::SuperblockSeal; +use nucleus::{ + ledger::{ACCOUNTSDB_SNAPSHOT_FILE, BlockstorePosition}, + shutdown::{CancellationToken, Service, ShutdownHandle, ShutdownManager, ShutdownReason}, +}; +use scc::HashMap; +use solana_keypair::Signer; +use solana_pubkey::Pubkey; +use tokio::{ + net::{TcpListener, TcpStream as AsyncTcpStream}, + runtime, + sync::broadcast, +}; +use tracing::{error, info, warn}; + +use crate::{ + IO_TIMEOUT, ReplicationError, Result, + metrics::{self, Operation}, + protocol::{ + self, Handshake, HandshakeRequest, HandshakeResponse, PROTO_VERSION, SnapshotMetadata, + }, +}; + +/// Accepts follower connections and assigns each one a blocking transfer worker. +pub struct ReplicationDispatcher { + /// Accepts inbound follower connections. + listener: TcpListener, + /// Engine whose local signer authenticates responses and whose ledger is served. + engine: Engine, + /// List of follower identities permitted to replicate. + allowed: Arc>>, + /// Cancels the accept loop and parents every per-connection worker. + shutdown: ShutdownHandle, +} + +/// Serves one follower from its requested durable cursor onward. +#[derive(Deref)] +struct ReplicationServer { + /// Blocking, timeout-bounded socket to the follower. + connection: TcpStream, + /// Cursor of the next byte owed to the follower. + position: ReplicationPosition, + /// Engine whose local signer authenticates responses and whose ledger is served. + #[deref] + engine: Engine, + /// Local follower identities permitted to replicate. + allowed: Arc>>, + /// Fires when the dispatcher shuts down. + cancellation: CancellationToken, +} + +/// Open blockstore and cursor from which the next byte must be sent. +struct ReplicationPosition { + /// Open blockstore file for `current.superblock`. + blockstore: File, + /// Durable-cursor updates broadcast by the appender. + stream: broadcast::Receiver, + /// Position of the next byte to send. + current: BlockstorePosition, +} + +/// Initial transfer selected after validating the follower cursor. +enum ReplicationAction { + /// Send a full accountsdb snapshot; the follower restarts from it. + Snapshot { archive: File, meta: SnapshotMetadata }, + /// Resume the blockstore stream from a still-retained cursor. + Stream { from: BlockstorePosition, blockstore: File }, +} + +impl ReplicationDispatcher { + /// Verifies the canonical signer, then binds `addr` and starts the accept loop. + pub async fn spawn( + addr: SocketAddr, + engine: Engine, + allowed: Arc<[Pubkey]>, + shutdown: &mut ShutdownManager, + ) -> Result<()> { + metrics::init(); + if engine.signer().pubkey() != engine.authority() { + warn!("dispatcher is disabled: node cannot act as replication relay"); + return Ok(()); + } + let listener = TcpListener::bind(addr).await?; + let shutdown = shutdown.handle(Service::ReplicationDispatcher); + let allowed = Arc::new(allowed.iter().map(|&identity| (identity, Arc::new(()))).collect()); + let service = Self { + listener, + engine, + allowed, + shutdown, + }; + tokio::spawn(service.run()); + info!(%addr, "replication dispatcher started"); + Ok(()) + } + + /// Accepts until cancellation or a listener failure; connection failures stay isolated. + async fn run(mut self) { + let reason = loop { + tokio::select! { + biased; + _ = self.shutdown.signalled() => break ShutdownReason::Signalled, + result = self.listener.accept() => match result { + Ok((stream, peer)) => { + if let Err(error) = self.dispatch(stream, peer) { + warn!(%peer, ?error, "failed to dispatch replication connection"); + } + } + Err(error) => break ShutdownReason::Error(Box::new(error)), + } + } + }; + // Release engine storage before the manager can reopen it. + drop(self.engine); + self.shutdown.terminate(reason); + } + + /// Converts the accepted async socket into a blocking, timeout-bounded stream + /// and hands it to a dedicated blocking server worker. + fn dispatch(&self, stream: AsyncTcpStream, peer: SocketAddr) -> Result<()> { + let stream = stream.into_std()?; + stream.set_nonblocking(false)?; + stream.set_read_timeout(Some(IO_TIMEOUT))?; + stream.set_write_timeout(Some(IO_TIMEOUT))?; + let engine = self.engine.clone(); + let cancellation = self.shutdown.child(); + let allowed = self.allowed.clone(); + ReplicationServer::spawn(stream, peer, engine, allowed, cancellation) + } +} + +impl ReplicationServer { + /// Starts a blocking worker without blocking the async dispatcher. + fn spawn( + connection: TcpStream, + peer: SocketAddr, + engine: Engine, + allowed: Arc>>, + cancellation: CancellationToken, + ) -> Result<()> { + // Subscribe before sampling so racing cursor updates remain queued. + let stream = engine.ledger().position.subscribe(); + let current = engine.ledger().position(); + let blockstore = blockstore(&engine, current.superblock)?; + let position = ReplicationPosition { stream, current, blockstore }; + let server = Self { + connection, + position, + engine, + allowed, + cancellation, + }; + let runtime = runtime::Builder::new_current_thread().build()?; + thread::Builder::new().name("replication-server".into()).spawn(move || { + let _connection = metrics::server_connection(); + runtime + .block_on(server.run()) + .inspect_err(|error| warn!(%peer, %error, "replication connection failed")) + })?; + Ok(()) + } + + /// Negotiates an initial transfer, catches up immediately, then follows durable cursors. + async fn run(mut self) -> Result<()> { + // Cursor updates arrive at every block and write new durable bytes. Those writes + // detect peer disconnects, exit this worker, and release its identity lease. + let (action, _lease) = match self.handshake() { + Ok(handshake) => handshake, + Err(error) => { + warn!(?error, "replication handshake rejected"); + let response = HandshakeResponse::Err(error.to_string()); + self.respond(response)?; + return Ok(()); + } + }; + + match action { + ReplicationAction::Snapshot { mut archive, meta } => { + let _timer = metrics::time(Operation::ServerSendSnapshot); + info!(?meta, "sending replication snapshot"); + let response = HandshakeResponse::Snapshot(meta); + self.respond(response)?; + send_range(&mut archive, &mut self.connection, 0, meta.len)?; + self.connection.flush()?; + return Ok(()); + } + ReplicationAction::Stream { from, blockstore } => { + let response = HandshakeResponse::Stream(self.position.current); + self.respond(response)?; + self.position.current = from; + self.position.blockstore = blockstore; + let through = self.ledger().position(); + self.advance(through)?; + info!(?from, ?through, "replication caught up"); + } + } + + loop { + tokio::select! { + biased; + _ = self.cancellation.cancelled() => return Ok(()), + result = self.position.stream.recv() => match result { + Ok(position) => self.advance(position)?, + Err(broadcast::error::RecvError::Lagged(skipped)) => { + let position = self.ledger().position(); + // Cursors are cumulative; advancing to latest covers skipped updates. + warn!(skipped, ?position, "replication cursor lagged"); + metrics::server_cursor_updates_skipped(skipped); + self.advance(position)?; + } + Err(broadcast::error::RecvError::Closed) => { + error!("ledger position stream closed unexpectedly"); + return Err(ReplicationError::StreamClosed); + } + } + } + } + } + + /// Signs and writes a leader handshake response with the local engine signer. + fn respond(&mut self, response: HandshakeResponse) -> Result<()> { + let handshake = Handshake::new(self.signer(), response)?; + protocol::write(&mut self.connection, &handshake) + } + + /// Selects retained streaming when possible, otherwise the newest ready snapshot. + fn handshake(&mut self) -> Result<(ReplicationAction, Arc<()>)> { + let _timer = metrics::time(Operation::ServerHandshake); + let handshake: Handshake = protocol::read(&mut self.connection)?; + handshake.verify()?; + let lease = self.reserve(handshake.identity)?; + if handshake.payload.version != PROTO_VERSION { + return Err(ReplicationError::VersionMismatch(PROTO_VERSION)); + } + + let requested = handshake.payload.position; + if requested > self.position.current { + return Err(ReplicationError::PositionNotFound(requested)); + } + let action = match self.ledger().cursor(requested.superblock) { + Some(end) if requested.offset <= end => ReplicationAction::Stream { + from: requested, + blockstore: blockstore(self, requested.superblock)?, + }, + Some(_) => return Err(ReplicationError::PositionNotFound(requested)), + None => self.snapshot()?, + }; + Ok((action, lease)) + } + + /// Falls back to the newest retained superblock that has a staged accountsdb + /// snapshot, used when the follower's cursor is no longer streamable. + fn snapshot(&self) -> Result { + for superblock in self.ledger().iter() { + let path = superblock.directory.join(ACCOUNTSDB_SNAPSHOT_FILE); + if !path.exists() { + continue; + } + let archive = File::open(path)?; + let meta = SnapshotMetadata { + len: archive.metadata()?.len(), + // Successor archives carry the predecessor seal metadata. + superblock: SuperblockSeal { + id: superblock.id.saturating_sub(1), + checksum: superblock.checksum(), + transactions: superblock.transactions(), + }, + }; + return Ok(ReplicationAction::Snapshot { archive, meta }); + } + Err(ReplicationError::SnapshotUnavailable) + } + + /// Sends every durable byte between the current cursor and `target`. + fn advance(&mut self, target: BlockstorePosition) -> Result<()> { + let _timer = metrics::time(Operation::ServerAdvance); + if target <= self.position.current { + return Ok(()); + } + while self.position.current.superblock < target.superblock { + let end = self + .ledger() + .cursor(self.position.current.superblock) + .ok_or(ReplicationError::PositionNotFound(self.position.current))?; + self.send(end)?; + let superblock = self.position.current.superblock + 1; + self.position.blockstore = blockstore(self, superblock)?; + self.position.current = BlockstorePosition { superblock, offset: 0 }; + } + self.send(target.offset) + } + + /// Sends blockstore bytes from the current offset up to `end`, advancing the cursor. + fn send(&mut self, end: u64) -> Result<()> { + let start = self.position.current.offset; + send_range( + &mut self.position.blockstore, + &mut self.connection, + start, + end - start, + )?; + self.position.current.offset = end; + Ok(()) + } + + /// Reserves an allowed identity until the returned lease drops. + fn reserve(&self, identity: Pubkey) -> Result> { + let Some(entry) = self.allowed.get_sync(&identity) else { + let msg = "replication access not allowed"; + return Err(ReplicationError::Handshake(msg.into())); + }; + if Arc::strong_count(entry.get()) > 1 { + let msg = "replication stream already active"; + return Err(ReplicationError::Handshake(msg.into())); + } + Ok(entry.get().clone()) + } +} + +/// Copies `len` bytes from `file` at `offset` to the socket, looping until the +/// full range is transferred (each `send_exact` may transfer only part of it). +fn send_range( + file: &mut File, + stream: &mut TcpStream, + mut offset: u64, + mut len: u64, +) -> Result<()> { + while len != 0 { + let sent = snedfile::send_exact(file, stream, len, offset)?; + offset += sent; + len -= sent; + } + Ok(()) +} + +/// Clones the blockstore file handle of a retained superblock for independent seeking. +fn blockstore(engine: &Engine, superblock: u64) -> Result { + let position = BlockstorePosition { superblock, offset: 0 }; + let candidate = engine.ledger().iter().find(|sb| sb.id == superblock); + let sb = candidate.ok_or(ReplicationError::PositionNotFound(position))?; + sb.blockstore.try_clone().map_err(Into::into) +} diff --git a/replicator/tests/integration.rs b/replicator/tests/integration.rs new file mode 100644 index 00000000..b545f489 --- /dev/null +++ b/replicator/tests/integration.rs @@ -0,0 +1,587 @@ +//! Full-stack, byte-exact replication invariants over real engines and loopback TCP. +//! +//! Followers must match the leader's cursor and seal checksums, not just state. +//! Mutations are non-idempotent so duplicate application is observable. + +#![cfg(test)] + +use std::{ + net::{Ipv4Addr, SocketAddr, TcpListener}, + sync::Arc, + time::Duration, +}; + +use engine::testkit::{Pacing, TestEngine}; +use keeper::testkit::{ + Dirs, WireVersion, keeper_builder, load_v42_data, patterned_bytes, sign_versioned_instructions, + v42_builder, v42_padded_value, +}; +use ledger::request::{BlockDetails, BlockParams, BlockResponse}; +use nucleus::{KB, Slot, config::Authority, ledger::BlockstorePosition, shutdown::ShutdownManager}; +use replicator::{ReplicationClient, ReplicationDispatcher}; +use solana_account::{AccountBuilder, AccountMode, ReadableAccount}; +use solana_keypair::{Keypair, Signer}; +use solana_pubkey::Pubkey; +use solana_sysvar::rent::Rent; +use tokio::{sync::broadcast, time}; +use v42_calculator_interface::builder::Expr as E; + +/// Bound for every asynchronous replication assertion. +const TIMEOUT: Duration = Duration::from_secs(4); + +type AccountSeed = (Pubkey, i64, AccountMode); + +/// Returns an unused loopback address, released immediately after discovery. +fn loopback_addr() -> SocketAddr { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + listener.local_addr().unwrap() +} + +/// Starts an engine over throwaway directories seeded with `accounts`. +async fn engine(authority: Authority, accounts: &[AccountSeed], pacing: Pacing) -> TestEngine { + let dirs = Dirs::default(); + let mut builder = keeper_builder(&dirs); + builder.authority = authority; + for &(key, value, mode) in accounts { + builder.accounts.insert(key, v42_builder(value, mode).build()); + } + TestEngine::from_builder(dirs, builder, pacing).await +} + +/// Starts a leader and follower with distinct local identities and direct trust. +async fn engines( + leader_accounts: &[AccountSeed], + follower_accounts: &[AccountSeed], +) -> (TestEngine, TestEngine) { + let leader_authority: Authority = Keypair::new().into(); + let follower_authority = Authority { + local: Arc::new(Keypair::new()), + remote: Some(leader_authority.local.pubkey()), + }; + let leader = engine(leader_authority, leader_accounts, Pacing::External).await; + let follower = engine(follower_authority, follower_accounts, Pacing::External).await; + (leader, follower) +} + +/// Starts a leader-side dispatcher at an already selected address. +async fn dispatcher(addr: SocketAddr, leader: &TestEngine, allowed: &[Pubkey]) -> ShutdownManager { + let mut shutdown = ShutdownManager::default(); + let engine = (*leader).clone(); + ReplicationDispatcher::spawn(addr, engine, Arc::from(allowed), &mut shutdown) + .await + .unwrap(); + shutdown +} + +/// Registers a replication client with the follower lifecycle manager. +fn replicate(addr: SocketAddr, follower: &mut TestEngine) { + let engine = follower.clone(); + let pacer = follower.pacer(); + ReplicationClient::spawn(addr, engine, pacer, follower.shutdown()).unwrap(); +} + +/// Subscribes before starting replication so the first published position cannot be missed. +fn stream(addr: SocketAddr, follower: &mut TestEngine) -> broadcast::Receiver { + let positions = follower.ledger().position.subscribe(); + replicate(addr, follower); + positions +} + +/// Stages a snapshot through replication and installs it on restart. +async fn restart_from_snapshot(addr: SocketAddr, mut follower: TestEngine) -> TestEngine { + replicate(addr, &mut follower); + follower.shutdown().wait().await; + let (dirs, authority) = follower.close().await; + TestEngine::with(dirs, authority).await +} + +/// Applies a non-idempotent mutation so duplicate replication changes the result. +/// Calls must be separated by a block advance to produce distinct signatures. +async fn increment(engine: &TestEngine, state: Pubkey) { + let ix = (E::acc(1) + E::lit(1)).compose(state, &[state]); + engine.execute(&[ix]).await.expect("increment commits"); +} + +/// Commits one increment and publishes its enclosing block cursor. +async fn commit_increment(engine: &mut TestEngine, state: Pubkey) -> BlockstorePosition { + increment(engine, state).await; + engine.advance(1).await; + engine.sync().await +} + +/// Waits for the exact cursor, flushes the follower, and verifies state and position. +async fn await_replication( + positions: &mut broadcast::Receiver, + follower: &TestEngine, + expected: BlockstorePosition, + state: Pubkey, + value: i64, +) { + time::timeout(TIMEOUT, async { + loop { + let observed = positions.recv().await.expect("position stream is open"); + if observed < expected { + continue; + } + assert_eq!(observed, expected, "follower advanced past the leader"); + return; + } + }) + .await + .expect("replication reaches the synced cursor in time"); + follower.sync().await; + assert_eq!(load_v42_data(follower, state), Some(value)); + assert_eq!(follower.superblocks().position(), expected); +} + +/// Loads the serialized transactions committed in `slot`, preserving ledger order. +async fn block_transactions(engine: &TestEngine, slot: Slot) -> Vec> { + let response = engine + .blocks() + .get(BlockParams { + slot, + details: BlockDetails::Transactions, + }) + .await + .expect("block read succeeds") + .expect("committed block exists"); + let BlockResponse::WithTransactions(block) = response else { + panic!("transaction detail request returns transactions"); + }; + block.transactions +} + +/// Large raw transaction frames survive both retained catch-up and live replay. +#[tokio::test(flavor = "multi_thread")] +async fn replays_large_transactions_during_catch_up_and_live_streaming() { + const BATCHED_INSTRUCTIONS: usize = 32; + const BATCHED_TERMS: usize = 16; + const CREATE_DATA_LEN: usize = 64 * KB + 1; + const UPDATE_DATA_LEN: usize = 128 * KB + 1; + const ACCOUNT_SLOT: Slot = 42; + + let state = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated)]; + let (mut leader, mut follower) = engines(&seed, &seed).await; + let account = Pubkey::new_unique(); + let owner = Pubkey::new_unique(); + let lamports = Rent::default().minimum_balance(UPDATE_DATA_LEN); + + // Persist a standard client transaction and a large private create before + // connecting, forcing both frames through retained-ledger catch-up. + let instructions: Vec<_> = (0..BATCHED_INSTRUCTIONS) + .map(|value| v42_padded_value(state, value as i64, BATCHED_TERMS)) + .collect(); + let (_, client_transaction) = sign_versioned_instructions( + leader.signer(), + WireVersion::Legacy, + &instructions, + leader.blockhash(), + ); + let client_transaction_len = client_transaction.len(); + assert!(client_transaction_len > 4 * KB); + assert!(client_transaction_len < u16::MAX as usize); + leader + .execute(client_transaction) + .await + .expect("large client transaction commits"); + + let created = AccountBuilder::default() + .lamports(lamports) + .owner(owner) + .mode(AccountMode::ReadOnly) + .slot(ACCOUNT_SLOT) + .data(patterned_bytes(CREATE_DATA_LEN, 1)); + leader + .account(account) + .create(created, None) + .await + .expect("large account creation commits"); + + let catch_up_slot = leader.blocks().current_slot(); + leader.advance(1).await; + let expected = leader.sync().await; + let transactions = block_transactions(&leader, catch_up_slot).await; + assert_eq!(transactions.len(), 2); + assert_eq!(transactions[0].len(), client_transaction_len); + assert!(transactions[1].len() > 64 * KB); + + let addr = loopback_addr(); + let follower_identity = follower.signer().pubkey(); + let mut dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + let mut positions = stream(addr, &mut follower); + await_replication( + &mut positions, + &follower, + expected, + state, + (BATCHED_INSTRUCTIONS - 1) as i64, + ) + .await; + assert_eq!(follower.get_account(account), leader.get_account(account)); + + // Grow and replace the same account while connected, exercising the live + // tail with a different multi-chunk payload and advancing account slot. + let updated = AccountBuilder::default() + .lamports(lamports) + .owner(owner) + .mode(AccountMode::ReadOnly) + .slot(ACCOUNT_SLOT + 1) + .data(patterned_bytes(UPDATE_DATA_LEN, 2)); + leader + .account(account) + .update(updated) + .await + .expect("large account update commits"); + + let live_slot = leader.blocks().current_slot(); + leader.advance(1).await; + let expected = leader.sync().await; + let transactions = block_transactions(&leader, live_slot).await; + assert_eq!(transactions.len(), 1); + assert!(transactions[0].len() > 64 * KB); + + await_replication( + &mut positions, + &follower, + expected, + state, + (BATCHED_INSTRUCTIONS - 1) as i64, + ) + .await; + assert_eq!(follower.get_account(account), leader.get_account(account)); + + dispatcher.terminate().await; + follower.close().await; + leader.close().await; +} + +/// An internally paced leader advances a follower whose delegated state survives restart. +#[tokio::test(flavor = "multi_thread")] +async fn internally_paced_replication_persists_across_restart() { + let state = Pubkey::new_unique(); + let volatile = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated), (volatile, 7, AccountMode::ReadOnly)]; + let leader_authority: Authority = Keypair::new().into(); + let follower_authority = Authority { + local: Arc::new(Keypair::new()), + remote: Some(leader_authority.local.pubkey()), + }; + let leader = engine(leader_authority, &seed, Pacing::Internal).await; + let mut follower = engine(follower_authority, &seed, Pacing::External).await; + + let addr = loopback_addr(); + let initial = follower.superblocks().position(); + let follower_identity = follower.signer().pubkey(); + let mut dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + let mut positions = stream(addr, &mut follower); + let observed = time::timeout(TIMEOUT, async { + loop { + let observed = positions.recv().await.expect("position stream is open"); + if observed > initial { + return observed; + } + } + }) + .await + .expect("follower observes an internally paced block in time"); + + let (dirs, follower_authority) = follower.close().await; + let follower = TestEngine::with(dirs, follower_authority).await; + assert!(follower.get_account(volatile).is_none()); + assert_eq!(load_v42_data(&follower, state), Some(0)); + assert!(follower.superblocks().position() >= observed); + + dispatcher.terminate().await; + follower.close().await; + leader.close().await; +} + +/// Catches up across sealed superblocks, resumes from the durable cursor without +/// re-applying, and drops volatile state on reset. +#[tokio::test(flavor = "multi_thread")] +async fn streams_and_resumes_without_duplicate_application() { + let state = Pubkey::new_unique(); + // Only reset may remove the read-only account; delegated state must survive it. + let volatile = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated), (volatile, 7, AccountMode::ReadOnly)]; + let (mut leader, mut follower) = engines(&seed, &seed).await; + + // Catch-up crosses two sealed superblocks and a live tail from byte zero. + for _ in 0..2 { + increment(&leader, state).await; + leader.seal_and_archive().await; + } + let expected = commit_increment(&mut leader, state).await; + + let upstream = loopback_addr(); + let follower_identity = follower.signer().pubkey(); + + // A valid follower signature is still rejected until its local identity is allowed. + let initial = follower.superblocks().position(); + let mut denied_dispatcher = dispatcher(upstream, &leader, &[]).await; + let mut rejected = ShutdownManager::default(); + ReplicationClient::spawn(upstream, follower.clone(), follower.pacer(), &mut rejected).unwrap(); + time::timeout(TIMEOUT, rejected.wait()) + .await + .expect("denied replication client terminates in time"); + assert_eq!(follower.superblocks().position(), initial); + rejected.terminate().await; + denied_dispatcher.terminate().await; + + let mut first_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; + let mut positions = stream(upstream, &mut follower); + + // Replay must reproduce the leader's seals and block height, not just state. + await_replication(&mut positions, &follower, expected, state, 3).await; + assert_eq!( + follower.superblocks().sealed(), + leader.superblocks().sealed() + ); + assert_eq!( + follower.blocks().latest().slot, + leader.blocks().latest().slot + ); + + let expected = commit_increment(&mut leader, state).await; + await_replication(&mut positions, &follower, expected, state, 4).await; + + // Resume after an outage from the durable cursor. Exactly-once application + // yields 5; applying the same entry twice would yield 6. + first_dispatcher.terminate().await; + let expected = commit_increment(&mut leader, state).await; + let mut second_dispatcher = dispatcher(upstream, &leader, &[follower_identity]).await; + await_replication(&mut positions, &follower, expected, state, 5).await; + + // Account creation debits the sponsor on both nodes before reset replenishes it. + let authority = leader.authority(); + let authority_before = leader.get_account(authority).expect("leader sponsor exists").lamports(); + let sponsored = Pubkey::new_unique(); + leader + .account(sponsored) + .create(v42_builder(0, AccountMode::Delegated), None) + .await + .expect("sponsored account creation commits"); + let authority_after = leader.get_account(authority).expect("leader sponsor remains").lamports(); + assert!( + authority_after < authority_before, + "account creation debits the leader sponsor" + ); + leader.advance(1).await; + let expected = leader.sync().await; + await_replication(&mut positions, &follower, expected, state, 5).await; + assert_eq!( + follower.get_account(authority).expect("follower sponsor remains").lamports(), + authority_after, + "sponsor debit replicates to the follower" + ); + + // Reset discards volatile accounts, retains delegated state, and replenishes the sponsor. + leader.reset(99).expect("leader reset records"); + assert_eq!( + leader + .get_account(authority) + .expect("leader sponsor exists after reset") + .lamports(), + authority_before, + "reset replenishes the leader sponsor" + ); + let expected = leader.sync().await; + await_replication(&mut positions, &follower, expected, state, 5).await; + assert!(follower.get_account(volatile).is_none()); + assert_eq!( + follower + .get_account(authority) + .expect("follower sponsor exists after reset") + .lamports(), + authority_before, + "reset replenishes the follower sponsor" + ); + + second_dispatcher.terminate().await; + follower.close().await; + leader.close().await; +} + +/// Models a real leader restart end to end: the connection drops, the engine +/// goes down and comes back from durable state, and the still-active follower +/// reconnects and resumes byte-exactly. An update committed just before the +/// restart is recovered, and a further update produced after it streams live — +/// each applied exactly once (the non-idempotent increment would overshoot on a +/// duplicate). +#[tokio::test(flavor = "multi_thread")] +async fn resumes_after_leader_restart() { + let state = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated)]; + let (mut leader, mut follower) = engines(&seed, &seed).await; + + let addr = loopback_addr(); + let follower_identity = follower.signer().pubkey(); + let mut first_dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + let mut positions = stream(addr, &mut follower); + + // Bring the follower current before the leader goes down. + let expected = commit_increment(&mut leader, state).await; + await_replication(&mut positions, &follower, expected, state, 1).await; + + // Take the dispatcher down, then commit an update the follower cannot see. + first_dispatcher.terminate().await; + let expected = commit_increment(&mut leader, state).await; + + // Restart the leader; the reopened engine must durably reload the update. + let (dirs, authority) = leader.close().await; + let mut leader = TestEngine::with(dirs, authority).await; + assert_eq!( + load_v42_data(&leader, state), + Some(2), + "reopened leader durably reloaded the update" + ); + + // Replication resumes on the same address; the still-active follower + // reconnects from its durable cursor and recovers the pre-restart update. + let mut second_dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + await_replication(&mut positions, &follower, expected, state, 2).await; + + // The reopened leader keeps producing: a new update streams live over the + // resumed connection, proving block production continues from the durable + // slot rather than restarting and stalling the follower. + let expected = commit_increment(&mut leader, state).await; + await_replication(&mut positions, &follower, expected, state, 3).await; + + second_dispatcher.terminate().await; + follower.close().await; + leader.close().await; +} + +/// Installs the newest retained snapshot on restart, then streams every durable +/// ledger entry committed after that snapshot. +#[tokio::test(flavor = "multi_thread")] +async fn restores_the_newest_snapshot_then_streams_its_tail() { + let state = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated)]; + // The empty follower can acquire `state` only through snapshot restoration. + let (mut leader, follower) = engines(&seed, &[]).await; + let set = |value| E::lit(value).compose(state, &[]); + + // Distinct values distinguish the newest snapshot from its streamed tail. + for value in [10, 20] { + leader.execute(&[set(value)]).await.unwrap(); + leader.seal_and_archive().await; + } + leader.execute(&[set(30)]).await.unwrap(); + leader.advance(1).await; + let expected = leader.sync().await; + // Make the newest snapshot the only possible handshake response. + leader.ledger().truncate().unwrap(); // superblock 0 + leader.ledger().truncate().unwrap(); // superblock 1 + assert!( + leader.ledger().cursor(0).is_none(), + "follower cursor was retained away" + ); + assert!( + leader.ledger().cursor(1).is_none(), + "older snapshot history was retained away" + ); + + let addr = loopback_addr(); + let follower_identity = follower.signer().pubkey(); + let mut dispatcher = dispatcher(addr, &leader, &[follower_identity]).await; + let mut follower = restart_from_snapshot(addr, follower).await; + assert_eq!( + load_v42_data(&follower, state), + Some(20), + "restart restores the newest completed snapshot, not the post-snapshot tail" + ); + + let mut positions = stream(addr, &mut follower); + await_replication(&mut positions, &follower, expected, state, 30).await; + + dispatcher.terminate().await; + follower.close().await; + leader.close().await; +} + +/// Replicated state is itself replicable: a follower seals and archives from +/// replicated blocks alone, and its reconstruction serves a further follower +/// both as a live stream and as a snapshot bootstrap. +#[tokio::test(flavor = "multi_thread")] +async fn cascades_replication_through_a_follower() { + let state = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated)]; + let shared = Arc::new(Keypair::new()); + let leader_authority: Authority = shared.clone().into(); + let middle_authority = Authority { + local: shared.clone(), + remote: Some(shared.pubkey()), + }; + let tail_authority = Authority { + local: Arc::new(Keypair::new()), + remote: Some(shared.pubkey()), + }; + let mut leader = engine(leader_authority, &seed, Pacing::External).await; + let mut middle = engine(middle_authority, &seed, Pacing::External).await; + let tail = engine(tail_authority, &[], Pacing::External).await; + + let leader_addr = loopback_addr(); + let middle_addr = loopback_addr(); + let middle_identity = middle.signer().pubkey(); + let tail_identity = tail.signer().pubkey(); + let mut leader_dispatcher = dispatcher(leader_addr, &leader, &[middle_identity]).await; + let mut middle_positions = stream(leader_addr, &mut middle); + let mut middle_dispatcher = dispatcher(middle_addr, &middle, &[tail_identity]).await; + + // Subscribe before the boundary: the seal is driven by replication, so + // `await_archive` would subscribe after the archiver it means to observe. + let mut archives = middle.accounts().subscribe_snapshots(); + + // Cross the seal live rather than during handshake catch-up. + increment(&leader, state).await; + leader.seal_and_archive().await; + let expected = leader.sync().await; + await_replication(&mut middle_positions, &middle, expected, state, 1).await; + assert_eq!( + middle.superblocks().sealed(), + leader.superblocks().sealed(), + "middle sealed the replicated boundary to the leader's checksum" + ); + time::timeout(TIMEOUT, archives.recv()) + .await + .expect("middle archives its replicated seal in time") + .unwrap(); + + // Keep the archive behind live state so restore and streaming are distinguishable. + let expected = commit_increment(&mut leader, state).await; + await_replication(&mut middle_positions, &middle, expected, state, 2).await; + + // The successor archive must become the only answer to the tail's cursor. + middle.ledger().truncate().unwrap(); // superblock 0 + assert!( + middle.ledger().cursor(0).is_none(), + "tail cursor was retained away" + ); + + let mut tail = restart_from_snapshot(middle_addr, tail).await; + assert_eq!( + load_v42_data(&tail, state), + Some(1), + "tail restores the middle's own archive, not the state streamed past it" + ); + + let mut tail_positions = stream(middle_addr, &mut tail); + await_replication(&mut tail_positions, &tail, expected, state, 2).await; + + // Both hops cross the next seal live. + increment(&leader, state).await; + leader.seal_and_archive().await; + let expected = leader.sync().await; + await_replication(&mut middle_positions, &middle, expected, state, 3).await; + await_replication(&mut tail_positions, &tail, expected, state, 3).await; + assert_eq!(middle.superblocks().sealed(), leader.superblocks().sealed()); + assert_eq!(tail.superblocks().sealed(), leader.superblocks().sealed()); + + leader_dispatcher.terminate().await; + middle_dispatcher.terminate().await; + tail.close().await; + middle.close().await; + leader.close().await; +} diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 48701262..bf5eb1a8 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,4 +1,4 @@ [toolchain] -channel = "1.94.1" +channel = "1.96.1" components = ["clippy", "rustfmt"] profile = "minimal"