From cd6bcc62a8a912e558386c821df0de381a6fc625 Mon Sep 17 00:00:00 2001
From: Mario Ignatov <mignatov90@gmail.com>
Date: Wed, 19 Aug 2026 14:58:39 +0300
Subject: [PATCH 1/4] feat: sender-affinity load balancing with live-successor
 failover

- add the SenderAffinityLoadBalancing hardfork, Never on real networks and Block(0) on local, layered above TransactionLoadBalancing
- key the committee-slot digest on the recovered sender under the fork, so one validator owns a sender's whole nonce chain instead of consecutive ranges scattering across pools and parking nonce-gapped
- add CommitteeSlots with a ring-walk covers() predicate so a down owner's senders fail over to the next live slot, agreed across nodes by the sorted committee order
- the gossip handler builds per-slot liveness from connected peers (own slot always live) and dispatches batches through covers()

- the fork gate reads the local canonical tip, so validators can briefly disagree across the fork block; worst case is a duplicate include that fails nonce-too-low at execution, never a consensus fork
- on local the fork's version byte (0x0f) outranks HybridRewards from block 0; real networks are unaffected since the fork stays Never
---
 .../src/batch-validator/src/validator.rs      | 168 ++++++++++++++----
 .../consensus/worker/src/network/handler.rs   |  44 ++---
 crates/execution/evm/src/chainspec.rs         |  56 ++++--
 crates/execution/evm/src/evm/hardforks/mod.rs |   3 +-
 .../types/src/worker/sealed_batch.rs          | 160 +++++++++++++++--
 5 files changed, 339 insertions(+), 92 deletions(-)

diff --git a/crates/consensus/worker/src/batch-validator/src/validator.rs b/crates/consensus/worker/src/batch-validator/src/validator.rs
index 93105768..025173f4 100644
--- a/crates/consensus/worker/src/batch-validator/src/validator.rs
+++ b/crates/consensus/worker/src/batch-validator/src/validator.rs
@@ -1,13 +1,14 @@
-//! Block validator
+//! Validation of peer batches and admission of gossiped and forwarded transactions.
 
 use rayls_execution_evm::{
-    bytes_to_txn, chainspec::RaylsHardforks, recover_signed_transaction, reth_env::RethEnv,
-    EthPooledTransaction, FixedBytes, PoolErrorKind, WorkerTxPool,
+    bytes_to_txn, chainspec::RaylsHardforks, recover_pooled_transaction,
+    recover_signed_transaction, reth_env::RethEnv, EthPooledTransaction, FixedBytes,
+    PoolErrorKind, PoolTransaction as _, WorkerTxPool,
 };
 use rayls_infrastructure_types::{
     gas_accumulator::BaseFeeContainer, max_batch_size, BatchValidation, BatchValidationError,
-    BlockHash, Epoch, SealedBatch, SubmitBatchError, TransactionSigned, TransactionTrait as _,
-    WorkerId,
+    BlockHash, CommitteeSlots, Epoch, SealedBatch, SubmitBatchError, TransactionSigned,
+    TransactionTrait as _, WorkerId,
 };
 use rayon::iter::{IntoParallelRefIterator as _, ParallelIterator as _};
 
