From 3217eacdd35bf0843ed037cda32855c59d520ab6 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 21 Aug 2026 16:21:46 +0400 Subject: [PATCH 1/4] perf: batch replication signature verification --- Cargo.lock | 16 ++ engine/Cargo.toml | 1 + engine/README.md | 6 + engine/src/accessor.rs | 9 +- engine/src/error.rs | 5 +- engine/src/lib.rs | 7 +- engine/src/transaction.rs | 72 ++++++- replicator/Cargo.toml | 1 + replicator/README.md | 7 + replicator/src/client.rs | 333 ++++++++++++++++++++++++++------ replicator/tests/integration.rs | 65 +++++++ 11 files changed, 446 insertions(+), 76 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5339bba5..3fd04ba6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1265,6 +1265,7 @@ checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" dependencies = [ "curve25519-dalek 4.1.3", "ed25519 2.2.3", + "merlin", "rand_core 0.6.4", "serde", "sha2 0.10.9", @@ -2337,6 +2338,7 @@ dependencies = [ "solana-program-runtime", "solana-pubkey", "solana-sdk-ids", + "solana-signature", "solana-signer", "solana-system-interface", "solana-system-program", @@ -2486,6 +2488,7 @@ name = "magicblock-replicator" version = "0.2.0" dependencies = [ "derive_more", + "flume", "magicblock-engine", "magicblock-engine-nucleus", "magicblock-keeper", @@ -2534,6 +2537,18 @@ dependencies = [ "libc", ] +[[package]] +name = "merlin" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "58c38e2799fc0978b65dfff8023ec7843e2330bb462f19198840b34b6582397d" +dependencies = [ + "byteorder", + "keccak", + "rand_core 0.6.4", + "zeroize", +] + [[package]] name = "mio" version = "1.2.2" @@ -4283,6 +4298,7 @@ version = "3.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b0364c7577c3c82a693ce28a1febc8d1b5d1b0a175fdc2114ae6186b69effe1e" dependencies = [ + "curve25519-dalek 4.1.3", "ed25519-dalek 2.2.0", "five8", "rand 0.9.5", diff --git a/engine/Cargo.toml b/engine/Cargo.toml index 00dc5f68..a2c58377 100644 --- a/engine/Cargo.toml +++ b/engine/Cargo.toml @@ -41,6 +41,7 @@ solana-program-runtime = { workspace = true } solana-pubkey = { workspace = true } solana-sdk-ids = { workspace = true } solana-signer = { workspace = true } +solana-signature = { workspace = true, features = ["batch-verify"] } solana-system-program = { workspace = true, features = ["agave-unstable-api"] } solana-transaction = { workspace = true, features = ["wincode"] } diff --git a/engine/README.md b/engine/README.md index c68ac2e4..95ff11e5 100644 --- a/engine/README.md +++ b/engine/README.md @@ -10,6 +10,12 @@ configured remote authority for a replica, or the local identity when no override is configured. Replication uses that distinction to sign locally while authenticating its immediate upstream. +Normal transaction submission sanitizes and verifies each transaction. +Replication instead uses `Engine::verifier` to sanitize, authority-check, and +batch-verify payloads, then consumes the resulting opaque transactions through +the trusted `TransactionAccessor::verified` path without repeating crypto. +Retained local-ledger replay has a separate private verification bypass. + ## Account replacement `AccountAccessor::{create, update}` composes complete-account MagicRoot patch diff --git a/engine/src/accessor.rs b/engine/src/accessor.rs index 2723bf43..7b56f03d 100644 --- a/engine/src/accessor.rs +++ b/engine/src/accessor.rs @@ -14,7 +14,7 @@ use tokio::time; use crate::{ Engine, IntoTransactionView, error::{EngineError, Result}, - transaction, + transaction::{self, VerifiedTransaction}, }; /// Upper bound on awaiting a submitted transaction's committed result. @@ -75,6 +75,13 @@ impl<'a> TransactionAccessor<'a> { Ok(Self { engine, transaction }) } + /// Enters the trusted replication path without repeating signature verification. + /// + /// The caller must only pass values produced by this Engine's verifier. + pub fn verified(engine: &'a Engine, verified: VerifiedTransaction) -> Self { + Self { engine, transaction: verified.0 } + } + /// Submits `transaction` for execution and awaits its committed result. /// A timeout does not cancel the submitted transaction. pub async fn execute(self) -> Result> { diff --git a/engine/src/error.rs b/engine/src/error.rs index 0e297bee..d0d20c7d 100644 --- a/engine/src/error.rs +++ b/engine/src/error.rs @@ -4,7 +4,7 @@ use agave_transaction_view::result::TransactionViewError; use derive_more::From; use keeper::error::KeeperError; use ledger::{LedgerError, LedgerRequestError}; -use nucleus::shutdown::Service; +use nucleus::{Slot, shutdown::Service}; use processor::ProcessorError; use solana_message::CompileError; use solana_transaction::{InstructionError, SignerError, TransactionError}; @@ -79,6 +79,9 @@ pub enum ReplayError { /// A retained transaction could not be sanitized into a transaction view. #[error("transaction sanitization: {0:?}")] Sanitization(TransactionViewError), + /// A replicated block boundary disagreed with the locally produced block hash. + #[error("replicated block hash mismatch at slot {0}")] + BlockhashMismatch(Slot), /// The replayed account state checksum diverged from the sealed superblock. #[error("replayed state checksum mismatch")] StateMismatch, diff --git a/engine/src/lib.rs b/engine/src/lib.rs index bd85dab0..58702cab 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -36,7 +36,7 @@ pub mod testkit; pub use accessor::{AccountAccessor, TransactionAccessor}; pub use error::{EngineError, ReplayError, Result}; -pub use transaction::IntoTransactionView; +pub use transaction::{IntoTransactionView, TransactionVerifier, VerifiedTransaction}; use crate::pacemaker::{ExternalPacer, PaceMaker}; @@ -133,6 +133,11 @@ impl Engine { Ok(TransactionAccessor { engine: self, transaction }) } + /// Returns an authority-bound verifier for replicated transaction batches. + pub fn verifier(&self) -> TransactionVerifier { + TransactionVerifier::new(self.authority()) + } + /// Drains in-flight execution and keeps the sequencer paused until the handle is dropped. pub async fn barrier(&self) -> Result { let (controller, guard) = runtime::barrier(); diff --git a/engine/src/transaction.rs b/engine/src/transaction.rs index 4d1a00ef..3c32ebe2 100644 --- a/engine/src/transaction.rs +++ b/engine/src/transaction.rs @@ -10,11 +10,50 @@ use solana_message::{ VersionedMessage, v1::{self, SIGNATURE_SIZE}, }; +use solana_pubkey::Pubkey; +use solana_signature::Signature; use solana_signer::Signer; use solana_transaction::{Message, Transaction, TransactionError, versioned::VersionedTransaction}; use crate::{Engine, error::EngineError, error::Result}; +/// Opaque transaction admitted through the replication verifier trust boundary. +pub struct VerifiedTransaction(pub(crate) TransactionView); + +/// Engine-authorized batch verifier for replicated transaction payloads. +#[derive(Clone, Copy)] +pub struct TransactionVerifier { + authority: Pubkey, +} + +impl TransactionVerifier { + pub(crate) fn new(authority: Pubkey) -> Self { + Self { authority } + } + + /// Sanitizes, validates, and batch-verifies every transaction atomically. + pub fn verify(&self, transactions: Vec>) -> Result> { + let verified = transactions + .into_iter() + .map(|transaction| { + let view = TransactionView::try_new_sanitized(transaction.into(), true)?; + validate_authority(&view, self.authority)?; + Ok(VerifiedTransaction(view)) + }) + .collect::>>()?; + + let mut signatures = Vec::with_capacity(verified.len()); + for transaction in &verified { + signatures.extend(signature_data(&transaction.0)); + } + if !Signature::batch_verify(signatures.into_iter()) { + return Err(EngineError::SignatureVerification); + } + + Ok(verified) + } +} + /// Conversion of anything composable into an executable /// transaction into a sanitized [`TransactionView`]. pub trait IntoTransactionView { @@ -59,24 +98,39 @@ impl IntoTransactionView for Vec { impl IntoTransactionView for TransactionView { fn compose(self, engine: &Engine) -> Result { - if matches!(self.version(), TransactionVersion::Magicblock) - && self.static_account_keys()[0] != engine.authority() - { - return Err(EngineError::SignatureVerification); - } + validate_authority(&self, engine.authority())?; Ok(self) } } +/// Enforces the authority encoded by private Magicblock transactions. +fn validate_authority(view: &TransactionView, authority: Pubkey) -> Result<()> { + if matches!(view.version(), TransactionVersion::Magicblock) + && view.static_account_keys()[0] != authority + { + return Err(EngineError::SignatureVerification); + } + Ok(()) +} + +/// Iterates each signature with its signer key and shared serialized message. +fn signature_data( + view: &TransactionView, +) -> impl ExactSizeIterator { + let message = view.message_data(); + view.signatures() + .iter() + .zip(view.static_account_keys()) + .map(move |(signature, key)| (signature, key.as_ref(), message)) +} + /// The engine's sole signature-verification point. /// /// Every public transaction accessor verifies here; trusted local replay is /// the only bypass. TODO: Remove the bypass before replaying untrusted ledgers. pub(super) fn sigverify(view: &TransactionView) -> Result<()> { - // Sanitization guarantees one static key for every required signature. - let message = view.message_data(); - for (signature, key) in view.signatures().iter().zip(view.static_account_keys()) { - if !signature.verify(key.as_ref(), message) { + for (signature, key, message) in signature_data(view) { + if !signature.verify(key, message) { return Err(EngineError::SignatureVerification); } } diff --git a/replicator/Cargo.toml b/replicator/Cargo.toml index fbd20941..475ff703 100644 --- a/replicator/Cargo.toml +++ b/replicator/Cargo.toml @@ -19,6 +19,7 @@ ledger = { workspace = true } nucleus = { workspace = true, features = ["ledger", "service"] } derive_more = { workspace = true, features = ["from"] } +flume = { workspace = true } scc = { workspace = true } snedfile = { workspace = true } thiserror = { workspace = true } diff --git a/replicator/README.md b/replicator/README.md index 9f6fae63..ba7e2c24 100644 --- a/replicator/README.md +++ b/replicator/README.md @@ -50,6 +50,13 @@ 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. +The follower uses persistent ingest and control threads joined at shutdown. +Ingest decodes transaction batches of at most 128 transactions and typically +128 KiB, fencing them before every block, superblock, reset, or reconnect. +A rendezvous channel assigns verification to an idle control thread; otherwise +ingest verifies while control schedules earlier work. Control alone advances +the engine and block pacer, preserving stream order without reordering state. + 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 diff --git a/replicator/src/client.rs b/replicator/src/client.rs index 687e26a4..df650578 100644 --- a/replicator/src/client.rs +++ b/replicator/src/client.rs @@ -2,11 +2,16 @@ use std::{ fs::{self, File}, io::{self, BufReader, Read}, net::{SocketAddr, TcpStream}, + sync::mpsc, thread, }; use derive_more::Deref; -use engine::{Engine, EngineError, ReplayError, pacemaker::ExternalBlock}; +use engine::{ + Engine, EngineError, ReplayError, TransactionAccessor, VerifiedTransaction, + pacemaker::ExternalBlock, +}; +use flume::{Sender, TrySendError}; use ledger::{ Superblock, schema::{Block, OwnedBlockstoreEntry, blockstore}, @@ -14,11 +19,12 @@ use ledger::{ use nucleus::{ KB, ledger::{ACCOUNTSDB_SNAPSHOT_FILE, BlockstorePosition}, - shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + runtime::BarrierHandle, + shutdown::{CancellationToken, Service, ShutdownHandle, ShutdownManager, ShutdownReason}, }; use tokio::{ runtime, - sync::mpsc::{Receiver, Sender}, + sync::mpsc::{Receiver as BlockReceiver, Sender as PacerSender}, time, }; use tracing::{error, info, warn}; @@ -32,6 +38,66 @@ use crate::{ }; type ReplicationStream = BufReader; +type ReconnectReply = mpsc::SyncSender; + +const MAX_BATCH_TRANSACTIONS: usize = 128; +const MAX_BATCH_BYTES: usize = 128 * KB; + +/// Consecutive transaction payloads accumulated between ordered stream fences. +struct TransactionsBatch { + transactions: Vec>, + bytes: usize, +} + +impl Default for TransactionsBatch { + fn default() -> Self { + Self { + transactions: Vec::with_capacity(MAX_BATCH_TRANSACTIONS), + bytes: 0, + } + } +} + +impl TransactionsBatch { + fn is_empty(&self) -> bool { + self.transactions.is_empty() + } + + fn push(&mut self, transaction: Vec) { + self.bytes = self.bytes.saturating_add(transaction.len()); + self.transactions.push(transaction); + } + + fn is_full(&self) -> bool { + self.transactions.len() >= MAX_BATCH_TRANSACTIONS || self.bytes >= MAX_BATCH_BYTES + } + + fn take(&mut self) -> Vec> { + self.bytes = 0; + std::mem::replace( + &mut self.transactions, + Vec::with_capacity(MAX_BATCH_TRANSACTIONS), + ) + } +} + +/// Ordered handoff from blocking stream ingest to asynchronous Engine control. +enum ReplicationMessage { + Unverified(Vec>), + Verified(engine::Result>), + Entry(OwnedBlockstoreEntry), + Connected, + Disconnected(ReconnectReply), +} + +/// Owns stream decoding, bounded transaction accumulation, and opportunistic verification. +struct Ingest { + engine: Engine, + addr: SocketAddr, + batch: TransactionsBatch, + tx: Sender, + shutdown: CancellationToken, +} /// Pulls a leader blockstore stream into an externally paced follower engine. #[derive(Deref)] @@ -42,9 +108,9 @@ pub struct ReplicationClient { /// Leader endpoint reused after transport loss. addr: SocketAddr, /// External pacemaker channel used to preserve block-boundary ordering. - pacer: Sender, + pacer: PacerSender, /// Locally committed block boundaries used to verify replicated output. - blocks: Receiver, + blocks: BlockReceiver, } impl ReplicationClient { @@ -52,7 +118,7 @@ impl ReplicationClient { pub fn spawn( addr: SocketAddr, engine: Engine, - pacer: Sender, + pacer: PacerSender, shutdown: &mut ShutdownManager, ) -> Result<()> { metrics::init(); @@ -64,13 +130,13 @@ impl ReplicationClient { 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)))?; + .spawn(move || rt.block_on(client.serve(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; + async fn serve(self, mut shutdown: ShutdownHandle) { + let result = self.run(&shutdown).await; if shutdown.requested() || result.is_ok() { shutdown.terminate(ShutdownReason::Signalled); return; @@ -87,30 +153,89 @@ impl ReplicationClient { } } - /// 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(); + /// Starts Ingest and joins it after Control stops consuming its ordered messages. + async fn run(self, shutdown: &ShutdownHandle) -> Result<()> { + let (guard, position) = self.resume().await?; + let (tx, rx) = flume::bounded(0); + let mut ingest = Ingest { + engine: self.engine.clone(), + addr: self.addr, + batch: Default::default(), + tx, + shutdown: shutdown.child(), + }; + let rt = runtime::Builder::new_current_thread().enable_time().build()?; + let ingest = thread::Builder::new() + .name("replication-ingest".into()) + .spawn(move || rt.block_on(ingest.run(position)))?; + let mut result = self.consume(shutdown, rx, guard).await; + match ingest.join() { + Ok(Ok(())) => info!("replication ingest has gracefully shutdown"), + Ok(Err(error)) => result = result.and(Err(error)), + Err(error) => error!(?error, "replication ingest panicked"), + } + result + } + + /// Consumes ordered Ingest messages until shutdown or a terminal failure. + async fn consume( + mut self, + shutdown: &ShutdownHandle, + rx: flume::Receiver, + guard: BarrierHandle, + ) -> Result<()> { + let verifier = self.verifier(); + let mut connected = None; + let mut barrier = Some(guard); + 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(); + // Complete a ready handoff before observing concurrent cancellation. + let message = tokio::select! { + biased; + message = rx.recv_async() => match message { + Ok(m) => m, + // Ingest has shutdown, the potential error will be captured by caller + Err(_) => return Ok(()), + }, + _ = shutdown.signalled() => break, + + }; + match message { + ReplicationMessage::Unverified(batch) => { + let verified = verifier.verify(batch)?; + self.schedule(verified).await?; + } + ReplicationMessage::Verified(result) => self.schedule(result?).await?, + ReplicationMessage::Entry(entry) => self.process(entry).await?, + ReplicationMessage::Connected => { + connected = Some(metrics::client_connection()); + barrier.take(); + } + ReplicationMessage::Disconnected(reply) => { + connected.take(); + let (guard, position) = self.resume().await?; + barrier = Some(guard); + if reply.send(position).is_err() { + return Err(ReplicationError::StreamClosed); + } } - Err(error) => Err(wincode::Error::from(error))?, } } + Ok(()) + } + + /// Schedules a verified batch in stream order without repeating admission checks. + async fn schedule(&self, transactions: Vec) -> Result<()> { + for transaction in transactions { + TransactionAccessor::verified(&self.engine, transaction).schedule().await?; + } + Ok(()) } - /// Applies one blockstore entry to the follower engine, holding block-boundary - /// ordering through the pacemaker and flagging superblock seal mismatches. + /// Applies one control entry after all preceding transactions are scheduled. async fn process(&mut self, entry: OwnedBlockstoreEntry) -> Result<()> { match entry { OwnedBlockstoreEntry::Block(block) => { @@ -119,8 +244,8 @@ impl ReplicationClient { 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"); + let error = ReplayError::BlockhashMismatch(block.slot); + Err(EngineError::from(error))?; } guard.await.map_err(EngineError::from)?; } @@ -133,24 +258,133 @@ impl ReplicationClient { Err(EngineError::Replay(ReplayError::StateMismatch))?; } } - entry => self.engine.replay(entry).await?, + OwnedBlockstoreEntry::Reset(slot) => { + self.engine.replay(OwnedBlockstoreEntry::Reset(slot)).await?; + } + OwnedBlockstoreEntry::Transaction(_) => (), + } + Ok(()) + } + + /// Flushes prior work and returns its durable cursor under a sequencing barrier. + async fn resume(&self) -> Result<(BarrierHandle, BlockstorePosition)> { + let guard = self.barrier().await?; + self.sync(false)?; + Ok((guard, self.superblocks().position())) + } +} + +impl Ingest { + /// Decodes the stream while preserving the order of transactions and control entries. + async fn run(&mut self, position: BlockstorePosition) -> Result<()> { + let mut stream = self.open(position).await?; + while !self.shutdown.is_cancelled() { + match blockstore::decode(&mut stream) { + Ok(OwnedBlockstoreEntry::Transaction(transaction)) => { + if !self.push(transaction) { + return Ok(()); + } + } + Ok(entry) => { + if !self.flush() { + return Ok(()); + } + self.tx + .send(ReplicationMessage::Entry(entry)) + .map_err(|_| ReplicationError::StreamClosed)?; + } + Err(wincode::error::ReadError::Io(error)) => { + warn!(%error, "replication stream disconnected"); + if !self.flush() { + return Ok(()); + } + stream = self.open(self.request_position()?).await?; + } + Err(error) => { + if !self.flush() { + return Ok(()); + } + return Err(wincode::Error::from(error).into()); + } + } } Ok(()) } - /// Handshakes with the leader at `position`; either stages a snapshot and - /// signals a required restart, or returns the resumed byte stream. + /// Adds a transaction and flushes once either batch bound is reached. + fn push(&mut self, transaction: Vec) -> bool { + self.batch.push(transaction); + !self.batch.is_full() || self.flush() + } + + /// Offers the batch to Control, verifying it locally when Control is occupied. + fn flush(&mut self) -> bool { + if self.batch.is_empty() { + return true; + } + let batch = self.batch.take(); + match self.tx.try_send(ReplicationMessage::Unverified(batch)) { + Ok(()) => true, + Err(TrySendError::Full(ReplicationMessage::Unverified(batch))) => { + let result = self.engine.verifier().verify(batch); + let valid = result.is_ok(); + self.tx.send(ReplicationMessage::Verified(result)).is_ok() && valid + } + Err(_) => false, + } + } + + /// Requests a durable resume cursor after Control finishes all preceding work. + fn request_position(&self) -> Result { + let (reply, response) = mpsc::sync_channel(0); + let _ = self.tx.send(ReplicationMessage::Disconnected(reply)); + response.recv().map_err(|_| ReplicationError::StreamClosed) + } + + /// Reconnects from `position` and tells Control it may release the barrier. + async fn open(&self, position: BlockstorePosition) -> Result { + let stream = self.reconnect(position).await?; + let _ = self.tx.send(ReplicationMessage::Connected); + Ok(stream) + } + + /// Retries transport establishment while the ordered resume cursor remains quiesced. + async fn reconnect(&self, position: BlockstorePosition) -> Result { + for attempt in 1..=MAX_RECONNECT_ATTEMPTS { + if self.shutdown.is_cancelled() { + 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, self.shutdown.cancelled()).await.is_ok() { + Err(ReplicationError::StreamClosed)?; + } + } + Err(ReplicationError::ReconnectExhausted) + } + + /// Handshakes at `position`, staging a snapshot when streaming cannot resume. 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)?; + let handshake = Handshake::new(self.engine.signer(), request)?; protocol::write(&mut connection, &handshake)?; let handshake = protocol::read::>(&mut connection)?; handshake.verify()?; - let expected = self.authority(); + let expected = self.engine.authority(); if handshake.identity != expected { let message = format!( "unexpected replication server identity {}; expected {expected}", @@ -172,41 +406,12 @@ impl ReplicationClient { } } - /// 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. + /// Stages a complete snapshot and installs its seal for the requested restart. 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 superblocks = self.engine.superblocks(); + let dir = Superblock::init_dir(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)?; @@ -217,7 +422,7 @@ impl ReplicationClient { file.sync_all()?; drop(file); fs::rename(temporary, archive)?; - self.superblocks().bootstrap(meta.superblock)?; + superblocks.bootstrap(meta.superblock)?; info!(?meta, "replication snapshot staged"); Ok(()) } diff --git a/replicator/tests/integration.rs b/replicator/tests/integration.rs index 0a1aa689..6df66dbf 100644 --- a/replicator/tests/integration.rs +++ b/replicator/tests/integration.rs @@ -263,6 +263,71 @@ async fn replays_large_transactions_during_catch_up_and_live_streaming() { leader.close().await; } +/// Proves a 128-plus-one catch-up batch preserves transaction order and its block fence. +#[tokio::test(flavor = "multi_thread")] +async fn batches_transactions_without_crossing_block_boundaries() { + const CATCH_UP_TRANSACTIONS: i64 = 129; + + let state = Pubkey::new_unique(); + let seed = [(state, 0, AccountMode::Delegated)]; + let (mut leader, mut follower) = engines(&seed, &seed).await; + + // Distinct assignments make any cross-batch reordering observable in final state. + for value in 1..=CATCH_UP_TRANSACTIONS { + let instruction = E::lit(value).compose(state, &[]); + leader.schedule(&[instruction]).await; + } + let catch_up_slot = leader.blocks().current_slot(); + leader.advance(1).await; + let expected = leader.sync().await; + let expected_transactions = block_transactions(&leader, catch_up_slot).await; + assert_eq!( + expected_transactions.len(), + CATCH_UP_TRANSACTIONS as usize, + "the stream crosses the transaction-count batch limit exactly once" + ); + + 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, + CATCH_UP_TRANSACTIONS, + ) + .await; + assert_eq!( + block_transactions(&follower, catch_up_slot).await, + expected_transactions, + "the 128-plus-one split retains byte-exact ledger order" + ); + + // A following live transaction must remain behind the catch-up block fence. + let live_value = CATCH_UP_TRANSACTIONS + 1; + let instruction = E::lit(live_value).compose(state, &[]); + leader.schedule(&[instruction]).await; + let live_slot = leader.blocks().current_slot(); + leader.advance(1).await; + let expected = leader.sync().await; + await_replication(&mut positions, &follower, expected, state, live_value).await; + assert_eq!( + block_transactions(&follower, live_slot).await, + block_transactions(&leader, live_slot).await, + "the live tail starts in the block after the catch-up fence" + ); + assert_eq!( + follower.ledger().transactions(), + leader.ledger().transactions() + ); + + 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() { From 6360dffe6ca32c23bff813211cab40c58dfcca30 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 21 Aug 2026 17:10:49 +0400 Subject: [PATCH 2/4] fix(replication): make reconnection race free --- replicator/README.md | 7 +++++-- replicator/src/client.rs | 24 ++++++++++++++++++++++-- replicator/src/error.rs | 3 +++ replicator/src/protocol.rs | 2 ++ replicator/src/server.rs | 10 ++++++---- 5 files changed, 38 insertions(+), 8 deletions(-) diff --git a/replicator/README.md b/replicator/README.md index ba7e2c24..f5a1116e 100644 --- a/replicator/README.md +++ b/replicator/README.md @@ -54,8 +54,11 @@ The follower uses persistent ingest and control threads joined at shutdown. Ingest decodes transaction batches of at most 128 transactions and typically 128 KiB, fencing them before every block, superblock, reset, or reconnect. A rendezvous channel assigns verification to an idle control thread; otherwise -ingest verifies while control schedules earlier work. Control alone advances -the engine and block pacer, preserving stream order without reordering state. +`Ingest::flush` verifies while Control schedules earlier work. Control alone +schedules verified transactions and advances the block pacer. During the +Control-held handshake, `Ingest::stage_snapshot` may write the snapshot archive +and bootstrap durable superblock state. These roles preserve stream order +without reordering state. A shared-key follower may also serve downstream followers. It derives and validates superblock seals from replicated block boundaries and archives its own diff --git a/replicator/src/client.rs b/replicator/src/client.rs index df650578..79c70d69 100644 --- a/replicator/src/client.rs +++ b/replicator/src/client.rs @@ -40,12 +40,16 @@ use crate::{ type ReplicationStream = BufReader; type ReconnectReply = mpsc::SyncSender; +/// Maximum transactions retained before offering a batch to Control. const MAX_BATCH_TRANSACTIONS: usize = 128; +/// Maximum transaction payload bytes retained before offering a batch to Control. const MAX_BATCH_BYTES: usize = 128 * KB; /// Consecutive transaction payloads accumulated between ordered stream fences. struct TransactionsBatch { + /// Raw transaction payloads in stream order. transactions: Vec>, + /// Cumulative payload bytes used to enforce the batch bound. bytes: usize, } @@ -83,19 +87,29 @@ impl TransactionsBatch { /// Ordered handoff from blocking stream ingest to asynchronous Engine control. enum ReplicationMessage { + /// Raw batch offered to Control for signature verification. Unverified(Vec>), + /// Verification result completed by Ingest while Control was occupied. Verified(engine::Result>), + /// Control entry fenced behind every preceding transaction batch. Entry(OwnedBlockstoreEntry), + /// Successful handshake allowing Control to release its sequencing barrier. Connected, + /// Lost stream requesting Control's next durable resume position. Disconnected(ReconnectReply), } /// Owns stream decoding, bounded transaction accumulation, and opportunistic verification. struct Ingest { + /// Engine used for batch verification and authenticated stream recovery. engine: Engine, + /// Upstream replication endpoint reused across reconnects. addr: SocketAddr, + /// Consecutive transactions awaiting an ordered handoff. batch: TransactionsBatch, + /// Rendezvous sender preserving ingest-to-Control message order. tx: Sender, + /// Cancellation scoped to the ingest worker lifecycle. shutdown: CancellationToken, } @@ -295,10 +309,12 @@ impl Ingest { } Err(wincode::error::ReadError::Io(error)) => { warn!(%error, "replication stream disconnected"); + drop(stream); if !self.flush() { return Ok(()); } - stream = self.open(self.request_position()?).await?; + let position = self.request_position()?; + stream = self.open(position).await?; } Err(error) => { if !self.flush() { @@ -342,7 +358,7 @@ impl Ingest { } /// Reconnects from `position` and tells Control it may release the barrier. - async fn open(&self, position: BlockstorePosition) -> Result { + async fn open(&mut self, position: BlockstorePosition) -> Result { let stream = self.reconnect(position).await?; let _ = self.tx.send(ReplicationMessage::Connected); Ok(stream) @@ -363,6 +379,9 @@ impl Ingest { Err(ReplicationError::IO(error)) => { warn!(attempt, ?error, "replication reconnect failed"); } + Err(ReplicationError::StreamActive) => { + warn!(attempt, "previous replication stream is still active"); + } Err(error) => return Err(error), } let timeout = RETRY_DELAY * attempt as u32; @@ -403,6 +422,7 @@ impl Ingest { Ok(BufReader::with_capacity(256 * KB, connection)) } HandshakeResponse::Err(message) => Err(ReplicationError::Handshake(message)), + HandshakeResponse::StreamActive => Err(ReplicationError::StreamActive), } } diff --git a/replicator/src/error.rs b/replicator/src/error.rs index 24e48b33..5c8a56d8 100644 --- a/replicator/src/error.rs +++ b/replicator/src/error.rs @@ -33,6 +33,9 @@ pub enum ReplicationError { /// The leader rejected the client's handshake. #[error("replication handshake rejected: {0}")] Handshake(String), + /// The follower identity still owns an earlier replication stream. + #[error("replication stream already active")] + StreamActive, /// The snapshot connection ended before the advertised byte count arrived. #[error("incomplete replication snapshot: expected {0} bytes, received {1}")] Snapshot(u64, u64), diff --git a/replicator/src/protocol.rs b/replicator/src/protocol.rs index 846a545a..51d0285f 100644 --- a/replicator/src/protocol.rs +++ b/replicator/src/protocol.rs @@ -54,6 +54,8 @@ pub(crate) enum HandshakeResponse { Stream(BlockstorePosition), /// Reason the leader rejected negotiation. Err(String), + /// The follower identity still owns an earlier stream. + StreamActive, } /// Describes the accountsdb snapshot a follower must stage before it can stream. diff --git a/replicator/src/server.rs b/replicator/src/server.rs index 7000a76b..39dc8d1a 100644 --- a/replicator/src/server.rs +++ b/replicator/src/server.rs @@ -179,7 +179,10 @@ impl ReplicationServer { Ok(handshake) => handshake, Err(error) => { warn!(?error, "replication handshake rejected"); - let response = HandshakeResponse::Err(error.to_string()); + let response = match error { + ReplicationError::StreamActive => HandshakeResponse::StreamActive, + other => HandshakeResponse::Err(other.to_string()), + }; self.respond(response)?; return Ok(()); } @@ -239,10 +242,10 @@ impl ReplicationServer { 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 lease = self.reserve(handshake.identity)?; let requested = handshake.payload.position; if requested > self.position.current { @@ -321,8 +324,7 @@ impl ReplicationServer { 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())); + return Err(ReplicationError::StreamActive); } Ok(entry.get().clone()) } From f2e8471bc8818faa56a650d8829c3ffe8da89f01 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 21 Aug 2026 23:16:55 +0400 Subject: [PATCH 3/4] fix: remove stale README doc --- replicator/README.md | 1 - 1 file changed, 1 deletion(-) diff --git a/replicator/README.md b/replicator/README.md index f5a1116e..d5a57907 100644 --- a/replicator/README.md +++ b/replicator/README.md @@ -50,7 +50,6 @@ 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. -The follower uses persistent ingest and control threads joined at shutdown. Ingest decodes transaction batches of at most 128 transactions and typically 128 KiB, fencing them before every block, superblock, reset, or reconnect. A rendezvous channel assigns verification to an idle control thread; otherwise From 3d897fc68620cc837983b755ddc50ea699140505 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 21 Aug 2026 23:27:12 +0400 Subject: [PATCH 4/4] fix(replication): cancel ingest before join --- replicator/src/client.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/replicator/src/client.rs b/replicator/src/client.rs index 79c70d69..575912a8 100644 --- a/replicator/src/client.rs +++ b/replicator/src/client.rs @@ -171,18 +171,20 @@ impl ReplicationClient { async fn run(self, shutdown: &ShutdownHandle) -> Result<()> { let (guard, position) = self.resume().await?; let (tx, rx) = flume::bounded(0); + let cancellation = shutdown.child(); let mut ingest = Ingest { engine: self.engine.clone(), addr: self.addr, batch: Default::default(), tx, - shutdown: shutdown.child(), + shutdown: cancellation.clone(), }; let rt = runtime::Builder::new_current_thread().enable_time().build()?; let ingest = thread::Builder::new() .name("replication-ingest".into()) .spawn(move || rt.block_on(ingest.run(position)))?; let mut result = self.consume(shutdown, rx, guard).await; + cancellation.cancel(); match ingest.join() { Ok(Ok(())) => info!("replication ingest has gracefully shutdown"), Ok(Err(error)) => result = result.and(Err(error)),