perf: batch replication signature verification

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"] }
 

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

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<TransactionResult<()>> {

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,

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<BarrierHandle> {
         let (controller, guard) = runtime::barrier();

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<Vec<u8>>) -> Result<Vec<VerifiedTransaction>> {
+        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::<Result<Vec<_>>>()?;
+
+        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<u8> {
 
 impl IntoTransactionView for TransactionView {
     fn compose(self, engine: &Engine) -> Result<TransactionView> {
-        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<Item = (&Signature, &[u8], &[u8])> {
+    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);
         }
     }

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 }

replicator/README.md

@@ -50,6 +50,15 @@ 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.
 
+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::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
 snapshots, while downstream clients continue to verify every response against