@@ -95,21 +96,15 @@ impl BatchValidation for BatchValidator {
         // obtain info for validation
         let transactions = batch.transactions();
 
-        // validate batch size (bytes)
-        // Use the parent timestamp for consistency with the batch builder.
         self.validate_batch_size_bytes(transactions, batch.epoch)?;
 
-        // validate txs decode
         let decoded_txs = self.decode_transactions(transactions, digest)?;
 
-        // validate no txs are eip4844
         self.validate_no_blob_txs(&decoded_txs)?;
 
-        // validate gas limit
-        // Use the parent timestamp for consistency with the batch builder.
         self.validate_batch_gas(&decoded_txs)?;
 
-        // validate base fee- all batches for a worker and epoch have the same base fee.
+        // all batches for a worker and epoch share one base fee
         self.validate_basefee(batch.base_fee_per_gas)?;
 
         self.validated_batches.retain(|_, v| *v > rayls_infrastructure_types::now() - 60_000); // keep last minute
@@ -123,8 +118,7 @@ impl BatchValidation for BatchValidator {
     fn submit_batch_if_mine(
         &self,
         txs_bytes: &[Vec<u8>],
-        committee_size: u64,
-        committee_slot: u64,
+        slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError> {
         if let Some(tx_pool) = &self.tx_pool {
             // loop to check if the batch is for this validator because some txns may be errors
@@ -132,11 +126,18 @@ impl BatchValidation for BatchValidator {
                 if tx.len() < 8 {
                     return Err(SubmitBatchError::InvalidTransactionBytes);
                 }
-                let digest = self.slot_digest(tx);
-                if (digest % committee_size) != committee_slot {
+                let owner = self.slot_digest(tx) % slots.size();
+                // Under sender-affinity a down owner's senders fail over to the next live slot on
+                // the ring; pre-fork ownership is exact, with no failover.
+                let mine = if self.sender_affinity_active() {
+                    slots.covers(owner)
+                } else {
+                    owner == slots.own_slot
+                };
+                if !mine {
                     return Ok(());
                 }
-                trace!(target: "worker::validator", ?digest, "tx accepted as committee owner");
+                trace!(target: "worker::validator", ?owner, "tx accepted as committee owner");
             }
 
             let parsed_txns = if txs_bytes.len() < 100 {
@@ -176,7 +177,7 @@ impl BatchValidation for BatchValidator {
 }
 
 impl BatchValidator {
-    /// Create a new instance of [Self]
+    /// Create a validator for one worker and epoch.
     pub fn new(
         reth_env: RethEnv,
         tx_pool: Option<WorkerTxPool>,
@@ -258,25 +259,41 @@ impl BatchValidator {
         Ok(())
     }
 
-    /// Compute the committee-slot dispatch digest for a single transaction.
-    /// Branches on the `TransactionLoadBalancing` hardfork at the next block: pre-fork
-    /// uses [`legacy_slot_digest`], post-fork uses [`fxhash_slot_digest`]. Caller must
-    /// ensure `tx.len() >= 8`.
+    /// Compute the committee-slot dispatch digest for one transaction, which must be at least 8
+    /// bytes.
     ///
-    /// Gate reads the local canonical tip, so validators can briefly disagree on the
-    /// algorithm across the fork block. Worst case is more than one validator includes
-    /// the tx in a batch; the duplicate executions then fail with `nonce too low`. No
+    /// The algorithm follows the hardforks active at the next block: [`legacy_slot_digest`] of
+    /// the bytes, then [`fxhash_slot_digest`] of the bytes under `TransactionLoadBalancing`, then
+    /// [`fxhash_slot_digest`] of the recovered sender under `SenderAffinityLoadBalancing`. The
+    /// gate reads the local canonical tip, so validators can briefly disagree across a fork block;
+    /// the worst case is a duplicate inclusion that fails nonce-too-low at execution, never a
     /// consensus fork.
     fn slot_digest(&self, tx: &[u8]) -> u64 {
         let chain_spec = self.reth_env.rayls_chain_spec();
         let next_block = self.reth_env.canonical_tip().number + 1;
-        if chain_spec.is_transaction_load_balancing_active_at_block(next_block) {
+        if chain_spec.is_sender_affinity_load_balancing_active_at_block(next_block) {
+            // Key the slot on the sender so one validator owns a sender's whole nonce chain instead
+            // of consecutive ranges scattering across pools and parking nonce-gapped.
+            if let Ok(pooled) = recover_pooled_transaction(tx) {
+                return fxhash_slot_digest(pooled.sender().as_slice());
+            }
+            fxhash_slot_digest(tx)
+        } else if chain_spec.is_transaction_load_balancing_active_at_block(next_block) {
             fxhash_slot_digest(tx)
         } else {
             legacy_slot_digest(tx)
         }
     }
 
+    /// Whether sender-affinity dispatch (and its live-successor failover) is active for the next
+    /// block. Reads the local canonical tip, so validators can briefly disagree across the fork.
+    fn sender_affinity_active(&self) -> bool {
+        let next_block = self.reth_env.canonical_tip().number + 1;
+        self.reth_env
+            .rayls_chain_spec()
+            .is_sender_affinity_load_balancing_active_at_block(next_block)
+    }
+
     /// Validate the block's basefee.
     ///
     /// After the EIP-1559 per-block fork, the payload builder computes the correct base fee
@@ -287,7 +304,7 @@ impl BatchValidator {
         let tip = self.reth_env.canonical_tip();
         let next_block = tip.number + 1;
         if chain_spec.is_eip1559_active_at_block(next_block) {
-            // per-block EIP-1559 active - skip exact match
+            // the payload builder derives the base fee from the parent header; see above
             return Ok(());
         }
         let expected_base_fee = self.base_fee.base_fee();
@@ -298,7 +315,7 @@ impl BatchValidator {
         }
     }
 
-    /// Validate the block's basefee
+    /// Reject a batch carrying an EIP-4844 transaction: the blob sidecar does not travel with it.
     fn validate_no_blob_txs(
         &self,
         transactions: &[TransactionSigned],
@@ -309,7 +326,7 @@ impl BatchValidator {
         Ok(())
     }
 
-    /// Helper function for decoding and recovering transactions.
+    /// Decode and recover one transaction, attributing a failure to the batch digest.
     fn recover_and_validate(
         tx: &[u8],
         digest: BlockHash,
@@ -319,7 +336,7 @@ impl BatchValidator {
     }
 }
 
-/// Noop validation struct that validates any block.
+/// Validator that accepts every batch and admits nothing.
 #[cfg(any(test, feature = "test-utils"))]
 #[derive(Default, Clone, Debug)]
 pub struct NoopBatchValidator;
@@ -334,8 +351,7 @@ impl BatchValidation for NoopBatchValidator {
     fn submit_batch_if_mine(
         &self,
         _tx_bytes: &[Vec<u8>],
-        _committee_size: u64,
-        _committee_slot: u64,
+        _slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError> {
         Ok(())
     }
@@ -463,7 +479,7 @@ mod tests {
 
         let txs = vec![vec![0u8; 4]];
         assert_matches!(
-            validator.submit_batch_if_mine(&txs, 4, 0),
+            validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
             Err(SubmitBatchError::InvalidTransactionBytes)
         );
     }
@@ -481,7 +497,10 @@ mod tests {
         let mismatching_slot = (matching_slot + 1) % committee_size;
         let txs = vec![FIXED_TX_BYTES.to_vec()];
         assert_matches!(
-            validator.submit_batch_if_mine(&txs, committee_size, mismatching_slot),
+            validator.submit_batch_if_mine(
+                &txs,
+                &CommitteeSlots::all_live(committee_size as usize, mismatching_slot)
+            ),
             Ok(())
         );
     }
@@ -497,7 +516,10 @@ mod tests {
         let matching_slot = legacy_slot_digest(&FIXED_TX_BYTES) % committee_size;
         let txs = vec![FIXED_TX_BYTES.to_vec()];
         assert_matches!(
-            validator.submit_batch_if_mine(&txs, committee_size, matching_slot),
+            validator.submit_batch_if_mine(
+                &txs,
+                &CommitteeSlots::all_live(committee_size as usize, matching_slot)
+            ),
             Ok(())
         );
     }
@@ -510,7 +532,10 @@ mod tests {
         let TestTools { validator, .. } = test_tools(tmp_dir.path(), &task_manager).await;
 
         let txs: Vec<Vec<u8>> = Vec::new();
-        assert_matches!(validator.submit_batch_if_mine(&txs, 4, 0), Ok(()));
+        assert_matches!(
+            validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
+            Ok(())
+        );
     }
 
     #[serial]
@@ -533,7 +558,10 @@ mod tests {
         // tx_pool=None short-circuits before any slot computation, so even a too-short tx
         // is silently ignored.
         let txs = vec![vec![0u8; 4]];
-        assert_matches!(validator.submit_batch_if_mine(&txs, 4, 0), Ok(()));
+        assert_matches!(
+            validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
+            Ok(())
+        );
     }
 
     /// Return the next valid sealed batch
@@ -965,4 +993,70 @@ mod tests {
             Err(BatchValidationError::InvalidTx4844(_))
         );
     }
+
+    #[serial]
+    #[tokio::test]
+    async fn sender_affinity_routes_a_sender_nonce_chain_to_one_slot() {
+        use rayls_execution_evm::RaylsChainSpec;
+        use rayls_infrastructure_types::RaylsNetwork;
+
+        let tmp_dir = TempDir::new().unwrap();
+        let task_manager = TaskManager::default();
+        let chain: Arc<RethChainSpec> = Arc::new(test_genesis().into());
+        // Local activates SenderAffinityLoadBalancing at block 0.
+        let rayls_chain_spec = Arc::new(
+            RaylsChainSpec::builder(chain.clone()).rayls_hardforks(RaylsNetwork::Local).build(),
+        );
+        let reth_env = RethEnv::new_for_temp_chain_with_rayls_spec(
+            chain.clone(),
+            rayls_chain_spec,
+            tmp_dir.path(),
+            &task_manager,
+            None,
+        )
+        .await
+        .unwrap();
+        let gas_price = reth_env.get_gas_price().unwrap();
+        let validator = BatchValidator::new(
+            reth_env,
+            None,
+            0,
+            BaseFeeContainer::default(),
+            0,
+            ETHEREUM_BLOCK_GAS_LIMIT_56BITS,
+        );
+
+        // two transactions from one sender at consecutive nonces
+        let mut factory = TransactionFactory::new();
+        let value = U256::from(1_000_000_000u64);
+        let tx0 = factory
+            .create_eip1559(
+                chain.clone(),
+                None,
+                gas_price,
+                Some(Address::ZERO),
+                value,
+                Bytes::new(),
+            )
+            .encoded_2718();
+        let tx1 = factory
+            .create_eip1559(
+                chain.clone(),
+                None,
+                gas_price,
+                Some(Address::ZERO),
+                value,
+                Bytes::new(),
+            )
+            .encoded_2718();
+
+        // the two encodings genuinely differ (different nonces)...
+        assert_ne!(fxhash_slot_digest(&tx0), fxhash_slot_digest(&tx1));
+        // ...yet sender-affinity keys the slot on the shared sender, so both route to one owner
+        assert_eq!(validator.slot_digest(&tx0), validator.slot_digest(&tx1));
+
+        // and that slot is exactly the fxhash of the sender address, not of the tx bytes
+        let sender = recover_pooled_transaction(&tx0).unwrap().sender();
+        assert_eq!(validator.slot_digest(&tx0), fxhash_slot_digest(sender.as_slice()));
+    }
 }
diff --git a/crates/consensus/worker/src/network/handler.rs b/crates/consensus/worker/src/network/handler.rs
index 9bfc0ae5..c72defda 100644
--- a/crates/consensus/worker/src/network/handler.rs
+++ b/crates/consensus/worker/src/network/handler.rs
@@ -1,3 +1,5 @@
+//! Handling of peer requests and gossip received by the worker network.
+
 use super::{
     error::{WorkerNetworkError, WorkerNetworkResult},
     message::WorkerGossip,
@@ -9,8 +11,8 @@ use rayls_infrastructure_config::{ConsensusConfig, LibP2pConfig};
 use rayls_infrastructure_network_types::{WorkerOthersBatchMessage, WorkerToPrimaryClient};
 use rayls_infrastructure_storage::tables::Batches;
 use rayls_infrastructure_types::{
-    encode, ensure, now, try_decode, Batch, BatchValidation, BlockHash, BlsPublicKey, Database,
-    DbTx, SealedBatch, WorkerId,
+    encode, ensure, now, try_decode, Batch, BatchValidation, BlockHash, BlsPublicKey,
+    CommitteeSlots, Database, DbTx, SealedBatch, WorkerId,
 };
 use std::sync::{Arc, LazyLock};
 use tracing::{debug, error};
@@ -40,7 +42,7 @@ pub struct RequestHandler<DB> {
     validator: Arc<dyn BatchValidation>,
     /// Consensus config with access to database.
     consensus_config: ConsensusConfig<DB>,
-    /// Network handle- so we can respond to gossip.
+    /// Network handle for fetching batches whose digests arrive by gossip.
     network_handle: WorkerNetworkHandle,
 }
 
@@ -63,7 +65,6 @@ where
     /// Workers gossip the Batch Digests once accepted so that non-committee peers can request the
     /// Batch.
     pub(super) async fn process_gossip(&self, msg: &GossipMessage) -> WorkerNetworkResult<()> {
-        // deconstruct message
         let GossipMessage { data, source: _, sequence_number: _, topic } = msg;
 
         // gossip is uncompressed
@@ -75,12 +76,10 @@ where
                     topic.to_string().eq(&LibP2pConfig::worker_batch_topic()),
                     WorkerNetworkError::InvalidTopic
                 );
-                // Retrieve the block...
                 let store = self.consensus_config.node_storage();
                 if !matches!(store.get::<Batches>(&batch_hash), Ok(Some(_))) {
-                    // If we don't have this batch already then try to get it.
-                    // If we are a CVV then we should already have it.
-                    // This allows non-CVVs to pre fetch batches they will soon need.
+                    // A committee member already holds the batch; a non-committee node prefetches
+                    // what it will soon need.
                     match self.network_handle.request_batches(vec![batch_hash]).await {
                         Ok(batches) => {
                             if let Some(batch) = batches.first() {
@@ -105,17 +104,23 @@ where
                 if let Some(authority) = self.consensus_config.authority() {
                     let committee = self.consensus_config.committee();
                     let authorities = committee.authorities();
-                    let size = authorities.len();
-                    for (slot, auth) in authorities.into_iter().enumerate() {
-                        if &auth == authority {
-                            if let Err(e) = self.validator.submit_batch_if_mine(
-                                &tx_bytes,
-                                size as u64,
-                                slot as u64,
-                            ) {
-                                error!(target: "worker:network", "failed to submit batch: {e}");
-                            }
-                            break;
+                    // Slot liveness is the committee members this node is connected to (own slot
+                    // always live), so a down owner's senders redirect to a live validator instead
+                    // of stranding. Views are per-node and can briefly disagree; the worst case is
+                    // duplicate inclusion, resolved by the nonce check at execution.
+                    let connected = self
+                        .network_handle
+                        .inner_handle()
+                        .connected_peers()
+                        .await
+                        .unwrap_or_default();
+                    if let Some(own_slot) = authorities.iter().position(|auth| auth == authority) {
+                        let keys: Vec<BlsPublicKey> =
+                            authorities.iter().map(|auth| *auth.protocol_key()).collect();
+                        let slots =
+                            CommitteeSlots::from_connectivity(own_slot as u64, &keys, &connected);
+                        if let Err(e) = self.validator.submit_batch_if_mine(&tx_bytes, &slots) {
+                            error!(target: "worker:network", "failed to submit batch: {e}");
                         }
                     }
                 }
@@ -138,7 +143,6 @@ where
 
         let client = self.consensus_config.local_network().clone();
         let store = self.consensus_config.node_storage().clone();
-        // validate batch - log error if invalid
         self.validator.validate_batch(sealed_batch.clone()).await?;
 
         let (mut batch, digest) = sealed_batch.split();
diff --git a/crates/execution/evm/src/chainspec.rs b/crates/execution/evm/src/chainspec.rs
index a057ac70..f9d8c472 100644
--- a/crates/execution/evm/src/chainspec.rs
+++ b/crates/execution/evm/src/chainspec.rs
@@ -60,6 +60,10 @@ hardfork!(
         /// boundary are discarded whole instead of force executed, and an overflow-forced jump
         /// prunes the parked entries it abandons.
         OutputSeqNormalization,
+        /// Key committee-slot dispatch on the first transaction's sender, so one validator owns a
+        /// sender's whole nonce chain instead of consecutive ranges scattering across pools and
+        /// parking nonce-gapped. Also enables live-successor failover for a down slot owner.
+        SenderAffinityLoadBalancing,
     }
 );
 
@@ -146,6 +150,10 @@ pub const MAINNET_LOAD_BALANCING_BLOCK: u64 = 893_558;
 /// Load Balancing activation block on local network.
 pub const LOCAL_LOAD_BALANCING_BLOCK: u64 = 0;
 
+/// Sender-affinity load balancing activation block on local network. Real networks stay `Never`
+/// until an activation block is chosen operationally.
+pub const LOCAL_SENDER_AFFINITY_LOAD_BALANCING_BLOCK: u64 = 0;
+
 // NOTE: UsdrSupplyCorrection is active on local and mainnet; testnet/devnet
 // stay `Never` until an activation block is chosen operationally. Flip the
 // relevant network entry in the schedule below from `ForkCondition::Never` to
@@ -215,11 +223,12 @@ impl RaylsHardFork {
             Self::DynamicCommitteeSizing => 0x0c,
             Self::HybridRewards => 0x0d,
             Self::OutputSeqNormalization => 0x0e,
+            Self::SenderAffinityLoadBalancing => 0x0f,
         }
     }
 
     /// Devnet hardfork schedule.
-    pub const fn devnet() -> [(Self, ForkCondition); 14] {
+    pub const fn devnet() -> [(Self, ForkCondition); 15] {
         [
             (Self::Eip1559, ForkCondition::Block(DEVNET_EIP1559_BLOCK)),
             (Self::BatchDigestV2, ForkCondition::Block(DEVNET_BATCH_DIGEST_V2_BLOCK)),
@@ -242,11 +251,13 @@ impl RaylsHardFork {
             // Never until SRE schedules a concrete devnet activation block.
             (Self::HybridRewards, ForkCondition::Never),
             (Self::OutputSeqNormalization, ForkCondition::Never),
+            // Never until an operational activation block is chosen; the mechanism ships dormant.
+            (Self::SenderAffinityLoadBalancing, ForkCondition::Never),
         ]
     }
 
     /// Testnet hardfork schedule.
-    pub const fn testnet() -> [(Self, ForkCondition); 14] {
+    pub const fn testnet() -> [(Self, ForkCondition); 15] {
         [
             (Self::Eip1559, ForkCondition::Block(TESTNET_EIP1559_BLOCK)),
             (Self::BatchDigestV2, ForkCondition::Block(TESTNET_BATCH_DIGEST_V2_BLOCK)),
@@ -267,11 +278,13 @@ impl RaylsHardFork {
             // Never until SRE schedules a concrete testnet activation block.
             (Self::HybridRewards, ForkCondition::Never),
             (Self::OutputSeqNormalization, ForkCondition::Never),
+            // Never until an operational activation block is chosen; the mechanism ships dormant.
+            (Self::SenderAffinityLoadBalancing, ForkCondition::Never),
         ]
     }
 
     /// Mainnet hardfork schedule.
-    pub const fn mainnet() -> [(Self, ForkCondition); 14] {
+    pub const fn mainnet() -> [(Self, ForkCondition); 15] {
         [
             (Self::Eip1559, ForkCondition::Block(MAINNET_EIP1559_BLOCK)),
             (Self::BatchDigestV2, ForkCondition::Block(MAINNET_BATCH_DIGEST_V2_BLOCK)),
@@ -299,11 +312,13 @@ impl RaylsHardFork {
             (Self::HybridRewards, ForkCondition::Never),
             // Never until SRE schedules a concrete mainnet activation block.
             (Self::OutputSeqNormalization, ForkCondition::Never),
+            // Never until an operational activation block is chosen; the mechanism ships dormant.
+            (Self::SenderAffinityLoadBalancing, ForkCondition::Never),
         ]
     }
 
     /// Local network hardfork schedule (first four hardforks active at genesis).
-    pub const fn local() -> [(Self, ForkCondition); 14] {
+    pub const fn local() -> [(Self, ForkCondition); 15] {
         [
             (Self::Eip1559, ForkCondition::Block(LOCAL_EIP1559_BLOCK)),
             (Self::BatchDigestV2, ForkCondition::Block(LOCAL_BATCH_DIGEST_V2_BLOCK)),
@@ -328,11 +343,15 @@ impl RaylsHardFork {
                 Self::OutputSeqNormalization,
                 ForkCondition::Block(LOCAL_OUTPUT_SEQ_NORMALIZATION_BLOCK),
             ),
+            (
+                Self::SenderAffinityLoadBalancing,
+                ForkCondition::Block(LOCAL_SENDER_AFFINITY_LOAD_BALANCING_BLOCK),
+            ),
         ]
     }
 
     /// Return the hardfork schedule for the given network.
-    pub const fn for_network(network: RaylsNetwork) -> [(Self, ForkCondition); 14] {
+    pub const fn for_network(network: RaylsNetwork) -> [(Self, ForkCondition); 15] {
         match network {
             RaylsNetwork::Devnet => Self::devnet(),
             RaylsNetwork::Testnet => Self::testnet(),
@@ -452,6 +471,11 @@ pub trait RaylsHardforks {
         self.is_rayls_fork_active_at_block(RaylsHardFork::TransactionLoadBalancing, block)
     }
 
+    /// Return true if the SenderAffinityLoadBalancing fork is active at `block`.
+    fn is_sender_affinity_load_balancing_active_at_block(&self, block: u64) -> bool {
+        self.is_rayls_fork_active_at_block(RaylsHardFork::SenderAffinityLoadBalancing, block)
+    }
+
     /// Return true if the EmptyOutputBlock fork is active at `block`.
     fn is_empty_output_block_active_at_block(&self, block: u64) -> bool {
         self.is_rayls_fork_active_at_block(RaylsHardFork::EmptyOutputBlock, block)
@@ -901,20 +925,21 @@ mod tests {
     fn local_network_version_byte_at_block_0() {
         let hardforks = RaylsChainHardforks::local();
         let version = hardforks.version_byte_at_block(0);
-        // OutputSeqNormalization (0x0e) activates at block 0 on local and is the highest such
-        // fork.
-        assert_eq!(version, Some(0x0e));
+        // SenderAffinityLoadBalancing (0x0f) activates at block 0 on local and is the highest
+        // such fork, so it owns the version byte from block 0.
+        assert_eq!(version, Some(0x0f));
     }
 
     #[test]
     fn local_network_version_byte_is_the_max_active_code() {
-        // OutputSeqNormalization (0x0e) is genesis-active on local, so it owns the version byte
-        // across every later activation (HybridRewards at 0x0d included): the byte reports the
-        // max active code, not the most recently crossed block.
+        // SenderAffinityLoadBalancing (0x0f) is genesis-active on local, so it owns the version
+        // byte across every later activation (HybridRewards at 0x0d included): the byte reports
+        // the max active code, not the most recently crossed block. On real networks it is Never,
+        // so there the highest active fork still advances the byte normally.
         let hardforks = RaylsChainHardforks::local();
-        assert_eq!(hardforks.version_byte_at_block(LOCAL_HYBRID_REWARDS_BLOCK - 1), Some(0x0e));
-        assert_eq!(hardforks.version_byte_at_block(LOCAL_HYBRID_REWARDS_BLOCK), Some(0x0e));
-        assert_eq!(hardforks.version_byte_at_block(1_000_000), Some(0x0e));
+        assert_eq!(hardforks.version_byte_at_block(LOCAL_HYBRID_REWARDS_BLOCK - 1), Some(0x0f));
+        assert_eq!(hardforks.version_byte_at_block(LOCAL_HYBRID_REWARDS_BLOCK), Some(0x0f));
+        assert_eq!(hardforks.version_byte_at_block(1_000_000), Some(0x0f));
     }
 
     #[test]
@@ -926,7 +951,7 @@ mod tests {
             RaylsNetwork::Local,
         ] {
             let schedule = RaylsHardFork::for_network(network);
-            assert_eq!(schedule.len(), 14, "expected 14 hardforks for {network}");
+            assert_eq!(schedule.len(), 15, "expected 15 hardforks for {network}");
             assert_eq!(schedule[0].0, RaylsHardFork::Eip1559);
             assert_eq!(schedule[1].0, RaylsHardFork::BatchDigestV2);
             assert_eq!(schedule[2].0, RaylsHardFork::AdminTransfer);
@@ -941,6 +966,7 @@ mod tests {
             assert_eq!(schedule[11].0, RaylsHardFork::DynamicCommitteeSizing);
             assert_eq!(schedule[12].0, RaylsHardFork::HybridRewards);
             assert_eq!(schedule[13].0, RaylsHardFork::OutputSeqNormalization);
+            assert_eq!(schedule[14].0, RaylsHardFork::SenderAffinityLoadBalancing);
         }
     }
 
diff --git a/crates/execution/evm/src/evm/hardforks/mod.rs b/crates/execution/evm/src/evm/hardforks/mod.rs
index de78e13c..1c588c71 100644
--- a/crates/execution/evm/src/evm/hardforks/mod.rs
+++ b/crates/execution/evm/src/evm/hardforks/mod.rs
@@ -135,7 +135,8 @@ where
             | RaylsHardFork::TransactionLoadBalancing
             | RaylsHardFork::EmptyOutputBlock
             | RaylsHardFork::DynamicCommitteeSizing
-            | RaylsHardFork::OutputSeqNormalization => continue,
+            | RaylsHardFork::OutputSeqNormalization
+            | RaylsHardFork::SenderAffinityLoadBalancing => continue,
         };
 
         // Pre-load accounts into cache and copy their real AccountInfo.
diff --git a/crates/infrastructure/types/src/worker/sealed_batch.rs b/crates/infrastructure/types/src/worker/sealed_batch.rs
index f2744770..eac0be82 100644
--- a/crates/infrastructure/types/src/worker/sealed_batch.rs
+++ b/crates/infrastructure/types/src/worker/sealed_batch.rs
@@ -71,12 +71,12 @@ pub struct Batch {
     /// A scalar representing EIP1559 base fee which can move up or down each batch according
     /// to a formula which is a function of gas used in parent batch and gas target
     /// (batch gas limit divided by elasticity multiplier) of parent batch.
-    /// The algorithm results in the base fee per gas increasing when batchs are
-    /// above the gas target, and decreasing when batchs are below the gas target. The base fee per
-    /// gas is sent to governance address.
+    /// The algorithm results in the base fee per gas increasing when batches are
+    /// above the gas target, and decreasing when batches are below the gas target. The base fee
+    /// per gas is sent to governance address.
     pub base_fee_per_gas: u64,
-    /// The worker id for the worker that orginated this batch.
-    /// Worker ids will be consistent accross validators (i.e. worker 0 talks to other worker 0s,
+    /// The worker id for the worker that originated this batch.
+    /// Worker ids will be consistent across validators (i.e. worker 0 talks to other worker 0s,
     /// etc). We can use this for tracking to support base fee calculations.
     /// Note: worker id 0 is the default.
     pub worker_id: WorkerId,
@@ -90,7 +90,7 @@ pub struct Batch {
     pub seq: u64,
     /// Timestamp of when the entity was received by another node. This will help
     /// calculate latencies that are not affected by clock drift or network
-    /// delays. This field is not set for own batchs.
+    /// delays. This field is not set for own batches.
     #[serde(skip)]
     // This field changes often so don't serialize (i.e. don't use it in the digest)
     pub received_at: Option<TimestampSec>,
@@ -133,7 +133,7 @@ impl Batch {
         BlockHash::from_slice(hasher.finalize().as_bytes())
     }
 
-    /// Pass a reference to a collection of transaction bytes;
+    /// Returns a reference to the collection of transaction bytes.
     pub fn transactions(&self) -> &Vec<Vec<u8>> {
         &self.transactions
     }
@@ -148,7 +148,7 @@ impl Batch {
         self.received_at
     }
 
-    /// Sets the recieved at field.
+    /// Sets the received-at time.
     pub fn set_received_at(&mut self, time: TimestampSec) {
         self.received_at = Some(time)
     }
@@ -198,19 +198,81 @@ impl From<&[u8]> for SealedBatch {
     }
 }
 
-/// Return the max gas per batch in effect at timestamp.
-/// Currently allways 30,000,000 but can change in the future at a fork.
+/// Return the max gas per batch in effect for the epoch.
+///
+/// Currently always 30,000,000; the epoch parameter lets a fork change it.
 pub fn max_batch_gas(_epoch: Epoch) -> u64 {
     ETHEREUM_BLOCK_GAS_LIMIT_56BITS
 }
 
-/// Max batch size in effect at a timestamp.  Measured in bytes.
-/// Currently allways 2,000,000 but can change in the future at a fork.
-/// More than this throws msg size error upon decoding
+/// Return the max batch size in bytes in effect for the epoch.
+///
+/// Currently always 2,000,000; the epoch parameter lets a fork change it. A larger batch fails
+/// the message size check on decode.
 pub fn max_batch_size(_epoch: Epoch) -> usize {
     2_000_000
 }
 
+/// Visit every committee slot once, in ring order, starting from `owner`.
+///
+/// The committee is numbered deterministically (sorted authority order), so every validator walks
+/// the same ring, which is what makes live-successor failover agree across nodes.
+pub fn ring_walk(owner: u64, size: u64) -> impl Iterator<Item = u64> {
+    (0..size).map(move |step| (owner + step) % size)
+}
+
+/// This validator's view of committee slot ownership and liveness for sender-affinity dispatch.
+///
+/// Slots are numbered by committee order (sorted by authority id), identical on every validator, so
+/// only `live` is a per-node view. A sender's natural owner is `slot_digest(sender) % size`; if
+/// that owner is down, its senders fail over to the next live slot on the ring via
+/// [`Self::covers`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CommitteeSlots {
+    /// This validator's slot in committee order.
+    pub own_slot: u64,
+    /// Per-slot liveness in committee order; the length is the committee size.
+    pub live: Vec<bool>,
+}
+
+impl CommitteeSlots {
+    /// Build a view in which every slot is live, for the degenerate healthy case and tests.
+    pub fn all_live(size: usize, own_slot: u64) -> Self {
+        Self { own_slot, live: vec![true; size] }
+    }
+
+    /// Build a view marking slot `i` live when it is our own slot or `keys[i]` is in `connected`.
+    ///
+    /// `keys` must be in committee order; a down owner (its key absent from `connected`) then fails
+    /// over to the next live slot via [`Self::covers`]. Own slot is always live.
+    pub fn from_connectivity<K: PartialEq>(own_slot: u64, keys: &[K], connected: &[K]) -> Self {
+        let live = keys
+            .iter()
+            .enumerate()
+            .map(|(slot, key)| slot as u64 == own_slot || connected.contains(key))
+            .collect();
+        Self { own_slot, live }
+    }
+
+    /// The committee size (number of slots).
+    pub fn size(&self) -> u64 {
+        self.live.len() as u64
+    }
+
+    /// Whether this validator covers `owner`'s senders: the first live slot on the ring from
+    /// `owner` is our own. A pure predicate of the view, so validators holding the same view
+    /// agree on the single covering slot.
+    pub fn covers(&self, owner: u64) -> bool {
+        let size = self.size();
+        if size == 0 {
+            return false;
+        }
+        ring_walk(owner % size, size)
+            .find(|slot| self.live[*slot as usize])
+            .is_some_and(|covering| covering == self.own_slot)
+    }
+}
+
 /// Defines the validation procedure for receiving either a new single transaction (from a client)
 /// of a batch of transactions (from another validator).
 ///
@@ -221,24 +283,23 @@ pub trait BatchValidation: Send + Sync + Debug {
     async fn validate_batch(&self, b: SealedBatch) -> Result<(), BatchValidationError>;
 
     /// Submit a batch (as bytes) for inclusion in a batch.
-    /// Will only submit if the txn hash fits the provided committee slot.
+    /// Will only submit if the batch's first transaction maps to a slot this validator covers.
     fn submit_batch_if_mine(
         &self,
         tx_bytes: &[Vec<u8>],
-        committee_size: u64,
-        committee_slot: u64,
+        slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError>;
 }
 
 /// Errors that can occur during batch submission.
 #[derive(Error, Debug)]
 pub enum SubmitBatchError {
-    /// The tx not correctly encoded
+    /// The transaction is not correctly encoded.
     #[error("Invalid transaction bytes")]
     InvalidTransactionBytes,
 }
 
-/// Block validation error types
+/// Errors from validating a peer's batch.
 #[derive(Error, Debug)]
 pub enum BatchValidationError {
     /// The sealed batch hash does not match this worker's calculated digest.
@@ -262,7 +323,7 @@ pub enum BatchValidationError {
         /// The gas limit in the batch header.
         gas_limit: u64,
     },
-    /// Error while calculating max possible gas from icluded transactions.
+    /// Error while calculating max possible gas from included transactions.
     #[error("Unable to reduce max possible gas limit for peer's batch")]
     CalculateMaxPossibleGas,
     /// Error when peer's transaction list exceeds the maximum bytes allowed.
@@ -288,3 +349,64 @@ pub enum BatchValidationError {
     #[error("Invalid epoch, expected epoch {expected} got epoch {found}")]
     InvalidEpoch { expected: Epoch, found: Epoch },
 }
+
+#[cfg(test)]
+mod committee_slots_tests {
+    use super::*;
+
+    #[test]
+    fn ring_walk_visits_every_slot_once_from_the_owner() {
+        assert_eq!(ring_walk(1, 4).collect::<Vec<_>>(), vec![1, 2, 3, 0]);
+        assert_eq!(ring_walk(0, 3).collect::<Vec<_>>(), vec![0, 1, 2]);
+    }
+
+    #[test]
+    fn all_live_owner_covers_only_itself() {
+        let slots = CommitteeSlots::all_live(4, 2);
+        assert!(slots.covers(2), "the natural owner covers its own senders");
+        assert!(!slots.covers(1), "a healthy peer's senders are not ours");
+        assert!(!slots.covers(3), "a healthy peer's senders are not ours");
+    }
+
+    #[test]
+    fn down_owner_fails_over_to_the_next_live_slot() {
+        // committee of 4, we are slot 2, slot 1 is down
+        let slots = CommitteeSlots { own_slot: 2, live: vec![true, false, true, true] };
+        // owner 1 is down: ring walk 1,2,3,0 finds slot 2 (us) first, so we cover it
+        assert!(slots.covers(1), "a down owner's senders fail over to the next live slot");
+        // owner 0 is live and covered by slot 0, not us
+        assert!(!slots.covers(0));
+    }
+
+    #[test]
+    fn failover_wraps_around_the_ring() {
+        // we are slot 0, and every other slot is down
+        let slots = CommitteeSlots { own_slot: 0, live: vec![true, false, false, false] };
+        // owner 3 is down: ring walk 3,0,1,2 wraps to slot 0 (us)
+        assert!(slots.covers(3), "failover wraps past the end of the ring");
+        assert!(slots.covers(1), "owner 1 down: walk 1,2,3,0 lands on slot 0 (us)");
+    }
+
+    #[test]
+    fn no_live_slot_covers_nothing() {
+        let slots = CommitteeSlots { own_slot: 0, live: vec![false, false] };
+        assert!(!slots.covers(0));
+        assert!(!slots.covers(1));
+    }
+
+    #[test]
+    fn from_connectivity_marks_own_slot_and_connected_peers_live() {
+        // committee-ordered peer keys; we are slot 1, and only slot 2's peer is connected
+        let keys = [10u8, 20, 30, 40];
+        let connected = [30u8];
+        let slots = CommitteeSlots::from_connectivity(1, &keys, &connected);
+
+        // own slot is always live; connected peer (slot 2) is live; the rest are down
+        assert_eq!(slots.live, vec![false, true, true, false]);
+        // senders owned by the down slots 0 and 3 fail over to us (slot 1)
+        assert!(slots.covers(0), "down slot 0 fails over to us");
+        assert!(slots.covers(3), "down slot 3 fails over to us");
+        // slot 2 is live, so it keeps its own senders
+        assert!(!slots.covers(2), "a live peer keeps its senders");
+    }
+}

From 667945c5713485054eaab43f129bfcc1c9e4c01c Mon Sep 17 00:00:00 2001
From: Mario Ignatov <mignatov90@gmail.com>
Date: Wed, 19 Aug 2026 16:02:59 +0300
Subject: [PATCH 2/4] feat: forward an observer's transactions directly to the
 committee

- add a per-epoch observer forwarder that submits pending transactions by request-response to the validator owning each sender's slot, falling back to gossip before the fork
- append WorkerRequest::SubmitTxns last and WorkerResponse::SubmitTxns after Error so every existing variant keeps its bcs index across a rolling upgrade; the ack carries the hashes the owner rejected as nonce-too-low
- share one node-scoped in-flight tracker between the pool and its role via init_txn_pool_with_in_flight, bringing the dormant forwarding marks alive
- an observer is no longer batch-producing: it forwards instead of running a batch builder that disburses
- gate re-sends behind the catch-up watermark while first sends of new transactions still flow, and ring-walk to the next live validator on failover, marking accepted hashes and suppressing acked-stale ones
- change WorkerGossip::Txn, publish_txn, and submit_batch_if_mine to Vec<Bytes> (bcs-identical to Vec<Vec<u8>>) so a payload is encoded once for both the direct and gossip paths
- move fxhash_slot_digest and legacy_slot_digest to the types crate so the forwarder and validator compute the identical owner slot

- bound inbound submit fan-out with a semaphore, shedding via the existing Error variant so the sender retries the next validator
---
 Cargo.lock                                    |   2 +-
 crates/consensus/primary/src/consensus_bus.rs |  12 +-
 .../worker/src/batch-validator/Cargo.toml     |   2 +-
 .../src/batch-validator/src/validator.rs      | 156 +++--
 .../consensus/worker/src/network/handler.rs   |   7 +-
 .../consensus/worker/src/network/message.rs   |  74 ++-
 crates/consensus/worker/src/network/mod.rs    | 141 ++++-
 crates/consensus/worker/src/worker.rs         |   7 +-
 .../execution/evm/src/reth_env/accessors.rs   |   9 +-
 crates/execution/evm/src/txn_pool.rs          |  45 +-
 crates/infrastructure/types/Cargo.toml        |   1 +
 .../types/src/worker/sealed_batch.rs          |  44 +-
 .../middleware/orchestrator/src/engine/mod.rs |  15 +-
 .../orchestrator/src/engine/node.rs           |  28 +-
 .../orchestrator/src/engine/node_builder.rs   |   2 +-
 .../orchestrator/src/engine/node_inner.rs     |  72 ++-
 .../orchestrator/src/engine/txn_forwarder.rs  | 576 ++++++++++++++++++
 .../orchestrator/src/epoch_manager/core.rs    |  74 ++-
 .../src/tests/batch_seq_gate_tests.rs         |   6 +-
 19 files changed, 1113 insertions(+), 160 deletions(-)
 create mode 100644 crates/middleware/orchestrator/src/engine/txn_forwarder.rs

diff --git a/Cargo.lock b/Cargo.lock
index 35ba0c68..76b36613 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7752,7 +7752,6 @@ dependencies = [
  "rayls-execution-evm",
  "rayls-infrastructure-types",
  "rayon",
- "rustc-hash",
  "serial_test",
  "tempfile",
  "tokio",
@@ -8094,6 +8093,7 @@ dependencies = [
  "reth-primitives",
  "reth-tasks",
  "roaring 0.10.12",
+ "rustc-hash",
  "secp256k1 0.31.1",
  "serde",
  "serde_repr",
diff --git a/crates/consensus/primary/src/consensus_bus.rs b/crates/consensus/primary/src/consensus_bus.rs
index 8c9c4946..521c5427 100644
--- a/crates/consensus/primary/src/consensus_bus.rs
+++ b/crates/consensus/primary/src/consensus_bus.rs
@@ -168,13 +168,15 @@ impl NodeMode {
         matches!(self, NodeMode::Observer)
     }
 
-    /// True if this node should run a batch builder (active CVV sequences, observer disburses).
+    /// True if this node should run a batch builder (active CVVs sequence into consensus).
     ///
-    /// A catching-up `CvvInactive` node must not: with no proposer draining `our_digests`, a
-    /// sealed batch wedges the worker batch-builder on `report_own_batch`, and that Drainable task
-    /// never observes shutdown, stalling the epoch-transition drain.
+    /// An `Observer` is not batch-producing: it cannot seal, so it forwards its pending
+    /// transactions to the committee instead (see the transaction forwarder). A catching-up
+    /// `CvvInactive` node must not either: with no proposer draining `our_digests`, a sealed batch
+    /// wedges the worker batch-builder on `report_own_batch`, and that Drainable task never
+    /// observes shutdown, stalling the epoch-transition drain.
     pub fn is_batch_producing(&self) -> bool {
-        matches!(self, NodeMode::CvvActive | NodeMode::Observer)
+        matches!(self, NodeMode::CvvActive)
     }
 }
 
diff --git a/crates/consensus/worker/src/batch-validator/Cargo.toml b/crates/consensus/worker/src/batch-validator/Cargo.toml
index 0a494ee6..39140c58 100644
--- a/crates/consensus/worker/src/batch-validator/Cargo.toml
+++ b/crates/consensus/worker/src/batch-validator/Cargo.toml
@@ -14,9 +14,9 @@ async-trait = { workspace = true }
 rayls-infrastructure-types = { workspace = true }
 rayls-execution-evm = { workspace = true }
 rayon = { workspace = true }
+tokio = { workspace = true, features = ["rt"] }
 tracing = { workspace = true }
 dashmap = { workspace = true }
-rustc-hash = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
diff --git a/crates/consensus/worker/src/batch-validator/src/validator.rs b/crates/consensus/worker/src/batch-validator/src/validator.rs
index 025173f4..f89edc3d 100644
--- a/crates/consensus/worker/src/batch-validator/src/validator.rs
+++ b/crates/consensus/worker/src/batch-validator/src/validator.rs
@@ -6,55 +6,51 @@ use rayls_execution_evm::{
     PoolErrorKind, PoolTransaction as _, WorkerTxPool,
 };
 use rayls_infrastructure_types::{
-    gas_accumulator::BaseFeeContainer, max_batch_size, BatchValidation, BatchValidationError,
-    BlockHash, CommitteeSlots, Epoch, SealedBatch, SubmitBatchError, TransactionSigned,
-    TransactionTrait as _, WorkerId,
+    fxhash_slot_digest, gas_accumulator::BaseFeeContainer, legacy_slot_digest, max_batch_size,
+    BatchValidation, BatchValidationError, BlockHash, Bytes, CommitteeSlots, Epoch, SealedBatch,
+    SubmitBatchError, TransactionSigned, TransactionTrait as _, WorkerId,
 };
 use rayon::iter::{IntoParallelRefIterator as _, ParallelIterator as _};
 
 use dashmap::DashMap;
-use rustc_hash::FxHasher;
-use std::hash::Hasher;
 use tracing::{debug, trace, warn};
 
-/// Type convenience for implementing block validation errors.
+/// Result alias for batch validation.
 type BatchValidationResult<T> = Result<T, BatchValidationError>;
 
-/// Pre-`TransactionLoadBalancing` slot digest: read the first 8 bytes as little-endian u64.
-/// Caller must ensure `tx.len() >= 8`.
-fn legacy_slot_digest(tx: &[u8]) -> u64 {
-    let mut bytes = [0u8; 8];
-    bytes.copy_from_slice(&tx[0..8]);
-    u64::from_le_bytes(bytes)
-}
-
-/// Post-`TransactionLoadBalancing` slot digest: `FxHasher` over the full transaction bytes.
-/// Do NOT replace with `FxBuildHasher::hash_one(tx)`: that path writes a slice-length
-/// prefix and changes the digest.
-fn fxhash_slot_digest(tx: &[u8]) -> u64 {
-    let mut hasher = FxHasher::default();
-    hasher.write(tx);
-    hasher.finish()
+/// Recover signers in parallel above this many transactions; below it the rayon fan-out costs more
+/// than it saves.
+const PARALLEL_PARSE_THRESHOLD: usize = 100;
+
+/// Recover signers for forwarded transactions, in parallel once the batch is large enough.
+///
+/// Invalid encodings are dropped; the pool validates the rest and reports any that are stale.
+fn recover_forwarded_txns<T: AsRef<[u8]> + Sync>(txs_bytes: &[T]) -> Vec<EthPooledTransaction> {
+    if txs_bytes.len() < PARALLEL_PARSE_THRESHOLD {
+        txs_bytes.iter().filter_map(|bytes| bytes_to_txn(bytes.as_ref()).ok()).collect()
+    } else {
+        txs_bytes.par_iter().filter_map(|bytes| bytes_to_txn(bytes.as_ref()).ok()).collect()
+    }
 }
 
-/// Batch validator
-/// Important note about batch validation, we rely on libp2p to verify that
-/// batches came from a committee member.  This means we do not generate or
-/// check our own signatures for batches since they all came from current
-/// committee members.
+/// Validator for peer batches and the dispatch gate for inbound transactions.
+///
+/// Batches carry no signature of their own: libp2p authenticates the sending peer as a committee
+/// member, so validation checks only the batch contents.
 #[derive(Clone, Debug)]
 pub struct BatchValidator {
-    /// Database provider to encompass tree and provider factory.
+    /// Execution environment providing the canonical tip and the chain spec.
     reth_env: RethEnv,
-    /// A handle to the transaction pool for submitting gossipped transactions.
+    /// The transaction pool inbound transactions are admitted to; `None` on a node that does not
+    /// pool for the committee.
     tx_pool: Option<WorkerTxPool>,
     /// Worker id for this validator.
     worker_id: WorkerId,
-    /// Current base fee for this validators worker.
+    /// Current base fee for this validator's worker.
     base_fee: BaseFeeContainer,
     /// Epoch we are validating for.
     epoch: Epoch,
-    /// holds recently validated batches to prevent re-validation
+    /// Digests validated within the last minute, so a re-gossiped batch is not re-validated.
     validated_batches: DashMap<FixedBytes<32>, u64>,
     /// Block gas limit.
     gas_limit: u64,
@@ -113,15 +109,16 @@ impl BatchValidation for BatchValidator {
         Ok(())
     }
 
-    /// Submit a transaction received from the gossip pool to the worker's transaction pool.
-    /// This method is only active if the node is part of the committee.
+    /// Admit a gossiped transaction message to the pool when this node owns its committee slot.
+    ///
+    /// The message's first transaction decides the owner slot, so a forwarder must keep one
+    /// sender's run in a single message. A node without a pool ignores every message.
     fn submit_batch_if_mine(
         &self,
-        txs_bytes: &[Vec<u8>],
+        txs_bytes: &[Bytes],
         slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError> {
         if let Some(tx_pool) = &self.tx_pool {
-            // loop to check if the batch is for this validator because some txns may be errors
             if let Some(tx) = txs_bytes.iter().next() {
                 if tx.len() < 8 {
                     return Err(SubmitBatchError::InvalidTransactionBytes);
@@ -174,6 +171,24 @@ impl BatchValidation for BatchValidator {
 
         Ok(())
     }
+
+    async fn submit_forwarded_txns(&self, tx_bytes: Vec<Bytes>) -> Vec<BlockHash> {
+        let Some(tx_pool) = &self.tx_pool else {
+            return Vec::new();
+        };
+        // Recover signers off the runtime; the owned Vec moves straight into the blocking task.
+        let parsed =
+            match tokio::task::spawn_blocking(move || recover_forwarded_txns(&tx_bytes)).await {
+                Ok(parsed) => parsed,
+                // A cancelled blocking task means runtime teardown; ack nothing so the sender keeps
+                // the transactions and retries.
+                Err(e) => {
+                    warn!(target: "worker::validator", ?e, "signer recovery task did not complete");
+                    return Vec::new();
+                }
+            };
+        tx_pool.add_forwarded_txns(parsed).await
+    }
 }
 
 impl BatchValidator {
@@ -350,11 +365,15 @@ impl BatchValidation for NoopBatchValidator {
 
     fn submit_batch_if_mine(
         &self,
-        _tx_bytes: &[Vec<u8>],
+        _tx_bytes: &[Bytes],
         _slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError> {
         Ok(())
     }
+
+    async fn submit_forwarded_txns(&self, _tx_bytes: Vec<Bytes>) -> Vec<BlockHash> {
+        Vec::new()
+    }
 }
 
 #[cfg(test)]
@@ -477,7 +496,7 @@ mod tests {
         let task_manager = TaskManager::default();
         let TestTools { validator, .. } = test_tools(tmp_dir.path(), &task_manager).await;
 
-        let txs = vec![vec![0u8; 4]];
+        let txs = vec![Bytes::from(vec![0u8; 4])];
         assert_matches!(
             validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
             Err(SubmitBatchError::InvalidTransactionBytes)
@@ -495,7 +514,7 @@ mod tests {
         let committee_size = 4_u64;
         let matching_slot = legacy_slot_digest(&FIXED_TX_BYTES) % committee_size;
         let mismatching_slot = (matching_slot + 1) % committee_size;
-        let txs = vec![FIXED_TX_BYTES.to_vec()];
+        let txs = vec![Bytes::from(FIXED_TX_BYTES.to_vec())];
         assert_matches!(
             validator.submit_batch_if_mine(
                 &txs,
@@ -514,7 +533,7 @@ mod tests {
 
         let committee_size = 4_u64;
         let matching_slot = legacy_slot_digest(&FIXED_TX_BYTES) % committee_size;
-        let txs = vec![FIXED_TX_BYTES.to_vec()];
+        let txs = vec![Bytes::from(FIXED_TX_BYTES.to_vec())];
         assert_matches!(
             validator.submit_batch_if_mine(
                 &txs,
@@ -531,7 +550,7 @@ mod tests {
         let task_manager = TaskManager::default();
         let TestTools { validator, .. } = test_tools(tmp_dir.path(), &task_manager).await;
 
-        let txs: Vec<Vec<u8>> = Vec::new();
+        let txs: Vec<Bytes> = Vec::new();
         assert_matches!(
             validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
             Ok(())
@@ -557,7 +576,7 @@ mod tests {
 
         // tx_pool=None short-circuits before any slot computation, so even a too-short tx
         // is silently ignored.
-        let txs = vec![vec![0u8; 4]];
+        let txs = vec![Bytes::from(vec![0u8; 4])];
         assert_matches!(
             validator.submit_batch_if_mine(&txs, &CommitteeSlots::all_live(4, 0)),
             Ok(())
@@ -1059,4 +1078,61 @@ mod tests {
         let sender = recover_pooled_transaction(&tx0).unwrap().sender();
         assert_eq!(validator.slot_digest(&tx0), fxhash_slot_digest(sender.as_slice()));
     }
+
+    #[serial]
+    #[tokio::test]
+    async fn submit_forwarded_txns_reports_only_nonce_too_low_as_stale() {
+        use rayls_infrastructure_types::{GenesisAccount, U256};
+
+        let tmp_dir = TempDir::new().unwrap();
+        let task_manager = TaskManager::default();
+        let mut factory = TransactionFactory::new();
+        // Seed the sender at nonce 1, so its nonce-0 transaction is already executed (stale) while
+        // its nonce-1 transaction is the next valid one.
+        let genesis = test_genesis().extend_accounts([(
+            factory.address(),
+            GenesisAccount::default().with_balance(U256::MAX).with_nonce(Some(1)),
+        )]);
+        let chain: Arc<RethChainSpec> = Arc::new(genesis.into());
+        let reth_env =
+            RethEnv::new_for_temp_chain(chain.clone(), tmp_dir.path(), &task_manager, None)
+                .await
+                .unwrap();
+        let tx_pool = reth_env.init_txn_pool().unwrap();
+        let gas_price = reth_env.get_gas_price().unwrap();
+        let validator = BatchValidator::new(
+            reth_env,
+            Some(tx_pool),
+            0,
+            BaseFeeContainer::default(),
+            0,
+            ETHEREUM_BLOCK_GAS_LIMIT_56BITS,
+        );
+
+        let value = U256::from(1);
+        factory.set_nonce(0);
+        let stale_tx = factory.create_eip1559(
+            chain.clone(),
+            None,
+            gas_price,
+            Some(Address::ZERO),
+            value,
+            Bytes::new(),
+        );
+        factory.set_nonce(1);
+        let ok_tx = factory.create_eip1559(
+            chain.clone(),
+            None,
+            gas_price,
+            Some(Address::ZERO),
+            value,
+            Bytes::new(),
+        );
+
+        let payloads =
+            vec![Bytes::from(stale_tx.encoded_2718()), Bytes::from(ok_tx.encoded_2718())];
+        let stale = validator.submit_forwarded_txns(payloads).await;
+
+        assert_eq!(stale, vec![*stale_tx.hash()], "only the nonce-too-low tx is reported stale");
+    }
 }
diff --git a/crates/consensus/worker/src/network/handler.rs b/crates/consensus/worker/src/network/handler.rs
index c72defda..2c352452 100644
--- a/crates/consensus/worker/src/network/handler.rs
+++ b/crates/consensus/worker/src/network/handler.rs
@@ -11,7 +11,7 @@ use rayls_infrastructure_config::{ConsensusConfig, LibP2pConfig};
 use rayls_infrastructure_network_types::{WorkerOthersBatchMessage, WorkerToPrimaryClient};
 use rayls_infrastructure_storage::tables::Batches;
 use rayls_infrastructure_types::{
-    encode, ensure, now, try_decode, Batch, BatchValidation, BlockHash, BlsPublicKey,
+    encode, ensure, now, try_decode, Batch, BatchValidation, BlockHash, BlsPublicKey, Bytes,
     CommitteeSlots, Database, DbTx, SealedBatch, WorkerId,
 };
 use std::sync::{Arc, LazyLock};
@@ -130,6 +130,11 @@ where
         Ok(())
     }
 
+    /// Admit transactions forwarded directly by an observer, returning the stale-hash ack.
+    pub(super) async fn submit_forwarded_txns(&self, transactions: Vec<Bytes>) -> Vec<BlockHash> {
+        self.validator.submit_forwarded_txns(transactions).await
+    }
+
     /// Process a new reported batch.
     pub(super) async fn process_report_batch(
         &self,
diff --git a/crates/consensus/worker/src/network/message.rs b/crates/consensus/worker/src/network/message.rs
index e80d80fe..314ab825 100644
--- a/crates/consensus/worker/src/network/message.rs
+++ b/crates/consensus/worker/src/network/message.rs
@@ -1,19 +1,21 @@
 //! Messages sent between workers.
 
 use rayls_consensus_network::{PeerExchangeMap, RLMessage};
-use rayls_infrastructure_types::{Batch, BlockHash, SealedBatch};
+use rayls_infrastructure_types::{Batch, BlockHash, Bytes, SealedBatch};
 use serde::{Deserialize, Serialize};
 
 /// Worker messages on the gossip network.
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum WorkerGossip {
-    /// A new is available.
+    /// A new batch is available by digest.
     Batch(BlockHash),
-    /// Transaction- published so a committee member can include in a batch.
-    Txn(Vec<Vec<u8>>),
+    /// Transactions published so a committee member can include them in a batch.
+    ///
+    /// `Vec<Bytes>` is bcs-identical to `Vec<Vec<u8>>` (each element is a length-prefixed byte run
+    /// either way), so this is not a wire-shape change: un-upgraded peers decode it unchanged.
+    Txn(Vec<Bytes>),
 }
 
-// impl RLMessage trait for types
 impl RLMessage for WorkerRequest {
     fn peer_exchange_msg(&self) -> Option<PeerExchangeMap> {
         match self {
@@ -38,7 +40,7 @@ pub enum WorkerRequest {
     },
     /// Request batches by digest from a peer.
     RequestBatches {
-        /// The requests batches by digests.
+        /// The digests of the requested batches.
         batch_digests: Vec<BlockHash>,
         /// Maximum expected response size.
         max_response_size: usize,
@@ -52,6 +54,17 @@ pub enum WorkerRequest {
         /// The peer information being exchanged.
         peers: PeerExchangeMap,
     },
+    /// Forward transactions directly to the committee member that owns their senders' slots.
+    ///
+    /// Appended last: bcs is positional, so this is wire-safe as long as un-upgraded peers never
+    /// receive it. Senders gate it on the `SenderAffinityLoadBalancing` fork and otherwise publish
+    /// on the txn gossip topic. `Bytes` is bcs-identical to `Vec<u8>` (both a length-prefixed byte
+    /// run).
+    SubmitTxns {
+        /// The forwarded transactions as encoded bytes: sender-contiguous, nonce-ascending runs so
+        /// each sender's chain pools in order from one message.
+        transactions: Vec<Bytes>,
+    },
 }
 
 impl From<PeerExchangeMap> for WorkerRequest {
@@ -60,12 +73,6 @@ impl From<PeerExchangeMap> for WorkerRequest {
     }
 }
 
-//
-//
-//=== Response types
-//
-//
-
 /// Response to worker requests.
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum WorkerResponse {
@@ -82,10 +89,20 @@ pub enum WorkerResponse {
     ///
     /// This is an application-layer error response.
     Error(WorkerRPCError),
+    /// The ack for [`WorkerRequest::SubmitTxns`], listing the hashes the owner rejected as stale.
+    ///
+    /// Appended after `Error`: bcs encodes the variant index positionally, so inserting before
+    /// `Error` would shift its index and break error decoding against un-upgraded peers on every
+    /// RPC path. Everything not listed was accepted or already known, so the sender stops
+    /// re-forwarding only the stale hashes.
+    SubmitTxns {
+        /// Hashes the owner rejected as nonce-too-low (already executed).
+        stale: Vec<BlockHash>,
+    },
 }
 
 impl WorkerResponse {
-    /// Helper method if the response is an error.
+    /// Returns `true` if the response is an application-layer error.
     pub fn is_err(&self) -> bool {
         matches!(self, WorkerResponse::Error(_))
     }
@@ -97,7 +114,7 @@ impl From<WorkerRPCError> for WorkerResponse {
     }
 }
 
-/// Application-specific error type while handling Worker request.
+/// Application-layer error returned while handling a worker request.
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
 pub struct WorkerRPCError(pub String);
 
@@ -106,3 +123,32 @@ impl From<PeerExchangeMap> for WorkerResponse {
         Self::PeerExchange { peers: value }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rayls_infrastructure_types::{encode, try_decode};
+
+    #[test]
+    fn worker_response_error_keeps_bcs_index_3() {
+        // Appending SubmitTxns AFTER Error must not shift Error's positional bcs discriminant: an
+        // un-upgraded peer decodes Error (index 3) on every RPC error path, so a shift breaks error
+        // decoding across versions.
+        let err = WorkerResponse::Error(WorkerRPCError("boom".into()));
+        assert_eq!(encode(&err)[0], 3, "Error must stay bcs variant index 3");
+        assert_eq!(
+            encode(&WorkerResponse::SubmitTxns { stale: vec![] })[0],
+            4,
+            "SubmitTxns is appended last, at index 4"
+        );
+
+        // Error round-trips as itself, not misread as SubmitTxns.
+        let decoded: WorkerResponse = try_decode(&encode(&err)).unwrap();
+        assert!(matches!(decoded, WorkerResponse::Error(WorkerRPCError(s)) if s == "boom"));
+    }
+
+    #[test]
+    fn worker_request_submit_txns_is_appended_last() {
+        assert_eq!(encode(&WorkerRequest::SubmitTxns { transactions: vec![] })[0], 3);
+    }
+}
diff --git a/crates/consensus/worker/src/network/mod.rs b/crates/consensus/worker/src/network/mod.rs
index 8e71187e..d99f1f8f 100644
--- a/crates/consensus/worker/src/network/mod.rs
+++ b/crates/consensus/worker/src/network/mod.rs
@@ -17,8 +17,8 @@ use rayls_infrastructure_network_types::{
 };
 use rayls_infrastructure_storage::tables::Batches;
 use rayls_infrastructure_types::{
-    encode, now, Batch, BatchValidation, BlockHash, BlsPublicKey, Database, DbTxMut, RaylsReceiver,
-    SealedBatch, TaskKind, TaskSpawner, WorkerId,
+    encode, now, Batch, BatchValidation, BlockHash, BlsPublicKey, Bytes, Database, DbTxMut,
+    RaylsReceiver, SealedBatch, TaskKind, TaskSpawner, TxHash, WorkerId,
 };
 use std::{collections::HashSet, sync::Arc, time::Duration};
 use tokio::sync::{oneshot, Semaphore};
@@ -28,9 +28,9 @@ pub(crate) mod error;
 pub(crate) mod handler;
 pub(crate) mod message;
 
-/// Convenience type for Primary network.
+/// Request type of the worker network.
 pub(crate) type Req = WorkerRequest;
-/// Convenience type for Primary network.
+/// Response type of the worker network.
 pub(crate) type Res = WorkerResponse;
 
 /// Soft cap on this handle's concurrently in-flight outbound batch requests.
@@ -40,6 +40,23 @@ pub(crate) type Res = WorkerResponse;
 /// connections. A frequency reducer, not a hard bound (see `outbound_failure_penalty`).
 const MAX_CONCURRENT_BATCH_REQUESTS: usize = 32;
 
+/// Cap on concurrently admitted inbound transaction submits.
+///
+/// Each submit recovers signers on the shared rayon pool that batch validation also uses, so an
+/// unbounded burst would starve consensus; excess submits are shed back to the sender, which
+/// retries the next validator.
+const MAX_CONCURRENT_SUBMIT_TXNS: usize = 16;
+
+/// Encoded size of a [`WorkerRequest::SubmitTxns`] carrying nothing: the envelope every direct
+/// submit pays on top of its transaction bytes.
+static TXN_SUBMIT_OVERHEAD: std::sync::LazyLock<usize> =
+    std::sync::LazyLock::new(|| encode(&WorkerRequest::SubmitTxns { transactions: vec![] }).len());
+
+/// Encoded size of a [`WorkerGossip::Txn`] carrying nothing: the frame every published payload
+/// pays on top of its transaction bytes.
+static TXN_GOSSIP_OVERHEAD: std::sync::LazyLock<usize> =
+    std::sync::LazyLock::new(|| encode(&WorkerGossip::Txn(vec![])).len());
+
 /// The wrapper around worker-specific network calls.
 #[derive(Clone, Debug)]
 pub struct WorkerNetworkHandle {
@@ -73,9 +90,7 @@ impl WorkerNetworkHandle {
         &self.task_spawner
     }
 
-    /// Convenience method for creating a new Self for tests- sends events no-where and does
-    /// nothing.
-    /// #[cfg(any(test, feature = "test-utils"))]
+    /// Create a handle for tests whose commands go nowhere.
     pub fn new_for_test(task_spawner: TaskSpawner) -> Self {
         let (tx, _rx) = tokio::sync::mpsc::channel(5);
         Self {
@@ -91,6 +106,42 @@ impl WorkerNetworkHandle {
         &self.handle
     }
 
+    /// Forward transactions directly to `peer_bls`, returning the hashes it rejected as stale.
+    pub async fn submit_txns(
+        &self,
+        peer_bls: BlsPublicKey,
+        transactions: Vec<Bytes>,
+        timeout: Duration,
+    ) -> NetworkResult<Vec<TxHash>> {
+        let request = WorkerRequest::SubmitTxns { transactions };
+        let res = self.handle.send_request(request, peer_bls).await?;
+        let res =
+            tokio::time::timeout(timeout, res).await.map_err(|_| NetworkError::Timeout)???;
+        match res {
+            WorkerResponse::SubmitTxns { stale } => Ok(stale),
+            WorkerResponse::ReportBatch
+            | WorkerResponse::RequestBatches { .. }
+            | WorkerResponse::PeerExchange { .. } => Err(NetworkError::RPCError(
+                "Got wrong response, not a submit txns response!".to_string(),
+            )),
+            WorkerResponse::Error(WorkerRPCError(s)) => Err(NetworkError::RPCError(s)),
+        }
+    }
+
+    /// The most transaction bytes one direct submit may carry, leaving room for the request
+    /// envelope under the RPC message-size cap.
+    pub fn direct_txn_payload_budget(&self) -> usize {
+        self.max_rpc_message_size.saturating_sub(*TXN_SUBMIT_OVERHEAD)
+    }
+
+    /// Transaction bytes that fit in one [`Self::publish_txn`] under `max_gossip_message_size`.
+    ///
+    /// Peers reject an over-budget message outright, so a publisher subtracts the encoding frame
+    /// itself rather than discover the cap on the receiving side.
+    pub fn txn_payload_budget(max_gossip_message_size: usize) -> usize {
+        max_gossip_message_size.saturating_sub(*TXN_GOSSIP_OVERHEAD)
+    }
+
     /// Publish a batch digest to the worker network.
     pub(crate) async fn publish_batch(&self, batch_digest: BlockHash) -> NetworkResult<()> {
         let data = encode(&WorkerGossip::Batch(batch_digest));
@@ -98,9 +149,9 @@ impl WorkerNetworkHandle {
         Ok(())
     }
 
-    /// Publish a transaction (as raw bytes) worker network.
-    /// Do this when not a committee member so a CVV can include the txn.
-    pub(crate) async fn publish_txn(&self, txn: Vec<Vec<u8>>) -> NetworkResult<()> {
+    /// Publish encoded transactions on the worker transaction topic so a committee member can
+    /// include them in a batch; the pre-fork path for a node that cannot seal batches.
+    pub async fn publish_txn(&self, txn: Vec<Bytes>) -> NetworkResult<()> {
         let data = encode(&WorkerGossip::Txn(txn));
         self.handle.publish(GOSSIP_TOPIC_TXN.into(), data).await?;
         Ok(())
@@ -125,6 +176,9 @@ impl WorkerNetworkHandle {
             WorkerResponse::PeerExchange { .. } => Err(NetworkError::RPCError(
                 "Got wrong response, not a report batch is peer exchange!".to_string(),
             )),
+            WorkerResponse::SubmitTxns { .. } => Err(NetworkError::RPCError(
+                "Got wrong response, not a report batch is submit txns!".to_string(),
+            )),
             WorkerResponse::Error(WorkerRPCError(s)) => Err(NetworkError::RPCError(s)),
         }
     }
@@ -195,6 +249,9 @@ impl WorkerNetworkHandle {
                 }
                 Ok(batches)
             }
+            WorkerResponse::SubmitTxns { .. } => Err(NetworkError::RPCError(
+                "Got wrong response, not a request batches is submit txns!".to_string(),
+            )),
             WorkerResponse::Error(WorkerRPCError(s)) => Err(NetworkError::RPCError(s)),
         }
     }
@@ -290,15 +347,17 @@ impl WorkerNetworkHandle {
     }
 }
 
-/// Handle inter-node communication between primaries.
+/// Event loop that handles inbound worker requests and gossip for one epoch.
 #[derive(Debug)]
 pub struct WorkerNetwork<DB, Events> {
     /// Receiver for network events.
     network_events: Events,
     /// Network handle to send commands.
     network_handle: WorkerNetworkHandle,
-    // Request handler to process requests and return responses.
+    /// Request handler to process requests and return responses.
     request_handler: RequestHandler<DB>,
+    /// Bounds concurrent inbound submit fan-out onto the shared rayon pool.
+    submit_permits: Arc<Semaphore>,
 }
 
 impl<DB, Events> WorkerNetwork<DB, Events>
@@ -316,7 +375,8 @@ where
     ) -> Self {
         let request_handler =
             RequestHandler::new(id, validator, consensus_config, network_handle.clone());
-        Self { network_events, network_handle, request_handler }
+        let submit_permits = Arc::new(Semaphore::new(MAX_CONCURRENT_SUBMIT_TXNS));
+        Self { network_events, network_handle, request_handler, submit_permits }
     }
 
     /// Run the network for the epoch.
@@ -332,9 +392,8 @@ where
         );
     }
 
-    /// Handle events concurrently.
+    /// Dispatch one network event to a task so the event loop never blocks on handling.
     fn process_network_event(&self, event: NetworkEvent<Req, Res>) {
-        // match event
         match event {
             NetworkEvent::Request { peer, request, channel, cancel } => match request {
                 WorkerRequest::ReportBatch { sealed_batch } => {
@@ -353,6 +412,9 @@ where
                     // expect this is intercepted by network layer
                     warn!(target: "worker::network", "worker application received unexpected peer exchange message");
                 }
+                WorkerRequest::SubmitTxns { transactions } => {
+                    self.process_submit_txns(transactions, channel, cancel);
+                }
             },
             NetworkEvent::Gossip(msg, gossip_source) => {
                 self.process_gossip(msg, gossip_source);
@@ -370,9 +432,50 @@ where
         }
     }
 
+    /// Process directly-submitted transactions from a forwarding peer.
+    ///
+    /// Spawns a task that admits them to the pool and replies with the stale hashes, so the sender
+    /// can prune what has already executed instead of re-forwarding it. Any connected peer may
+    /// submit (observers are not committee members, so there is no membership gate to apply) and
+    /// pool validation still rejects anything invalid; signer recovery hops to the blocking pool
+    /// inside `submit_forwarded_txns`, so a flood cannot stall the network task.
+    fn process_submit_txns(
+        &self,
+        transactions: Vec<Bytes>,
+        channel: ResponseChannel<WorkerResponse>,
+        cancel: oneshot::Receiver<()>,
+    ) {
+        // Shed the submit when the inbound fan-out is at capacity: replying with the existing error
+        // variant (no wire change) makes the sender retry the next live validator immediately
+        // instead of queueing behind a rayon backlog that would also delay consensus validation.
+        let Ok(permit) = self.submit_permits.clone().try_acquire_owned() else {
+            let network_handle = self.network_handle.clone();
+            self.network_handle.get_task_spawner().spawn_task("submit-txns-shed", async move {
+                let response = WorkerResponse::Error(WorkerRPCError("submit backlog".into()));
+                let _ = network_handle.handle.send_response(response, channel).await;
+            });
+            return;
+        };
+
+        let request_handler = self.request_handler.clone();
+        let network_handle = self.network_handle.clone();
+        self.network_handle.get_task_spawner().spawn_task("process-submit-txns", async move {
+            // hold the permit across the rayon fan-out so the cap bounds concurrent CPU work
+            let _permit = permit;
+            tokio::select! {
+                stale = request_handler.submit_forwarded_txns(transactions) => {
+                    let response = WorkerResponse::SubmitTxns { stale };
+                    let _ = network_handle.handle.send_response(response, channel).await;
+                }
+                // cancel notification from the network layer
+                _ = cancel => (),
+            }
+        });
+    }
+
     /// Process a new reported batch.
     ///
-    /// Spawn a task to evaluate a peer's proposed header and return a response.
+    /// Spawn a task to validate the peer's batch and return a response.
     fn process_report_batch(
         &self,
         peer: BlsPublicKey,
@@ -458,10 +561,10 @@ where
     }
 }
 
-/// Defines how the network receiver handles incoming primary messages.
+/// Handler for the local primary's requests to this worker.
 #[derive(Debug)]
 pub(super) struct PrimaryReceiverHandler<DB> {
-    /// The batch store
+    /// The batch store.
     pub store: DB,
     /// Timeout on RequestBatches RPC.
     pub request_batches_timeout: Duration,
@@ -469,7 +572,7 @@ pub(super) struct PrimaryReceiverHandler<DB> {
     pub network: Option<WorkerNetworkHandle>,
     /// Fetch certificate payloads from other workers.
     pub batch_fetcher: Option<BatchFetcher<DB>>,
-    /// Validate incoming batches
+    /// Validates incoming batches.
     pub validator: Arc<dyn BatchValidation>,
 }
 
diff --git a/crates/consensus/worker/src/worker.rs b/crates/consensus/worker/src/worker.rs
index d088ab8f..e94ea58d 100644
--- a/crates/consensus/worker/src/worker.rs
+++ b/crates/consensus/worker/src/worker.rs
@@ -18,8 +18,8 @@ use rayls_infrastructure_storage::tables::{BatchSeqCounter, Batches, ConsensusBl
 use rayls_infrastructure_types::{
     batch_tracker::{BatchTracker, SealFailureReason},
     error::BlockSealError,
-    AuthorityIdentifier, BatchReceiver, BatchSender, BatchValidation, Database, Epoch, SealedBatch,
-    SenderNonceRanges, TaskKind, TaskManager, WorkerId,
+    AuthorityIdentifier, BatchReceiver, BatchSender, BatchValidation, Bytes, Database, Epoch,
+    SealedBatch, SenderNonceRanges, TaskKind, TaskManager, WorkerId,
 };
 use std::{sync::Arc, time::Duration};
 use tracing::{error, info, warn};
@@ -329,7 +329,8 @@ impl<DB: Database, QW: QuorumWaiterTrait> Worker<DB, QW> {
     /// Send all the txns in sealed_batch to CVVs so they can be included in blocks.
     /// Use this when not a CVV so that transactions you accept can be included in a block.
     pub async fn disburse_txns(&self, sealed_batch: SealedBatch) -> Result<(), BlockSealError> {
-        if let Err(err) = self.network_handle.publish_txn(sealed_batch.batch.transactions).await {
+        let payloads = sealed_batch.batch.transactions.into_iter().map(Bytes::from).collect();
+        if let Err(err) = self.network_handle.publish_txn(payloads).await {
             error!(target: "worker::batch_provider", "Error publishing transaction: {err}");
         }
         Ok(())
diff --git a/crates/execution/evm/src/reth_env/accessors.rs b/crates/execution/evm/src/reth_env/accessors.rs
index 462cab80..523a55c5 100644
--- a/crates/execution/evm/src/reth_env/accessors.rs
+++ b/crates/execution/evm/src/reth_env/accessors.rs
@@ -22,14 +22,15 @@ use std::{ops::RangeInclusive, sync::Arc};
 use reth_primitives_traits::Block as _;
 
 impl RethEnv {
-    /// Initialize a new transaction pool for worker.
+    /// Initialize a new transaction pool for worker with a fresh in-flight tracker.
     pub fn init_txn_pool(&self) -> eyre::Result<WorkerTxPool> {
         self.init_txn_pool_with_in_flight(crate::in_flight::InFlightTracker::new())
     }
 
-    /// Initialize the worker transaction pool around a caller-owned in-flight tracker, so the
-    /// same tracker can be handed to the executor engine, which releases the marks of
-    /// execution-dropped transactions.
+    /// Initialize a worker transaction pool sharing the given node-scoped in-flight tracker.
+    ///
+    /// The node holds one tracker so the pool and whichever role it runs (the batch builder or the
+    /// forwarder) mark and release the same set of in-flight hashes.
     pub fn init_txn_pool_with_in_flight(
         &self,
         in_flight: crate::in_flight::InFlightTracker,
diff --git a/crates/execution/evm/src/txn_pool.rs b/crates/execution/evm/src/txn_pool.rs
index 2238e349..35cceab2 100644
--- a/crates/execution/evm/src/txn_pool.rs
+++ b/crates/execution/evm/src/txn_pool.rs
@@ -20,7 +20,7 @@ use reth_provider::{
 };
 use reth_rpc_eth_types::utils::recover_raw_transaction as reth_recover_raw_transaction;
 use reth_transaction_pool::{
-    error::{Eip4844PoolTransactionError, InvalidPoolTransactionError, PoolError},
+    error::{Eip4844PoolTransactionError, InvalidPoolTransactionError, PoolError, PoolErrorKind},
     identifier::TransactionId,
     maintain::{maintain_transaction_pool_future, MaintainPoolConfig},
     AddedTransactionOutcome, BestTransactions, CoinbaseTipOrdering, EthPooledTransaction, Pool,
@@ -118,9 +118,9 @@ pub type RaylsTransactionPool =
 pub struct WorkerTxPool {
     /// The reth pool.
     pool: RaylsTransactionPool,
-    /// Hashes sealed into a batch but not yet observed mined, so the sealer skips re-sealing
-    /// them while their batch is in flight; shared with the batch builder via
-    /// [`WorkerTxPool::in_flight`].
+    /// Node-scoped in-flight marks shared with the role the node runs: the batch builder marks
+    /// what it sealed so the next round skips it, the forwarder marks what it sent so it does not
+    /// re-send before the mark is due. Handed out via [`WorkerTxPool::in_flight`].
     in_flight_tracker: InFlightTracker,
     /// File the pending and queued transactions are snapshotted to on graceful shutdown and
     /// reloaded from on boot.
@@ -135,6 +135,10 @@ impl From<WorkerTxPool> for RaylsTransactionPool {
 
 impl WorkerTxPool {
     /// Builds the pool and spawns its node-scoped maintenance and in-flight release tasks.
+    ///
+    /// The `in_flight` tracker is node-scoped and shared with whichever role the node runs (the
+    /// batch builder's sealing marks or the forwarder's dissemination marks), so its
+    /// forwarding-role paths only come alive once a forwarder arms them.
     pub fn new(
         node_config: &NodeConfig<ChainSpec>,
         task_spawner: &TaskSpawner,
@@ -210,11 +214,11 @@ impl WorkerTxPool {
         self.pool.pending_transactions_listener_for(TransactionListenerKind::All)
     }
 
-    /// Returns the in-flight tracker shared with this pool's batch builder.
+    /// Returns the node-scoped in-flight tracker shared with the batch builder or the forwarder.
     ///
-    /// The builder marks a batch's hashes here on quorum so the next sealing round skips them
-    /// while the batch is in flight; the pool releases them again via [`Self::reconcile_in_flight`]
-    /// once execution removes them from the pending sub-pool.
+    /// The role marks hashes here (on quorum for a sealed batch, on send for a forwarded one) and
+    /// the pool releases them via [`Self::reconcile_in_flight`] once execution removes them from
+    /// the pending sub-pool.
     pub fn in_flight(&self) -> InFlightTracker {
         self.in_flight_tracker.clone()
     }
@@ -267,6 +271,31 @@ impl WorkerTxPool {
         self.pool.add_transaction(TransactionOrigin::External, tx).await
     }
 
+    /// Admit forwarded transactions and return the hashes the pool rejected as stale.
+    ///
+    /// Stale means `nonce too low`: the sender already executed that nonce, so the forwarder must
+    /// stop re-sending it. Every other rejection (fee caps, pool limits, already-imported) is not
+    /// stale: those transactions are still wanted, so they are omitted from the ack and stay
+    /// eligible for a later resend.
+    pub async fn add_forwarded_txns(&self, txs: Vec<EthPooledTransaction>) -> Vec<TxHash> {
+        self.pool
+            .add_transactions(TransactionOrigin::External, txs)
+            .await
+            .into_iter()
+            .filter_map(|res| match res {
+                Err(err)
+                    if matches!(
+                        &err.kind,
+                        PoolErrorKind::InvalidTransaction(invalid) if invalid.is_nonce_too_low()
+                    ) =>
+                {
+                    Some(err.hash)
+                }
+                _ => None,
+            })
+            .collect()
+    }
+
     /// Adds a transaction with local origin and subscribes to its events.
     pub async fn add_transaction_and_subscribe_local(
         &self,
diff --git a/crates/infrastructure/types/Cargo.toml b/crates/infrastructure/types/Cargo.toml
index 9262ae02..98ba306a 100644
--- a/crates/infrastructure/types/Cargo.toml
+++ b/crates/infrastructure/types/Cargo.toml
@@ -11,6 +11,7 @@ publish = false
 async-trait = { workspace = true }
 bcs = { workspace = true }
 bincode = { workspace = true }
+rustc-hash = { workspace = true }
 # direct reth currently for circular deps
 reth = { workspace = true }
 reth-primitives = { workspace = true }
diff --git a/crates/infrastructure/types/src/worker/sealed_batch.rs b/crates/infrastructure/types/src/worker/sealed_batch.rs
index eac0be82..3f56e153 100644
--- a/crates/infrastructure/types/src/worker/sealed_batch.rs
+++ b/crates/infrastructure/types/src/worker/sealed_batch.rs
@@ -4,11 +4,11 @@
 //! have reached quorum.
 
 use crate::{
-    crypto, encode, Address, BlockHash, Epoch, ExecHeader, TimestampSec,
+    crypto, encode, Address, BlockHash, Bytes, Epoch, ExecHeader, TimestampSec,
     ETHEREUM_BLOCK_GAS_LIMIT_56BITS, MIN_PROTOCOL_BASE_FEE,
 };
 use serde::{Deserialize, Serialize};
-use std::fmt::Debug;
+use std::{fmt::Debug, hash::Hasher as _};
 use thiserror::Error;
 
 use super::WorkerId;
@@ -213,6 +213,26 @@ pub fn max_batch_size(_epoch: Epoch) -> usize {
     2_000_000
 }
 
+/// Pre-`TransactionLoadBalancing` slot digest: read the first 8 bytes as little-endian u64.
+///
+/// Caller must ensure `input.len() >= 8`.
+pub fn legacy_slot_digest(input: &[u8]) -> u64 {
+    let mut bytes = [0u8; 8];
+    bytes.copy_from_slice(&input[0..8]);
+    u64::from_le_bytes(bytes)
+}
+
+/// `FxHasher` over `input`, the committee-slot digest for load balancing.
+///
+/// Both the forwarder (over a sender address) and the receiving validator (over the same key)
+/// call this, so they must agree byte for byte. Do NOT replace with `FxBuildHasher::hash_one`:
+/// that path writes a slice-length prefix and changes the digest.
+pub fn fxhash_slot_digest(input: &[u8]) -> u64 {
+    let mut hasher = rustc_hash::FxHasher::default();
+    hasher.write(input);
+    hasher.finish()
+}
+
 /// Visit every committee slot once, in ring order, starting from `owner`.
 ///
 /// The committee is numbered deterministically (sorted authority order), so every validator walks
@@ -273,22 +293,28 @@ impl CommitteeSlots {
     }
 }
 
-/// Defines the validation procedure for receiving either a new single transaction (from a client)
-/// of a batch of transactions (from another validator).
+/// Validation of a peer's batch and admission of transactions received from other nodes.
 ///
-/// Invalid transactions will not receive further processing.
+/// Invalid transactions receive no further processing.
 #[async_trait::async_trait]
 pub trait BatchValidation: Send + Sync + Debug {
-    /// Determines if this batch can be voted on
+    /// Determines whether this batch can be voted on.
     async fn validate_batch(&self, b: SealedBatch) -> Result<(), BatchValidationError>;
 
-    /// Submit a batch (as bytes) for inclusion in a batch.
-    /// Will only submit if the batch's first transaction maps to a slot this validator covers.
+    /// Admit a gossiped transaction message to the pool if its first transaction maps to a slot
+    /// this validator covers.
     fn submit_batch_if_mine(
         &self,
-        tx_bytes: &[Vec<u8>],
+        tx_bytes: &[Bytes],
         slots: &CommitteeSlots,
     ) -> Result<(), SubmitBatchError>;
+
+    /// Admit transactions forwarded directly by an observer, returning the hashes rejected as
+    /// stale (already executed) so the sender stops re-forwarding them.
+    ///
+    /// Takes the decoded bytes by value so the owned buffer moves into the blocking recovery task
+    /// without a copy.
+    async fn submit_forwarded_txns(&self, tx_bytes: Vec<Bytes>) -> Vec<BlockHash>;
 }
 
 /// Errors that can occur during batch submission.
diff --git a/crates/middleware/orchestrator/src/engine/mod.rs b/crates/middleware/orchestrator/src/engine/mod.rs
index ff064a0d..f1eed6db 100644
--- a/crates/middleware/orchestrator/src/engine/mod.rs
+++ b/crates/middleware/orchestrator/src/engine/mod.rs
@@ -1,19 +1,14 @@
-//! Engine mod for Rayls Node
+//! Execution layer for worker and primary roles.
 //!
-//! This module contains all execution layer implementations for worker and primary nodes.
-//!
-//! The worker's execution components track the canonical tip to construct blocks for the worker to
-//! propose. The execution state is also used to validate proposed blocks from other peers.
-//!
-//! The engine for the primary executes consensus output, extends the canonical tip, and updates the
-//! final state of the chain.
-//!
-//! The methods in this module are thread-safe wrappers for the inner type that contains logic.
+//! The worker components track the canonical tip to build batches and validate peers' batches; the
+//! primary's engine executes consensus output and extends the canonical tip. [`ExecutionNode`] is
+//! the thread-safe wrapper around the inner type holding the logic.
 
 mod node;
 mod node_builder;
 mod node_inner;
 mod rayls_builder;
+mod txn_forwarder;
 
 pub use node::*;
 pub use rayls_builder::*;
diff --git a/crates/middleware/orchestrator/src/engine/node.rs b/crates/middleware/orchestrator/src/engine/node.rs
index b69974f8..20bd15af 100644
--- a/crates/middleware/orchestrator/src/engine/node.rs
+++ b/crates/middleware/orchestrator/src/engine/node.rs
@@ -87,7 +87,7 @@ impl ExecutionNode {
         guard.respawn_worker_network_tasks(network_handle).await
     }
 
-    /// Batch maker
+    /// Spawn the batch builder for one epoch.
     pub async fn start_batch_builder(
         &self,
         worker_id: WorkerId,
@@ -112,7 +112,31 @@ impl ExecutionNode {
             .await
     }
 
-    /// Batch validator
+    /// Spawn the observer transaction forwarder for one epoch.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn start_txn_forwarder(
+        &self,
+        worker_id: WorkerId,
+        network_handle: WorkerNetworkHandle,
+        executed_anchor: watch::Receiver<ConsensusHeader>,
+        last_seen_header: watch::Receiver<ConsensusHeader>,
+        committee: Vec<BlsPublicKey>,
+        task_spawner: &TaskSpawner,
+        max_gossip_message_size: usize,
+    ) -> eyre::Result<()> {
+        let guard = self.internal.read().await;
+        guard.start_txn_forwarder(
+            worker_id,
+            network_handle,
+            executed_anchor,
+            last_seen_header,
+            committee,
+            task_spawner,
+            max_gossip_message_size,
+        )
+    }
+
+    /// Create the batch validator for one epoch.
     pub async fn new_batch_validator(
         &self,
         worker_id: &WorkerId,
diff --git a/crates/middleware/orchestrator/src/engine/node_builder.rs b/crates/middleware/orchestrator/src/engine/node_builder.rs
index a12378f7..68ef1dc0 100644
--- a/crates/middleware/orchestrator/src/engine/node_builder.rs
+++ b/crates/middleware/orchestrator/src/engine/node_builder.rs
@@ -41,8 +41,8 @@ impl ExecutionNodeBuilder {
             opt_faucet_args: self.opt_faucet_args,
             rayls_infrastructure_config: self.rayls_infrastructure_config,
             workers: Vec::default(),
-            in_flight: rayls_execution_evm::in_flight::InFlightTracker::new(),
             own_executed_sequence: None,
+            in_flight_tracker: rayls_execution_evm::in_flight::InFlightTracker::new(),
         })
     }
 }
diff --git a/crates/middleware/orchestrator/src/engine/node_inner.rs b/crates/middleware/orchestrator/src/engine/node_inner.rs
index 5b778c14..ca3b0ca8 100644
--- a/crates/middleware/orchestrator/src/engine/node_inner.rs
+++ b/crates/middleware/orchestrator/src/engine/node_inner.rs
@@ -1,7 +1,7 @@
-//! Inner-execution node components for both Worker and Primary execution.
-//!
-//! This module contains the logic for execution.
+//! Execution-layer components behind [`ExecutionNode`](super::ExecutionNode), for both worker and
+//! primary roles.
 
+use super::txn_forwarder::TxnForwarder;
 use crate::types::ExecutionError;
 use eyre::OptionExt;
 use jsonrpsee::http_client::HttpClient;
@@ -9,6 +9,7 @@ use rayls_batch_builder::{BatchBuilder, BatchBuilderConfig, OwnWatermarkReceiver
 use rayls_batch_validator::BatchValidator;
 use rayls_consensus_worker::WorkerNetworkHandle;
 use rayls_execution_evm::{
+    chainspec::RaylsHardforks,
     in_flight::InFlightTracker,
     reth_env::RethEnv,
     system_calls::EpochState,
@@ -48,13 +49,12 @@ pub(super) struct ExecutionNodeInner {
     /// Collection of execution components by worker.
     /// Index of vec is worker id.
     pub(super) workers: Vec<WorkerComponents>,
-    /// Node-scoped pool in-flight tracker, shared between the worker transaction pool and each
-    /// epoch's executor engine so execution-dropped txs are released for re-sealing from the
-    /// first epoch after boot.
-    pub(super) in_flight: InFlightTracker,
     /// This authority's executed batch-sequence watch, captured when the engine starts and handed
     /// to each batch builder so it resumes and paces sealing off its own execution progress.
     pub(super) own_executed_sequence: Option<watch::Receiver<Option<u64>>>,
+    /// Node-scoped in-flight tracker shared by the pool and whichever role the node runs (the
+    /// batch builder's sealing marks or the forwarder's dissemination marks).
+    pub(super) in_flight_tracker: InFlightTracker,
 }
 
 impl ExecutionNodeInner {
@@ -103,7 +103,7 @@ impl ExecutionNodeInner {
             engine_idle_tx,
             last_consensus_header,
             executed_batch_registry,
-            self.in_flight.clone(),
+            self.in_flight_tracker.clone(),
         );
         if let Some(tracker) = batch_tracker {
             rayls_middleware_processor.set_batch_tracker(tracker);
@@ -143,7 +143,7 @@ impl ExecutionNodeInner {
         Ok(())
     }
 
-    /// The worker's RPC, TX pool, and block builder
+    /// Spawn the worker's batch builder for one epoch.
     pub(super) async fn start_batch_builder(
         &mut self,
         worker_id: WorkerId,
@@ -203,9 +203,51 @@ impl ExecutionNodeInner {
         Ok(())
     }
 
+    /// Spawn the observer transaction forwarder for one epoch.
+    ///
+    /// `committee` is slot-ordered (authorities sorted by id) to match receiver-side dispatch. The
+    /// direct-submit path is fork-gated and re-read per tick, so an un-upgraded peer never receives
+    /// the new request and the fork may activate mid-epoch.
+    pub(super) fn start_txn_forwarder(
+        &self,
+        worker_id: WorkerId,
+        network_handle: WorkerNetworkHandle,
+        executed_anchor: watch::Receiver<ConsensusHeader>,
+        last_seen_header: watch::Receiver<ConsensusHeader>,
+        committee: Vec<BlsPublicKey>,
+        epoch_task_spawner: &TaskSpawner,
+        max_gossip_message_size: usize,
+    ) -> eyre::Result<()> {
+        let transaction_pool = self
+            .workers
+            .get(worker_id as usize)
+            .ok_or_eyre("worker components missing for {worker_id}")?
+            .pool();
+
+        let reth_env = self.reth_env.clone();
+        let direct_submit = Box::new(move || {
+            let next_block = reth_env.canonical_tip().number + 1;
+            reth_env
+                .rayls_chain_spec()
+                .is_sender_affinity_load_balancing_active_at_block(next_block)
+        });
+
+        TxnForwarder::new(
+            transaction_pool,
+            network_handle,
+            executed_anchor,
+            last_seen_header,
+            committee,
+            direct_submit,
+            max_gossip_message_size,
+        )
+        .spawn(self.rayls_infrastructure_config.parameters.max_batch_delay, epoch_task_spawner);
+
+        Ok(())
+    }
+
     /// Initialize the worker's transaction pool and public RPC.
-    /// Must call this function in accending worker_id order or will panic,
-    /// for instance call for worker id 0, then 1, etc.
+    /// Call in ascending worker_id order (0, then 1, ...); any other order panics.
     pub(super) async fn initialize_worker_components<EP>(
         &mut self,
         worker_id: WorkerId,
@@ -216,7 +258,7 @@ impl ExecutionNodeInner {
         EP: EngineToPrimary + Send + Sync + 'static,
     {
         let transaction_pool =
-            self.reth_env.init_txn_pool_with_in_flight(self.in_flight.clone())?;
+            self.reth_env.init_txn_pool_with_in_flight(self.in_flight_tracker.clone())?;
 
         let network = WorkerNetwork::new(
             self.reth_env.chainspec(),
@@ -264,7 +306,7 @@ impl ExecutionNodeInner {
 
         // take ownership of worker components
         let components = WorkerComponents::new(rpc_handle, transaction_pool, network);
-        // Must call this function in accending worker_id order or will panic.
+        // call in ascending worker_id order; any other order panics
         if worker_id as usize != self.workers.len() {
             panic!("initialize_worker_components not called with sequencial worker ids!")
         }
@@ -305,7 +347,7 @@ impl ExecutionNodeInner {
     /// Fetch the last executed state from the database.
     ///
     /// This method is called when the primary spawns to retrieve
-    /// the last committed sub dag from it's database in the case
+    /// the last committed sub dag from its database in the case
     /// of the node restarting.
     ///
     /// This returns the hash of the last executed ConsensusHeader on the consensus chain.
@@ -365,7 +407,7 @@ impl ExecutionNodeInner {
         Ok(blocks)
     }
 
-    /// Return an database provider.
+    /// Return a database provider.
     pub(super) fn get_reth_env(&self) -> RethEnv {
         self.reth_env.clone()
     }
diff --git a/crates/middleware/orchestrator/src/engine/txn_forwarder.rs b/crates/middleware/orchestrator/src/engine/txn_forwarder.rs
new file mode 100644
index 00000000..28131d4a
--- /dev/null
+++ b/crates/middleware/orchestrator/src/engine/txn_forwarder.rs
@@ -0,0 +1,576 @@
+//! Forwards a non-committee node's pending transactions to the committee.
+//!
+//! A node that cannot seal a batch still has to get its RPC-accepted transactions into consensus.
+//! Under the sender-affinity fork it submits them by request-response to the validator owning each
+//! sender's committee slot; pre-fork it publishes on the worker transaction topic. The ack lists
+//! hashes the validator rejected as stale (already executed), which the acked-stale mark suppresses
+//! from resends until local execution prunes them. Delivery is never assumed: transactions stay in
+//! the local pool, and the shared in-flight tracker re-marks anything still pending once its mark
+//! is due ([`FORWARD_POLICY`]) but only while this node is caught up, since a lagging node cannot
+//! tell a lost send from its own lag.
+
+use alloy::primitives::map::AddressMap;
+use prometheus::{
+    default_registry, register_histogram_with_registry, register_int_counter_with_registry,
+    Histogram, IntCounter, Registry,
+};
+use rayls_consensus_worker::WorkerNetworkHandle;
+use rayls_execution_evm::{
+    in_flight::{DuePolicy, ForwardMarks, ForwardProbe},
+    PoolTxn, TxPool, WorkerTxPool,
+};
+use rayls_infrastructure_types::{
+    fxhash_slot_digest, ring_walk, B256Set, BlsPublicKey, Bytes, ConsensusHeader,
+    Encodable2718 as _, TaskKind, TaskSpawner, TxHash,
+};
+use std::{
+    sync::{Arc, LazyLock},
+    time::{Duration, Instant},
+};
+use tokio::{sync::watch, time::MissedTickBehavior};
+use tracing::{debug, warn};
+
+/// Bytes charged to each transaction on top of its own length, covering the bcs length prefix that
+/// precedes it in the published vector.
+const TXN_FRAME_BYTES: usize = 5;
+
+/// Maximum transactions carried in one forward message.
+///
+/// The byte budget alone lets a ~2 MiB message hold ~19k transactions, which the receiver admits in
+/// a single call. If that admission outruns the submit timeout the sender re-sends the identical
+/// payload to the next validator, fanning the same work across the committee and multiplying
+/// duplicate seals. Capping the count bounds per-message receiver work independently of wire size.
+const MAX_TXNS_PER_MESSAGE: usize = 2_000;
+
+/// How long one direct submit waits for the validator's ack before trying the next live one.
+const SUBMIT_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Maximum blocks local execution may trail the latest seen header before re-sends stop.
+///
+/// A node that is behind advances its anchor on schedule while re-sending transactions the network
+/// executed long ago, flooding the committee with rejections. While the lag exceeds this bound loss
+/// is indistinguishable from lag, so only first sends flow; by catch-up, reconcile has pruned what
+/// executed and almost nothing is left to re-send.
+const RESEND_MAX_LOCAL_LAG: u64 = 20;
+
+/// When a forwarded transaction is due for another send.
+///
+/// The 10s base sits well above worst-case inclusion latency (~20 rounds at a 500ms round), so a
+/// healthy path never double-sends. The 20-block anchor margin makes a re-send mean "execution
+/// passed where this should have landed and it is still pending". The doubling cap bounds re-send
+/// amplification at 16x the base window. A re-send is cheap: the target pool rejects a hash it
+/// holds.
+const FORWARD_POLICY: DuePolicy =
+    DuePolicy { after: Duration::from_secs(10), backoff_shift_cap: 4, min_anchor_advance: 20 };
+
+/// Instrumentation for the per-tick pool scan, settling whether re-examining the whole pending set
+/// each tick is a real cost or noise absorbed by the interval.
+#[derive(Clone)]
+struct ForwardMetrics {
+    /// Seconds spent scanning the pending pool and grouping due transactions per tick.
+    scan_duration: Histogram,
+    /// Total pending transactions inspected across ticks (the scan's input size).
+    pending_examined: IntCounter,
+    /// Total transactions that passed the send gate across ticks (the scan's useful output).
+    forwarded: IntCounter,
+    /// Subset of `forwarded` that re-sent an already-published hash; a sustained rate is a flood.
+    resent: IntCounter,
+    /// Hashes a validator acked as already-executed (stale) on a direct submit.
+    acked_stale: IntCounter,
+    /// Ticks where re-sends were gated because local execution trailed the seen header.
+    resend_gated: IntCounter,
+}
+
+impl ForwardMetrics {
+    /// Register the family on `registry`, failing if a name is already registered there.
+    fn register(registry: &Registry) -> Result<Self, prometheus::Error> {
+        Ok(Self {
+            scan_duration: register_histogram_with_registry!(
+                "rayls_txn_forwarder_scan_duration_seconds",
+                "Seconds spent scanning the pending pool and grouping due transactions per tick",
+                vec![
+                    0.00005, 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1
+                ],
+                registry
+            )?,
+            pending_examined: register_int_counter_with_registry!(
+                "rayls_txn_forwarder_pending_examined_total",
+                "Total pending transactions inspected across forward ticks",
+                registry
+            )?,
+            forwarded: register_int_counter_with_registry!(
+                "rayls_txn_forwarder_forwarded_total",
+                "Total transactions that passed the send gate across forward ticks",
+                registry
+            )?,
+            resent: register_int_counter_with_registry!(
+                "rayls_txn_forwarder_resent_total",
+                "Sends that re-forwarded an already-published hash (a sustained rate is a flood)",
+                registry
+            )?,
+            acked_stale: register_int_counter_with_registry!(
+                "rayls_txn_forwarder_acked_stale_total",
+                "Hashes a validator acked as already-executed on a direct submit",
+                registry
+            )?,
+            resend_gated: register_int_counter_with_registry!(
+                "rayls_txn_forwarder_resend_gated_total",
+                "Ticks where re-sends were withheld because this node trailed the seen header",
+                registry
+            )?,
+        })
+    }
+
+    /// Register against a private registry, for the fallback when the default already holds the
+    /// family (a second process-wide registration).
+    fn register_fresh() -> Self {
+        Self::register(&Registry::new()).expect("a fresh registry should always succeed")
+    }
+
+    /// Record one scan tick: its duration and the pending, forwarded, and re-sent counts.
+    fn on_scan(&self, duration: Duration, pending: u64, forwarded: u64, resent: u64) {
+        self.scan_duration.observe(duration.as_secs_f64());
+        self.pending_examined.inc_by(pending);
+        self.forwarded.inc_by(forwarded);
+        self.resent.inc_by(resent);
+    }
+
+    /// Record a tick whose re-sends were withheld because this node trailed the seen header.
+    fn on_resend_gated(&self) {
+        self.resend_gated.inc();
+    }
+
+    /// Record hashes a validator acked as already-executed on a direct submit.
+    fn on_acked_stale(&self, count: u64) {
+        self.acked_stale.inc_by(count);
+    }
+}
+
+/// Registered once per process; the forwarder re-spawns each epoch but these outlive it. Falls back
+/// to a private registry if the default already holds the family, so a second registration degrades
+/// to unscraped instead of aborting.
+static FORWARD_METRICS: LazyLock<ForwardMetrics> = LazyLock::new(|| {
+    ForwardMetrics::register(default_registry())
+        .unwrap_or_else(|_| ForwardMetrics::register_fresh())
+});
+
+/// One sender's forward-pending transactions, kept whole so its nonce chain stays contiguous.
+///
+/// The slot owner (computed once when the sender is first seen) plus each transaction as
+/// `(nonce, hash, txn)`; the nonce is kept only to sort the run before it goes on the wire.
+type SenderGroup = (u64, Vec<(u64, TxHash, Arc<PoolTxn>)>);
+
+/// Epoch-scoped application actor that forwards an observer's pending transactions.
+///
+/// The pool and its forwarding marks remain node-scoped; this actor only coordinates one epoch's
+/// committee, progress signals, and transport mode.
+pub(super) struct TxnForwarder {
+    /// The shared worker transaction pool holding the pending transactions to forward.
+    pool: WorkerTxPool,
+    /// The worker network handle for direct submit and gossip publish.
+    network_handle: WorkerNetworkHandle,
+    /// The node-scoped forwarding marks driving the resend policy.
+    marks: ForwardMarks,
+    /// The highest consensus header executed locally, the anchor the resend policy advances on.
+    executed_anchor: watch::Receiver<ConsensusHeader>,
+    /// The latest consensus header this node has seen, used to gate re-sends when lagging.
+    last_seen_header: watch::Receiver<ConsensusHeader>,
+    /// The committee, slot-ordered (authorities sorted by id), matching receiver-side dispatch.
+    committee: Vec<BlsPublicKey>,
+    /// Re-evaluated each tick so the sender-affinity fork can activate mid-epoch.
+    direct_submit: Box<dyn Fn() -> bool + Send + Sync>,
+    /// Transaction-byte budget for one gossip publish.
+    gossip_budget: usize,
+    /// Transaction-byte budget for one direct submit.
+    direct_budget: usize,
+    /// Scan and delivery instrumentation, a clone of the process-wide family.
+    metrics: ForwardMetrics,
+}
+
+impl TxnForwarder {
+    /// Create a forwarder for one epoch.
+    ///
+    /// `committee` is slot-ordered (authorities sorted by id), matching receiver-side dispatch;
+    /// `direct_submit` is evaluated on each tick so the sender-affinity fork can activate during
+    /// the epoch.
+    #[allow(clippy::too_many_arguments)]
+    pub(super) fn new(
+        pool: WorkerTxPool,
+        network_handle: WorkerNetworkHandle,
+        executed_anchor: watch::Receiver<ConsensusHeader>,
+        last_seen_header: watch::Receiver<ConsensusHeader>,
+        committee: Vec<BlsPublicKey>,
+        direct_submit: Box<dyn Fn() -> bool + Send + Sync>,
+        max_gossip_message_size: usize,
+    ) -> Self {
+        // Arming keeps the sweep uninstalled: these marks are re-driven by the due check below, and
+        // a node-scoped flat sweep releasing them would erase the backoff state.
+        let marks = pool.in_flight().arm_forwarding(FORWARD_POLICY);
+        let gossip_budget = WorkerNetworkHandle::txn_payload_budget(max_gossip_message_size);
+        let direct_budget = network_handle.direct_txn_payload_budget();
+
+        Self {
+            pool,
+            network_handle,
+            marks,
+            executed_anchor,
+            last_seen_header,
+            committee,
+            direct_submit,
+            gossip_budget,
+            direct_budget,
+            metrics: FORWARD_METRICS.clone(),
+        }
+    }
+
+    /// Spawn this actor for the epoch.
+    ///
+    /// [`TaskKind::Cancel`] is correct because the actor owns no epoch-tied state: forwarding marks
+    /// remain node-scoped, so an epoch transition does not republish transactions.
+    pub(super) fn spawn(self, forward_interval: Duration, task_spawner: &TaskSpawner) {
+        task_spawner.spawn_classified_task(
+            "txn forwarder",
+            async move { self.run(forward_interval).await },
+            TaskKind::Cancel,
+        );
+    }
+
+    /// Run until the epoch task manager cancels the forwarder.
+    async fn run(self, forward_interval: Duration) {
+        // Drain at the batch cadence so a forwarded transaction reaches the committee no later than
+        // the batch path it replaces.
+        let mut interval = tokio::time::interval(forward_interval);
+        // Delay, not Burst: a stalled tick has nothing to catch up on, since the next drain sees
+        // the same pool the skipped ones would have.
+        interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
+        loop {
+            interval.tick().await;
+            self.forward_once().await;
+        }
+    }
+
+    /// Apply the forwarding policy and deliver the resulting sender runs once.
+    async fn forward_once(&self) {
+        let now = Instant::now();
+        let anchor = self.executed_anchor.borrow().number;
+        // First sends of new transactions always flow; re-sends are gated while this node lags,
+        // because a node behind the network cannot tell a lost send from its own lag and must not
+        // re-flood transactions the network may already have executed.
+        let allow_resend = Self::is_caught_up(self.last_seen_header.borrow().number, anchor);
+        if !allow_resend {
+            self.metrics.on_resend_gated();
+            debug!(target: "worker::txn_forwarder", anchor, "local execution behind the seen header; re-sends gated, first sends still flow");
+        }
+
+        // Timed region: the O(pending) CPU work (snapshot, send-gate probe, grouping), isolated
+        // from the network sends below. `select_pending` records the scan metrics itself.
+        let pool = self.pool.clone();
+        let committee = self.committee.clone();
+        let marks = self.marks.clone();
+        let metrics = self.metrics.clone();
+        let by_sender = match tokio::task::spawn_blocking(move || {
+            Self::select_pending(&pool, &marks, &committee, &metrics, now, anchor, allow_resend)
+        })
+        .await
+        {
+            Ok(by_sender) => by_sender,
+            Err(err) => {
+                warn!(target: "worker::txn_forwarder", ?err, "forward-plan preparation task failed");
+                return;
+            }
+        };
+        if by_sender.is_empty() {
+            return;
+        }
+
+        if !self.committee.is_empty() && (self.direct_submit)() {
+            self.submit_to_committee(by_sender, anchor).await;
+        } else {
+            self.publish_to_gossip(by_sender, anchor).await;
+        }
+    }
+
+    /// Select transactions due for delivery, grouped into nonce-contiguous sender runs.
+    fn select_pending(
+        pool: &WorkerTxPool,
+        marks: &ForwardMarks,
+        committee: &[BlsPublicKey],
+        metrics: &ForwardMetrics,
+        now: Instant,
+        anchor: u64,
+        allow_resend: bool,
+    ) -> AddressMap<SenderGroup> {
+        let scan_start = Instant::now();
+        let mut by_sender = AddressMap::default();
+        let mut forwarded = 0u64;
+        let mut resent = 0u64;
+        let committee_size = committee.len() as u64;
+
+        let best_transactions = {
+            let mut best_transactions = pool.best_transactions();
+            best_transactions.no_updates();
+            best_transactions
+        };
+
+        for txn in best_transactions {
+            // EIP-4844 transactions are excluded for the same reason the batch builder skips them:
+            // the blob sidecar does not travel with the encoded transaction, so the receiver cannot
+            // pool it.
+            if txn.is_eip4844() {
+                continue;
+            }
+            // One tracker probe provides both the due decision and re-send classification.
+            let probe = marks.probe(txn.hash(), now, anchor);
+            if !Self::should_send(probe, allow_resend) {
+                continue;
+            }
+            resent += u64::from(probe.forwarded);
+
+            let sender = txn.sender();
+            let (_, group) = by_sender.entry(sender).or_insert_with(|| {
+                // An empty committee has no owner; slot zero is only used by the gossip fallback.
+                let owner =
+                    fxhash_slot_digest(sender.as_slice()).checked_rem(committee_size).unwrap_or(0);
+                (owner, Vec::new())
+            });
+            group.push((txn.nonce(), *txn.hash(), txn.clone()));
+            forwarded += 1;
+        }
+
+        metrics.on_scan(scan_start.elapsed(), pool.pool_size().pending as u64, forwarded, resent);
+        by_sender
+    }
+
+    /// Deliver direct, sender-affinity streams. Owner streams proceed concurrently, but chunks
+    /// within a stream remain sequential to preserve nonce ordering across message boundaries.
+    async fn submit_to_committee(&self, by_sender: AddressMap<SenderGroup>, anchor: u64) {
+        let connected =
+            self.network_handle.inner_handle().connected_peers().await.unwrap_or_default();
+        let by_owner = Self::aggregate_by_owner(by_sender, self.committee.len());
+        let connected = &connected;
+
+        futures::future::join_all(
+            by_owner.into_iter().enumerate().filter(|(_, txns)| !txns.is_empty()).map(
+                |(owner, txns)| async move {
+                    for message in
+                        Self::chunk_under_budget(txns, self.direct_budget, MAX_TXNS_PER_MESSAGE)
+                    {
+                        self.submit_message(owner as u64, connected, message, anchor).await;
+                    }
+                },
+            ),
+        )
+        .await;
+    }
+
+    /// Deliver legacy gossip streams. A sender run must remain separate because the receiver
+    /// derives a whole message's owner from its first transaction.
+    async fn publish_to_gossip(&self, by_sender: AddressMap<SenderGroup>, anchor: u64) {
+        for (_, group) in by_sender.into_values() {
+            for message in Self::chunk_under_budget(
+                Self::nonce_sorted(group),
+                self.gossip_budget,
+                MAX_TXNS_PER_MESSAGE,
+            ) {
+                self.publish_message(message, anchor).await;
+            }
+        }
+    }
+
+    /// Submit a message to the first live validator on the ring from the owner's slot.
+    ///
+    /// A failure or timeout walks to the next connected validator; if all fail, the message stays
+    /// unstamped and the next tick retries it. An acknowledged stale hash is never re-sent while it
+    /// stays pending, whereas every accepted hash is stamped with the current retry anchor.
+    async fn submit_message(
+        &self,
+        owner: u64,
+        connected: &[BlsPublicKey],
+        message: Vec<(TxHash, Arc<PoolTxn>)>,
+        anchor: u64,
+    ) {
+        // Encode once outside the failover loop: a retry to the next validator re-uses the same
+        // hashes and clones the `Bytes` payloads (refcount bumps, not re-serialization).
+        let hashes = message.iter().map(|(hash, _)| *hash).collect::<Vec<_>>();
+        let payloads: Vec<Bytes> = message.iter().map(|(_, txn)| encode_txn(txn)).collect();
+
+        for slot in ring_walk(owner, self.committee.len() as u64) {
+            let peer = self.committee[slot as usize];
+            if !connected.contains(&peer) {
+                continue;
+            }
+            match self.network_handle.submit_txns(peer, payloads.clone(), SUBMIT_TIMEOUT).await {
+                Ok(stale) => {
+                    debug!(
+                        target: "worker::txn_forwarder",
+                        txns = message.len(),
+                        stale = stale.len(),
+                        slot,
+                        "submitted transactions"
+                    );
+                    let stale = Self::validate_stale(&hashes, stale);
+                    if stale.is_empty() {
+                        self.marks.mark_forwarded(hashes.iter().copied(), Instant::now(), anchor);
+                    } else {
+                        let accepted = hashes.iter().copied().filter(|hash| !stale.contains(hash));
+                        self.marks.mark_forwarded(accepted, Instant::now(), anchor);
+                        self.metrics.on_acked_stale(stale.len() as u64);
+                        self.marks.mark_acked_stale(stale);
+                    }
+                    return;
+                }
+                Err(e) => {
+                    debug!(target: "worker::txn_forwarder", ?e, slot, "submit failed, trying next live validator")
+                }
+            }
+        }
+        warn!(
+            target: "worker::txn_forwarder",
+            txns = message.len(),
+            "no live validator accepted the message; retrying next tick"
+        );
+    }
+
+    /// Publish one message on the worker transaction topic (the pre-fork path).
+    async fn publish_message(&self, message: Vec<(TxHash, Arc<PoolTxn>)>, anchor: u64) {
+        let (hashes, payloads): (Vec<TxHash>, Vec<Bytes>) =
+            message.into_iter().map(|(hash, txn)| (hash, encode_txn(&txn))).unzip();
+
+        match self.network_handle.publish_txn(payloads).await {
+            Ok(()) => {
+                debug!(target: "worker::txn_forwarder", txns = hashes.len(), "forwarded transactions");
+                // Stamp only what is on the wire: a failed publish stays due for the next tick.
+                self.marks.mark_forwarded(hashes, Instant::now(), anchor);
+            }
+            Err(e) => warn!(target: "worker::txn_forwarder", ?e, "failed to publish transactions"),
+        }
+    }
+
+    /// Whether local execution is close enough to the peer-latest header to allow re-sends.
+    ///
+    /// The `seen >= anchor` guard rejects an understated `seen`, which happens when
+    /// `last_consensus_header` resets at an epoch transition.
+    fn is_caught_up(seen: u64, anchor: u64) -> bool {
+        seen >= anchor && seen - anchor <= RESEND_MAX_LOCAL_LAG
+    }
+
+    /// A first send flows while syncing; a re-send additionally needs the caught-up gate.
+    fn should_send(probe: ForwardProbe, allow_resend: bool) -> bool {
+        (allow_resend || !probe.forwarded) && probe.due
+    }
+
+    /// Restrict a stale reply to hashes this node actually sent, preventing a peer from suppressing
+    /// an unrelated pending transaction.
+    fn validate_stale(hashes: &[TxHash], mut stale: Vec<TxHash>) -> B256Set {
+        if stale.is_empty() {
+            return B256Set::default();
+        }
+        if stale.len() <= 4 {
+            stale.retain(|hash| hashes.iter().any(|sent_hash| sent_hash == hash));
+            return stale.into_iter().collect();
+        }
+        let sent: B256Set = hashes.iter().copied().collect();
+        stale.into_iter().filter(|hash| sent.contains(hash)).collect()
+    }
+
+    /// Split a nonce-ordered stream by wire budget and receiver admission count.
+    fn chunk_under_budget(
+        txns: Vec<(TxHash, Arc<PoolTxn>)>,
+        budget: usize,
+        max_count: usize,
+    ) -> Vec<Vec<(TxHash, Arc<PoolTxn>)>> {
+        let mut messages = Vec::new();
+        let mut message = Vec::with_capacity(max_count.min(128));
+        let mut size = 0;
+        for (hash, txn) in txns {
+            let cost = txn.encoded_length() + TXN_FRAME_BYTES;
+            // A single oversized transaction still goes out alone; dropping it would lose it.
+            if (size + cost > budget || message.len() >= max_count) && !message.is_empty() {
+                messages.push(std::mem::take(&mut message));
+                size = 0;
+            }
+            size += cost;
+            message.push((hash, txn));
+        }
+        if !message.is_empty() {
+            messages.push(message);
+        }
+        messages
+    }
+
+    /// Concatenate each owner slot's sender runs into one nonce-ordered stream.
+    fn aggregate_by_owner(
+        by_sender: AddressMap<SenderGroup>,
+        committee_size: usize,
+    ) -> Vec<Vec<(TxHash, Arc<PoolTxn>)>> {
+        let mut by_owner = vec![Vec::new(); committee_size.max(1)];
+        for (owner, group) in by_sender.into_values() {
+            by_owner[owner as usize].extend(Self::nonce_sorted(group));
+        }
+        by_owner
+    }
+
+    /// Sort one sender's transactions by nonce and remove the nonce from the wire payload.
+    fn nonce_sorted(mut group: Vec<(u64, TxHash, Arc<PoolTxn>)>) -> Vec<(TxHash, Arc<PoolTxn>)> {
+        group.sort_unstable_by_key(|(nonce, _, _)| *nonce);
+        group.into_iter().map(|(_, hash, txn)| (hash, txn)).collect()
+    }
+}
+
+/// Encode a pooled transaction to its EIP-2718 wire bytes.
+fn encode_txn(txn: &Arc<PoolTxn>) -> Bytes {
+    let mut buf = Vec::with_capacity(txn.encoded_length());
+    txn.transaction.transaction().encode_2718(&mut buf);
+    Bytes::from(buf)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn is_caught_up_gates_on_local_lag() {
+        assert!(TxnForwarder::is_caught_up(100, 100), "level with the anchor is caught up");
+        assert!(TxnForwarder::is_caught_up(120, 100), "20 behind is at the bound, still caught up");
+        assert!(!TxnForwarder::is_caught_up(121, 100), "21 behind exceeds the bound");
+        assert!(
+            !TxnForwarder::is_caught_up(99, 100),
+            "seen below anchor (epoch reset) is not caught up"
+        );
+    }
+
+    #[test]
+    fn should_send_flows_first_sends_but_gates_resends() {
+        let first_due = ForwardProbe { forwarded: false, due: true };
+        let resend_due = ForwardProbe { forwarded: true, due: true };
+        let not_due = ForwardProbe { forwarded: true, due: false };
+
+        // a first send flows whether or not re-sends are currently allowed
+        assert!(TxnForwarder::should_send(first_due, false));
+        assert!(TxnForwarder::should_send(first_due, true));
+        // a re-send additionally needs the caught-up gate
+        assert!(!TxnForwarder::should_send(resend_due, false));
+        assert!(TxnForwarder::should_send(resend_due, true));
+        // nothing sends when the mark is not due
+        assert!(!TxnForwarder::should_send(not_due, true));
+    }
+
+    #[test]
+    fn validate_stale_keeps_only_hashes_this_node_sent() {
+        let sent = [TxHash::repeat_byte(1), TxHash::repeat_byte(2)];
+        // the peer claims a hash we sent plus one we never sent
+        let claimed = vec![TxHash::repeat_byte(2), TxHash::repeat_byte(9)];
+        let stale = TxnForwarder::validate_stale(&sent, claimed);
+        assert_eq!(stale.len(), 1);
+        assert!(stale.contains(&TxHash::repeat_byte(2)));
+        assert!(!stale.contains(&TxHash::repeat_byte(9)), "a peer cannot suppress an unsent hash");
+    }
+
+    #[test]
+    fn chunk_under_budget_splits_by_count() {
+        // only the empty case is covered: a count split needs a real pooled transaction fixture
+        let txns: Vec<(TxHash, std::sync::Arc<PoolTxn>)> = Vec::new();
+        let chunks = TxnForwarder::chunk_under_budget(txns, usize::MAX, 2);
+        assert!(chunks.is_empty(), "no transactions produce no messages");
+    }
+}
diff --git a/crates/middleware/orchestrator/src/epoch_manager/core.rs b/crates/middleware/orchestrator/src/epoch_manager/core.rs
index 4e499c24..1b4e8eda 100644
--- a/crates/middleware/orchestrator/src/epoch_manager/core.rs
+++ b/crates/middleware/orchestrator/src/epoch_manager/core.rs
@@ -624,34 +624,60 @@ where
         // starts feeding the engine (serialized, no dual delivery).
         let _ = execution_replay_completed_rx.changed().await;
 
-        // Only eligible nodes build batches, and not while a transition is pending - the outer
-        // select is about to tear this epoch down.
+        // Start nothing while a transition is pending - the outer select is about to tear this
+        // epoch down. Otherwise an active CVV builds batches and an observer forwards.
         let mode = *self.consensus_bus.node_mode().borrow();
         let transition_pending = self.consensus_bus.mode_transition().borrow().is_some();
-        if mode.is_batch_producing() && !transition_pending {
-            match self.resolve_initial_batch_seq(&worker, &primary, current_epoch).await {
-                InitialBatchSeq::Use(seq) => {
-                    // Spawn the worker-side consumer before the engine-side producer so the
-                    // batch channel has a receiver for the first sealed batch.
-                    let worker_task_manager_name = worker_task_manager_name(worker_node.id().await);
-                    worker.spawn_batch_builder(&worker_task_manager_name, &epoch_task_manager);
-                    engine
-                        .start_batch_builder(
-                            worker.id(),
-                            worker.batches_tx(),
-                            &epoch_task_manager.get_spawner(),
-                            gas_accumulator.base_fee(worker.id()),
-                            current_epoch,
-                            seq,
-                            epoch_boundary,
-                        )
-                        .await?;
-                }
-                InitialBatchSeq::Defer => {
-                    info!(target: "epoch-manager",
+        if !transition_pending {
+            if mode.is_batch_producing() {
+                match self.resolve_initial_batch_seq(&worker, &primary, current_epoch).await {
+                    InitialBatchSeq::Use(seq) => {
+                        // Spawn the worker-side consumer before the engine-side producer so the
+                        // batch channel has a receiver for the first sealed batch.
+                        let worker_task_manager_name =
+                            worker_task_manager_name(worker_node.id().await);
+                        worker.spawn_batch_builder(&worker_task_manager_name, &epoch_task_manager);
+                        engine
+                            .start_batch_builder(
+                                worker.id(),
+                                worker.batches_tx(),
+                                &epoch_task_manager.get_spawner(),
+                                gas_accumulator.base_fee(worker.id()),
+                                current_epoch,
+                                seq,
+                                epoch_boundary,
+                            )
+                            .await?;
+                    }
+                    InitialBatchSeq::Defer => {
+                        info!(target: "epoch-manager",
                         "execution replay incomplete; deferring batch builder to next epoch");
+                    }
+                    InitialBatchSeq::Shutdown => return Ok(()),
                 }
-                InitialBatchSeq::Shutdown => return Ok(()),
+            } else if mode.is_observer() {
+                // An observer cannot seal, so it forwards its RPC-accepted transactions to the
+                // committee instead of the batch builder.
+                engine
+                    .start_txn_forwarder(
+                        worker.id(),
+                        worker.network_handle(),
+                        self.consensus_bus.executed_anchor().subscribe(),
+                        // peer-derived latest header: the catch-up gate compares it against the
+                        // executed anchor so a lagging node never re-sends
+                        self.consensus_bus.last_consensus_header().subscribe(),
+                        // slot-ordered (authorities sorted by id), matching receiver-side dispatch
+                        primary
+                            .current_committee()
+                            .await
+                            .authorities()
+                            .iter()
+                            .map(|authority| *authority.protocol_key())
+                            .collect(),
+                        &epoch_task_manager.get_spawner(),
+                        network_config.libp2p_config().max_gossip_message_size,
+                    )
+                    .await?;
             }
         }
 
diff --git a/crates/middleware/orchestrator/src/tests/batch_seq_gate_tests.rs b/crates/middleware/orchestrator/src/tests/batch_seq_gate_tests.rs
index 1405a021..b864f19f 100644
--- a/crates/middleware/orchestrator/src/tests/batch_seq_gate_tests.rs
+++ b/crates/middleware/orchestrator/src/tests/batch_seq_gate_tests.rs
@@ -14,11 +14,11 @@ fn cvv_inactive_does_not_produce_batches() {
     assert!(!NodeMode::CvvInactive.is_batch_producing());
 }
 
-/// Active CVVs and observers both produce batches.
+/// Active CVVs produce batches; observers forward instead of sealing.
 #[test]
-fn active_cvv_and_observer_produce_batches() {
+fn only_active_cvv_produces_batches() {
     assert!(NodeMode::CvvActive.is_batch_producing());
-    assert!(NodeMode::Observer.is_batch_producing());
+    assert!(!NodeMode::Observer.is_batch_producing());
 }
 
 fn channels() -> (watch::Sender<bool>, watch::Sender<Option<NodeMode>>, Notifier) {

From 8627a3cc965d76cf273b337d685d805d44ffed94 Mon Sep 17 00:00:00 2001
From: Mario Ignatov <mignatov90@gmail.com>
Date: Wed, 19 Aug 2026 16:29:58 +0300
Subject: [PATCH 3/4] test: cover the forwarding mark backup restart round trip

- assert save_mark_backup persists a forwarding snapshot and load_mark_backup reloads it on a fresh pool, coming live once forwarding re-arms
- assert a sealing snapshot writes no file, guarding the contract that restoring sealing marks would wedge the builder
---
 crates/execution/evm/src/txn_pool/backup.rs | 51 +++++++++++++++++++++
 1 file changed, 51 insertions(+)

diff --git a/crates/execution/evm/src/txn_pool/backup.rs b/crates/execution/evm/src/txn_pool/backup.rs
index afa7e9d7..6ad51b2a 100644
--- a/crates/execution/evm/src/txn_pool/backup.rs
+++ b/crates/execution/evm/src/txn_pool/backup.rs
@@ -385,4 +385,55 @@ mod tests {
         assert_eq!(read.bytes, 2);
         assert!(receiver.try_recv().is_err());
     }
+
+    /// A forwarding snapshot survives a restart through the on-disk mark backup: it persists,
+    /// reloads on a fresh pool, and comes live once the forwarding role re-arms. A sealing
+    /// snapshot writes nothing, since restoring sealing marks would suppress exactly the head
+    /// transactions the committed-state seq recovery re-seals and wedge the builder into
+    /// park/force-drain cycles (see [`WorkerTxPool::save_mark_backup`]).
+    #[tokio::test]
+    async fn forwarding_marks_survive_the_backup_file_round_trip() {
+        use crate::{
+            in_flight::{DuePolicy, InFlightTracker},
+            reth_env::RethEnv,
+        };
+        use alloy::primitives::TxHash;
+        use rayls_infrastructure_types::TaskManager;
+        use std::time::{Duration, Instant};
+
+        let tmp = tempfile::tempdir().expect("temporary directory");
+        let task_manager = TaskManager::new("mark-backup-test");
+        let env = RethEnv::new_for_test(tmp.path(), &task_manager, None).await.expect("reth env");
+        let hash = TxHash::from([7u8; 32]);
+        let policy = DuePolicy::ttl(Duration::from_secs(10));
+
+        // A sealing snapshot is never persisted: no file, so nothing reloads.
+        let sealer = InFlightTracker::new();
+        let sealing_pool = env.init_txn_pool_with_in_flight(sealer.clone()).expect("sealing pool");
+        sealer.arm_sealing(policy).mark([hash]);
+        assert_eq!(
+            sealing_pool.save_mark_backup(),
+            0,
+            "a sealing snapshot must die with the process"
+        );
+        assert!(!sealing_pool.mark_backup_path().exists(), "sealing writes no backup file");
+
+        // A forwarding snapshot is persisted to the file.
+        let saver = InFlightTracker::new();
+        let saving_pool = env.init_txn_pool_with_in_flight(saver.clone()).expect("saving pool");
+        saver.arm_forwarding(policy).mark_forwarded([hash], Instant::now(), 0);
+        assert_eq!(saving_pool.save_mark_backup(), 1, "a forwarding snapshot is persisted");
+        assert!(saving_pool.mark_backup_path().exists(), "forwarding writes the backup file");
+
+        // A fresh pool on the same datadir reloads it; the first forwarding arm makes it live.
+        let loader = InFlightTracker::new();
+        let loading_pool = env.init_txn_pool_with_in_flight(loader.clone()).expect("loading pool");
+        assert_eq!(loading_pool.load_mark_backup().await, 1, "the saved mark reloads");
+        assert!(
+            !loading_pool.mark_backup_path().exists(),
+            "load deletes the backup for at-most-once replay"
+        );
+        loader.arm_forwarding(policy);
+        assert!(loader.is_in_flight(&hash), "the reloaded mark is live once forwarding re-arms");
+    }
 }

From 2661bdd763e6c07581cba56911ff0c83b439f57e Mon Sep 17 00:00:00 2001
From: Mario Ignatov <mignatov90@gmail.com>
Date: Wed, 19 Aug 2026 21:01:05 +0300
Subject: [PATCH 4/4] fix: let a non-configured observer promote and guard the
 slot modulo

- the mode-write skip treated ANY node currently in Observer mode as sticky: a staked
  validator that merely booted into Observer (catching up) could then never be promoted
  back to the committee; stickiness now keys on the operator's --observer flag
  (is_observer_sticky), the contract the skip was meant to honor
- an empty committee-slot table (a gap mid-transition) reached the dispatch modulo, and
  division by zero aborts the node - a gossip message must never be able to do that;
  submit_batch_if_mine now drops the message instead

Hardening: both paths need a mid-transition committee fault to demonstrate red, so this
lands without failing-first tests.
---
 .../src/batch-validator/src/validator.rs      | 10 +++++--
 .../orchestrator/src/epoch_manager/network.rs | 24 ++++++++++++----
 .../src/epoch_manager/transition.rs           | 28 ++++++++++---------
 3 files changed, 42 insertions(+), 20 deletions(-)

diff --git a/crates/consensus/worker/src/batch-validator/src/validator.rs b/crates/consensus/worker/src/batch-validator/src/validator.rs
index f89edc3d..5a001f9d 100644
--- a/crates/consensus/worker/src/batch-validator/src/validator.rs
+++ b/crates/consensus/worker/src/batch-validator/src/validator.rs
@@ -2,8 +2,8 @@
 
 use rayls_execution_evm::{
     bytes_to_txn, chainspec::RaylsHardforks, recover_pooled_transaction,
-    recover_signed_transaction, reth_env::RethEnv, EthPooledTransaction, FixedBytes,
-    PoolErrorKind, PoolTransaction as _, WorkerTxPool,
+    recover_signed_transaction, reth_env::RethEnv, EthPooledTransaction, FixedBytes, PoolErrorKind,
+    PoolTransaction as _, WorkerTxPool,
 };
 use rayls_infrastructure_types::{
     fxhash_slot_digest, gas_accumulator::BaseFeeContainer, legacy_slot_digest, max_batch_size,
@@ -123,6 +123,12 @@ impl BatchValidation for BatchValidator {
                 if tx.len() < 8 {
                     return Err(SubmitBatchError::InvalidTransactionBytes);
                 }
+                // An empty slot table (a committee gap mid-transition) must not reach the
+                // modulo: division by zero aborts the node, and a gossip message must never
+                // be able to do that.
+                if slots.size() == 0 {
+                    return Ok(());
+                }
                 let owner = self.slot_digest(tx) % slots.size();
                 // Under sender-affinity a down owner's senders fail over to the next live slot on
                 // the ring; pre-fork ownership is exact, with no failover.
diff --git a/crates/middleware/orchestrator/src/epoch_manager/network.rs b/crates/middleware/orchestrator/src/epoch_manager/network.rs
index 2da23da9..9ad950f9 100644
--- a/crates/middleware/orchestrator/src/epoch_manager/network.rs
+++ b/crates/middleware/orchestrator/src/epoch_manager/network.rs
@@ -48,10 +48,10 @@ pub(crate) fn decide_node_mode(
             NodeMode::CvvInactive => (NodeMode::CvvInactive, "prior-mode-inactive"),
             // Promote a dynamic observer just admitted to the committee, instead of leaving it
             // Observer forever. Reaching here means in_committee, !observer_flag, !initial_epoch,
-            // prior==Observer — the only path is "was not-in-committee last epoch, just staked in."
+            // prior==Observer - the only path is "was not-in-committee last epoch, just staked in."
             // Join as CvvInactive to catch up on the boundary without proposing/voting; the bridge
             // subscriber requests CvvActive once synced. (Staying Observer would leave a silent
-            // committee member — counted toward quorum but never certifying — stalling consensus.)
+            // committee member - counted toward quorum but never certifying - stalling consensus.)
             NodeMode::Observer => (NodeMode::CvvInactive, "joined-committee"),
         };
     }
@@ -62,6 +62,20 @@ pub(crate) fn decide_node_mode(
     }
 }
 
+/// Returns whether the mode write must be skipped because the node is an explicitly-configured
+/// observer being promoted away from Observer.
+///
+/// Stickiness is an operator contract (`--observer`), not a property of the current mode: a
+/// staked node that merely booted into Observer (e.g. while catching up) must stay promotable,
+/// or it can never rejoin the committee.
+pub(crate) fn is_observer_sticky(
+    observer_flag: bool,
+    prior_mode: NodeMode,
+    target_mode: NodeMode,
+) -> bool {
+    observer_flag && prior_mode == NodeMode::Observer && target_mode != NodeMode::Observer
+}
+
 /// Returns whether the node has executed any consensus output in the chain's history.
 ///
 /// Chain-wide by design, not per-epoch. `committed_round` is the wrong source: `reset_for_epoch`
@@ -206,10 +220,10 @@ where
         let explicit_target = *self.consensus_bus.mode_transition().borrow();
 
         // Single-validator dev chain: the sole member is always the canonical source
-        // of truth — it can never be "behind" with no peers to catch up from.
+        // of truth - it can never be "behind" with no peers to catch up from.
         // Resolve CvvActive directly without going through decide_node_mode, which
         // would otherwise apply the has-local-history -> CvvInactive branch and hang.
-        // An explicitly-configured observer is still honored (Observer is sticky) —
+        // An explicitly-configured observer is still honored (Observer is sticky)  -
         // decide_node_mode handles that below.
         #[cfg(feature = "dev-single-node-setup")]
         if in_committee && consensus_config.committee().size() == 1 && !observer_flag {
@@ -220,7 +234,7 @@ where
                 authority_id = ?consensus_config.authority_id(),
                 ?mode,
                 reason,
-                "identify_node_mode: sole committee member (dev) — boot active"
+                "identify_node_mode: sole committee member (dev) - boot active"
             );
             self.consensus_bus.node_mode().send_modify(|v| *v = mode);
             return Ok(mode);
diff --git a/crates/middleware/orchestrator/src/epoch_manager/transition.rs b/crates/middleware/orchestrator/src/epoch_manager/transition.rs
index 07ef1564..3d0bc7ca 100644
--- a/crates/middleware/orchestrator/src/epoch_manager/transition.rs
+++ b/crates/middleware/orchestrator/src/epoch_manager/transition.rs
@@ -1,6 +1,6 @@
 use crate::{
     engine::ExecutionNode,
-    epoch_manager::types::EpochManager,
+    epoch_manager::{network::is_observer_sticky, types::EpochManager},
     primary::PrimaryNode,
     types::{EpochTransitionCheckpoint, EpochTransitionPhase, ShutdownOutcome, TransitionCtx},
 };
@@ -102,27 +102,27 @@ where
     /// where they collide with that epoch's `get_missing_consensus` anchor snapshot: an output
     /// the engine finishes concurrently is dropped as stale (the demote→rejoin flap race).
     ///
-    /// We wait on the engine's own `engine_idle` signal — `pending_task.is_none() &&
-    /// queued.is_empty()`, i.e. it has executed everything it admitted — NOT on the recorded
+    /// We wait on the engine's own `engine_idle` signal - `pending_task.is_none() &&
+    /// queued.is_empty()`, i.e. it has executed everything it admitted - NOT on the recorded
     /// consensus tip. The producers are stopped before this runs (and the forwarder was cancelled
     /// by the run_epoch select even earlier), so nothing new can be admitted: the engine just
     /// finishes its fixed admitted queue and reports idle. Outputs recorded but never admitted
     /// (stranded in `consensus_output` when the forwarder was cancelled) are not in the engine,
-    /// so they must NOT be waited on — they replay cleanly via the next epoch's
+    /// so they must NOT be waited on - they replay cleanly via the next epoch's
     /// `get_missing_consensus` (unadmitted ⇒ not stale-dropped).
     ///
     /// Normally this returns the instant the engine reports idle (immediately if it already is).
     /// The engine also publishes idle when its task exits, so a dying engine (shutdown /
     /// ConsensusFork / stream-close) unblocks us promptly. `backstop` only guards a genuinely
-    /// *hung* engine that never publishes — it bounds the wait (matching the rest of the
+    /// *hung* engine that never publishes - it bounds the wait (matching the rest of the
     /// transition pipeline) and warns rather than stalling the transition forever.
     async fn drain_engine_backlog(&self, backstop: Duration) {
         let mut idle_rx = self.consensus_bus.engine_idle().subscribe();
         let wait = async {
             // The engine publishes idle=true when its queue empties (poll Pending path) and also
             // when its task exits (node_inner publishes on exit), so a dying engine unblocks us
-            // promptly. `changed()` itself won't error on engine-task exit — the bus retains
-            // `tx_engine_idle` for the node's lifetime, so the channel never closes — hence the
+            // promptly. `changed()` itself won't error on engine-task exit - the bus retains
+            // `tx_engine_idle` for the node's lifetime, so the channel never closes - hence the
             // timeout below is the real backstop for a genuinely hung engine that never publishes.
             while !*idle_rx.borrow_and_update() {
                 if idle_rx.changed().await.is_err() {
@@ -136,7 +136,7 @@ where
                 target: "epoch-manager",
                 ?backstop,
                 "engine-idle drain backstop fired before mode transition; engine did not report \
-                 idle in time — proceeding (next epoch's get_missing_consensus replay covers any \
+                 idle in time - proceeding (next epoch's get_missing_consensus replay covers any \
                  unexecuted remainder)"
             );
         }
@@ -293,7 +293,7 @@ where
         // next run_epoch starts with a settled anchor and get_missing_consensus has nothing
         // in-flight to race (the demote→rejoin flap stale-drop). Producers are stopped
         // above; we wait on the engine's idle signal (executed == admitted), NOT the
-        // recorded tip — stranded-but-unadmitted outputs replay cleanly next epoch. Bounded by a
+        // recorded tip - stranded-but-unadmitted outputs replay cleanly next epoch. Bounded by a
         // backstop so a hung engine warns and proceeds instead of stalling the transition forever.
         self.drain_engine_backlog(Duration::from_secs(15)).await;
 
@@ -320,8 +320,10 @@ where
         info!(target: "epoch-manager", ?target_mode, "mode-change phase 2/3: PERSISTENCE_FLUSH");
 
         // Phase 3: APPLY - switch node mode, clear transition request, clear guard.
-        // skip the mode write if caller tried to demote Observer
-        if prior_mode == NodeMode::Observer && target_mode != NodeMode::Observer {
+        // skip the mode write only for a CONFIGURED observer: a node that merely booted into
+        // Observer mode (catch-up) must stay promotable or it can never rejoin the committee
+        let observer_flag = self.builder.rayls_infrastructure_config.observer;
+        if is_observer_sticky(observer_flag, prior_mode, target_mode) {
             warn!(target: "epoch-manager", ?target_mode, "mode-change phase 3/3: APPLY (mode write skipped - Observer is sticky)");
         } else {
             self.consensus_bus.node_mode().send_replace(target_mode);
@@ -375,8 +377,8 @@ where
                 // Decide from the durable execution state: the boundary output runs concludeEpoch,
                 // so if the closing epoch executed, the canonical tip's epoch state has advanced
                 // past it. (The former recently_executed_blocks/parent_beacon fast-path was dead
-                // here — recently_executed_blocks is empty this early in startup,
-                // before any replay — and fragile: a drained parked batch makes the
+                // here - recently_executed_blocks is empty this early in startup,
+                // before any replay - and fragile: a drained parked batch makes the
                 // tip's beacon differ from target_hash.)
                 let tip_state = engine.epoch_state_from_canonical_tip().await?;
                 let execution_done = tip_state.epoch > epoch;