Support receiving verified transactions from the vortexor (#5321)

* Support receiving verified transactions from the vortexor

* vortexor receiver adapter

* Update some comments

* Update some comments

* Gaurd against 0 packets

* update some debug messages

* disable regular tpu streamers if vortexor is in-place

* disable regular tpu streamers if vortexor is in-place

* Addressed some comments from Andrew. Hide the tpu-vortexor-receiver-address argument by default

Cargo.lock

@@ -7574,6 +7574,7 @@ dependencies = [
  "solana-unified-scheduler-logic",
  "solana-unified-scheduler-pool",
  "solana-version",
+ "solana-vortexor-receiver",
  "solana-vote",
  "solana-vote-program",
  "solana-wen-restart",
@@ -11137,6 +11138,15 @@ dependencies = [
  "x509-parser",
 ]
 
+[[package]]
+name = "solana-vortexor-receiver"
+version = "2.3.0"
+dependencies = [
+ "assert_matches",
+ "solana-perf",
+ "solana-streamer",
+]
+
 [[package]]
 name = "solana-vote"
 version = "2.3.0"

Cargo.toml

@@ -154,6 +154,7 @@ members = [
     "validator",
     "version",
     "vortexor",
+    "vortexor-receiver",
     "vote",
     "watchtower",
     "wen-restart",
@@ -572,6 +573,7 @@ solana-type-overrides = { path = "type-overrides", version = "=2.3.0" }
 solana-udp-client = { path = "udp-client", version = "=2.3.0" }
 solana-validator-exit = "2.2.1"
 solana-version = { path = "version", version = "=2.3.0" }
+solana-vortexor-receiver = { path = "vortexor-receiver", version = "=2.3.0" }
 solana-vote = { path = "vote", version = "=2.3.0" }
 solana-vote-interface = "2.2.4"
 solana-vote-program = { path = "programs/vote", version = "=2.3.0", default-features = false }

core/Cargo.toml

@@ -98,6 +98,7 @@ solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
 solana-unified-scheduler-pool = { workspace = true }
 solana-version = { workspace = true }
+solana-vortexor-receiver = { workspace = true }
 solana-vote = { workspace = true }
 solana-vote-program = { workspace = true }
 solana-wen-restart = { workspace = true }

core/src/lib.rs

@@ -41,6 +41,7 @@ mod tpu_entry_notifier;
 pub mod tvu;
 pub mod unfrozen_gossip_verified_vote_hashes;
 pub mod validator;
+mod vortexor_receiver_adapter;
 pub mod vote_simulator;
 pub mod voting_service;
 pub mod warm_quic_cache_service;

core/src/tpu.rs

@@ -23,6 +23,7 @@ use {
         staked_nodes_updater_service::StakedNodesUpdaterService,
         tpu_entry_notifier::TpuEntryNotifier,
         validator::{BlockProductionMethod, GeneratorConfig, TransactionStructure},
+        vortexor_receiver_adapter::VortexorReceiverAdapter,
     },
     bytes::Bytes,
     crossbeam_channel::{bounded, unbounded, Receiver},
@@ -73,18 +74,34 @@ pub struct TpuSockets {
     pub vote_quic: Vec<UdpSocket>,
     /// Client-side socket for the forwarding votes.
     pub vote_forwards_client: UdpSocket,
+    pub vortexor_receivers: Option<Vec<UdpSocket>>,
+}
+
+/// The `SigVerifier` enum is used to determine whether to use a local or remote signature verifier.
+enum SigVerifier {
+    Local(SigVerifyStage),
+    Remote(VortexorReceiverAdapter),
+}
+
+impl SigVerifier {
+    fn join(self) -> thread::Result<()> {
+        match self {
+            SigVerifier::Local(sig_verify_stage) => sig_verify_stage.join(),
+            SigVerifier::Remote(vortexor_receiver_adapter) => vortexor_receiver_adapter.join(),
+        }
+    }
 }
 
 pub struct Tpu {
     fetch_stage: FetchStage,
-    sigverify_stage: SigVerifyStage,
+    sig_verifier: SigVerifier,
     vote_sigverify_stage: SigVerifyStage,
     banking_stage: BankingStage,
     forwarding_stage: JoinHandle<()>,
     cluster_info_vote_listener: ClusterInfoVoteListener,
     broadcast_stage: BroadcastStage,
-    tpu_quic_t: thread::JoinHandle<()>,
-    tpu_forwards_quic_t: thread::JoinHandle<()>,
+    tpu_quic_t: Option<thread::JoinHandle<()>>,
+    tpu_forwards_quic_t: Option<thread::JoinHandle<()>>,
     tpu_entry_notifier: Option<TpuEntryNotifier>,
     staked_nodes_updater_service: StakedNodesUpdaterService,
     tracer_thread_hdl: TracerThread,
@@ -143,6 +160,7 @@ impl Tpu {
             transactions_forwards_quic: transactions_forwards_quic_sockets,
             vote_quic: tpu_vote_quic_sockets,
             vote_forwards_client: vote_forwards_client_socket,
+            vortexor_receivers,
         } = sockets;
 
         let (packet_sender, packet_receiver) = unbounded();
@@ -196,47 +214,75 @@ impl Tpu {
         )
         .unwrap();
 
-        // Streamer for TPU
-        let SpawnServerResult {
-            endpoints: _,
-            thread: tpu_quic_t,
-            key_updater,
-        } = spawn_server_multi(
-            "solQuicTpu",
-            "quic_streamer_tpu",
-            transactions_quic_sockets,
-            keypair,
-            packet_sender,
-            exit.clone(),
-            staked_nodes.clone(),
-            tpu_quic_server_config,
-        )
-        .unwrap();
+        let (tpu_quic_t, key_updater) = if vortexor_receivers.is_none() {
+            // Streamer for TPU
+            let SpawnServerResult {
+                endpoints: _,
+                thread: tpu_quic_t,
+                key_updater,
+            } = spawn_server_multi(
+                "solQuicTpu",
+                "quic_streamer_tpu",
+                transactions_quic_sockets,
+                keypair,
+                packet_sender,
+                exit.clone(),
+                staked_nodes.clone(),
+                tpu_quic_server_config,
+            )
+            .unwrap();
+            (Some(tpu_quic_t), Some(key_updater))
+        } else {
+            (None, None)
+        };
 
-        // Streamer for TPU forward
-        let SpawnServerResult {
-            endpoints: _,
-            thread: tpu_forwards_quic_t,
-            key_updater: forwards_key_updater,
-        } = spawn_server_multi(
-            "solQuicTpuFwd",
-            "quic_streamer_tpu_forwards",
-            transactions_forwards_quic_sockets,
-            keypair,
-            forwarded_packet_sender,
-            exit.clone(),
-            staked_nodes.clone(),
-            tpu_fwd_quic_server_config,
-        )
-        .unwrap();
+        let (tpu_forwards_quic_t, forwards_key_updater) = if vortexor_receivers.is_none() {
+            // Streamer for TPU forward
+            let SpawnServerResult {
+                endpoints: _,
+                thread: tpu_forwards_quic_t,
+                key_updater: forwards_key_updater,
+            } = spawn_server_multi(
+                "solQuicTpuFwd",
+                "quic_streamer_tpu_forwards",
+                transactions_forwards_quic_sockets,
+                keypair,
+                forwarded_packet_sender,
+                exit.clone(),
+                staked_nodes.clone(),
+                tpu_fwd_quic_server_config,
+            )
+            .unwrap();
+            (Some(tpu_forwards_quic_t), Some(forwards_key_updater))
+        } else {
+            (None, None)
+        };
 
         let (forward_stage_sender, forward_stage_receiver) = bounded(1024);
-        let sigverify_stage = {
+        let sig_verifier = if let Some(vortexor_receivers) = vortexor_receivers {
+            info!("starting vortexor adapter");
+            let sockets = vortexor_receivers.into_iter().map(Arc::new).collect();
+            let adapter = VortexorReceiverAdapter::new(
+                sockets,
+                Duration::from_millis(5),
+                tpu_coalesce,
+                non_vote_sender,
+                enable_block_production_forwarding.then(|| forward_stage_sender.clone()),
+                exit.clone(),
+            );
+            SigVerifier::Remote(adapter)
+        } else {
+            info!("starting regular sigverify stage");
             let verifier = TransactionSigVerifier::new(
                 non_vote_sender,
                 enable_block_production_forwarding.then(|| forward_stage_sender.clone()),
             );
-            SigVerifyStage::new(packet_receiver, verifier, "solSigVerTpu", "tpu-verifier")
+            SigVerifier::Local(SigVerifyStage::new(
+                packet_receiver,
+                verifier,
+                "solSigVerTpu",
+                "tpu-verifier",
+            ))
         };
 
         let vote_sigverify_stage = {
@@ -320,10 +366,18 @@ impl Tpu {
             turbine_quic_endpoint_sender,
         );
 
+        let mut key_updaters: Vec<Arc<dyn NotifyKeyUpdate + Send + Sync>> = Vec::new();
+        if let Some(key_updater) = key_updater {
+            key_updaters.push(key_updater);
+        }
+        if let Some(forwards_key_updater) = forwards_key_updater {
+            key_updaters.push(forwards_key_updater);
+        }
+        key_updaters.push(vote_streamer_key_updater);
         (
             Self {
                 fetch_stage,
-                sigverify_stage,
+                sig_verifier,
                 vote_sigverify_stage,
                 banking_stage,
                 forwarding_stage,
@@ -336,21 +390,21 @@ impl Tpu {
                 tracer_thread_hdl,
                 tpu_vote_quic_t,
             },
-            vec![key_updater, forwards_key_updater, vote_streamer_key_updater],
+            key_updaters,
         )
     }
 
     pub fn join(self) -> thread::Result<()> {
         let results = vec![
             self.fetch_stage.join(),
-            self.sigverify_stage.join(),
+            self.sig_verifier.join(),
             self.vote_sigverify_stage.join(),
             self.cluster_info_vote_listener.join(),
             self.banking_stage.join(),
             self.forwarding_stage.join(),
             self.staked_nodes_updater_service.join(),
-            self.tpu_quic_t.join(),
-            self.tpu_forwards_quic_t.join(),
+            self.tpu_quic_t.map_or(Ok(()), |t| t.join()),
+            self.tpu_forwards_quic_t.map_or(Ok(()), |t| t.join()),
             self.tpu_vote_quic_t.join(),
         ];
         let broadcast_result = self.broadcast_stage.join();

core/src/validator.rs

@@ -1580,6 +1580,7 @@ impl Validator {
                 transactions_forwards_quic: node.sockets.tpu_forwards_quic,
                 vote_quic: node.sockets.tpu_vote_quic,
                 vote_forwards_client: node.sockets.tpu_vote_forwards_client,
+                vortexor_receivers: node.sockets.vortexor_receivers,
             },
             &rpc_subscriptions,
             transaction_status_sender,

core/src/vortexor_receiver_adapter.rs

@@ -0,0 +1,131 @@
+//! Vortexor receiver adapter which wraps the VerifiedPacketReceiver
+//! to receive packet batches from the remote and sends the packets to the
+//! banking stage.
+
+use {
+    crate::banking_trace::TracedSender,
+    agave_banking_stage_ingress_types::BankingPacketBatch,
+    crossbeam_channel::{unbounded, Receiver, RecvTimeoutError, Sender},
+    solana_perf::packet::PacketBatch,
+    solana_vortexor_receiver::receiver::VerifiedPacketReceiver,
+    std::{
+        net::UdpSocket,
+        sync::{atomic::AtomicBool, Arc},
+        thread::{self, Builder, JoinHandle},
+        time::{Duration, Instant},
+    },
+};
+
+#[inline]
+fn send(sender: &TracedSender, batch: Arc<Vec<PacketBatch>>, count: usize) -> Result<(), String> {
+    match sender.send(batch) {
+        Ok(_) => {
+            trace!("Sent batch: {count} received from vortexor successfully");
+            Ok(())
+        }
+        Err(err) => Err(format!("Failed to send batch {count} down {err:?}")),
+    }
+}
+
+pub struct VortexorReceiverAdapter {
+    thread_hdl: JoinHandle<()>,
+    receiver: VerifiedPacketReceiver,
+}
+
+const MAX_PACKET_BATCH_SIZE: usize = 8;
+
+impl VortexorReceiverAdapter {
+    pub fn new(
+        sockets: Vec<Arc<UdpSocket>>,
+        recv_timeout: Duration,
+        tpu_coalesce: Duration,
+        packets_sender: TracedSender,
+        forward_stage_sender: Option<Sender<(BankingPacketBatch, bool)>>,
+        exit: Arc<AtomicBool>,
+    ) -> Self {
+        let (batch_sender, batch_receiver) = unbounded();
+
+        let receiver =
+            VerifiedPacketReceiver::new(sockets, &batch_sender, tpu_coalesce, None, exit.clone());
+
+        let thread_hdl = Builder::new()
+            .name("vtxRcvAdptr".to_string())
+            .spawn(move || {
+                if let Err(msg) = Self::recv_send(
+                    batch_receiver,
+                    recv_timeout,
+                    MAX_PACKET_BATCH_SIZE,
+                    packets_sender,
+                    forward_stage_sender,
+                ) {
+                    info!("Quiting VortexorReceiverAdapter: {msg}");
+                }
+            })
+            .unwrap();
+        Self {
+            thread_hdl,
+            receiver,
+        }
+    }
+
+    pub fn join(self) -> thread::Result<()> {
+        self.thread_hdl.join()?;
+        self.receiver.join()
+    }
+
+    fn recv_send(
+        packet_batch_receiver: Receiver<PacketBatch>,
+        recv_timeout: Duration,
+        batch_size: usize,
+        traced_sender: TracedSender,
+        forward_stage_sender: Option<Sender<(BankingPacketBatch, bool)>>,
+    ) -> Result<(), String> {
+        loop {
+            match Self::receive_until(packet_batch_receiver.clone(), recv_timeout, batch_size) {
+                Ok(packet_batch) => {
+                    let count = packet_batch.len();
+                    // Send out packet batches
+                    if let Some(forward_stage_sender) = &forward_stage_sender {
+                        send(&traced_sender, packet_batch.clone(), count)?;
+                        // Send out packet batches to forward stage
+                        let _ = forward_stage_sender
+                            .try_send((packet_batch, false /* reject non-vote */));
+                    } else {
+                        send(&traced_sender, packet_batch, count)?;
+                    }
+                }
+                Err(err) => match err {
+                    RecvTimeoutError::Timeout => {
+                        continue;
+                    }
+                    RecvTimeoutError::Disconnected => {
+                        return Err("Disconnected from the input channel".to_string());
+                    }
+                },
+            }
+        }
+    }
+
+    /// Receives packet batches from VerifiedPacketReceiver with a timeout
+    fn receive_until(
+        packet_batch_receiver: Receiver<PacketBatch>,
+        recv_timeout: Duration,
+        batch_size: usize,
+    ) -> Result<BankingPacketBatch, RecvTimeoutError> {
+        let start = Instant::now();
+
+        let message = packet_batch_receiver.recv_timeout(recv_timeout)?;
+        let mut packet_batches = Vec::new();
+        packet_batches.push(message);
+
+        while let Ok(message) = packet_batch_receiver.try_recv() {
+            packet_batches.push(message);
+
+            if start.elapsed() >= recv_timeout || packet_batches.len() >= batch_size {
+                break;
+            }
+        }
+
+        Ok(Arc::new(packet_batches))
+    }
+}

gossip/src/cluster_info.rs

@@ -2370,6 +2370,7 @@ pub struct Sockets {
     pub quic_vote_client: UdpSocket,
     /// Client-side socket for RPC/SendTransactionService.
     pub rpc_sts_client: UdpSocket,
+    pub vortexor_receivers: Option<Vec<UdpSocket>>,
 }
 
 pub struct NodeConfig {
@@ -2378,6 +2379,8 @@ pub struct NodeConfig {
     pub bind_ip_addr: IpAddr,
     pub public_tpu_addr: Option<SocketAddr>,
     pub public_tpu_forwards_addr: Option<SocketAddr>,
+    pub vortexor_receiver_addr: Option<SocketAddr>,
+
     /// The number of TVU receive sockets to create
     pub num_tvu_receive_sockets: NonZeroUsize,
     /// The number of TVU retransmit sockets to create
@@ -2538,6 +2541,7 @@ impl Node {
                 quic_forwards_client,
                 quic_vote_client,
                 rpc_sts_client,
+                vortexor_receivers: None,
             },
         }
     }
@@ -2712,6 +2716,7 @@ impl Node {
                 quic_vote_client,
                 quic_forwards_client,
                 rpc_sts_client,
+                vortexor_receivers: None,
             },
         }
     }
@@ -2726,6 +2731,7 @@ impl Node {
             num_tvu_receive_sockets,
             num_tvu_retransmit_sockets,
             num_quic_endpoints,
+            vortexor_receiver_addr,
         } = config;
 
         let (gossip_port, (gossip, ip_echo)) =
@@ -2845,6 +2851,23 @@ impl Node {
         info.set_serve_repair(QUIC, (addr, serve_repair_quic_port))
             .unwrap();
 
+        let vortexor_receivers = vortexor_receiver_addr.map(|vortexor_receiver_addr| {
+            multi_bind_in_range_with_config(
+                vortexor_receiver_addr.ip(),
+                (
+                    vortexor_receiver_addr.port(),
+                    vortexor_receiver_addr.port() + 1,
+                ),
+                socket_config_reuseport,
+                32,
+            )
+            .unwrap_or_else(|_| {
+                panic!("Could not bind to the set vortexor_receiver_addr {vortexor_receiver_addr}")
+            })
+            .1
+        });
+
+        info!("vortexor_receivers is {vortexor_receivers:?}");
         trace!("new ContactInfo: {:?}", info);
         let sockets = Sockets {
             gossip,
@@ -2869,6 +2892,7 @@ impl Node {
             quic_vote_client,
             quic_forwards_client,
             rpc_sts_client,
+            vortexor_receivers,
         };
         info!("Bound all network sockets as follows: {:#?}", &sockets);
         Node { info, sockets }
@@ -3344,6 +3368,7 @@ mod tests {
             num_tvu_receive_sockets: MINIMUM_NUM_TVU_RECEIVE_SOCKETS,
             num_tvu_retransmit_sockets: MINIMUM_NUM_TVU_RECEIVE_SOCKETS,
             num_quic_endpoints: DEFAULT_NUM_QUIC_ENDPOINTS,
+            vortexor_receiver_addr: None,
         };
 
         let node = Node::new_with_external_ip(&solana_pubkey::new_rand(), config);
@@ -3368,6 +3393,7 @@ mod tests {
             num_tvu_receive_sockets: MINIMUM_NUM_TVU_RECEIVE_SOCKETS,
             num_tvu_retransmit_sockets: MINIMUM_NUM_TVU_RECEIVE_SOCKETS,
             num_quic_endpoints: DEFAULT_NUM_QUIC_ENDPOINTS,
+            vortexor_receiver_addr: None,
         };
 
         let node = Node::new_with_external_ip(&solana_pubkey::new_rand(), config);

programs/sbf/Cargo.lock

@@ -5912,6 +5912,7 @@ dependencies = [
  "solana-turbine",
  "solana-unified-scheduler-pool",
  "solana-version",
+ "solana-vortexor-receiver",
  "solana-vote",
  "solana-vote-program",
  "solana-wen-restart",
@@ -9168,6 +9169,14 @@ dependencies = [
  "solana-serde-varint",
 ]
 
+[[package]]
+name = "solana-vortexor-receiver"
+version = "2.3.0"
+dependencies = [
+ "solana-perf",
+ "solana-streamer",
+]
+
 [[package]]
 name = "solana-vote"
 version = "2.3.0"

svm/examples/Cargo.lock

@@ -5770,6 +5770,7 @@ dependencies = [
  "solana-turbine",
  "solana-unified-scheduler-pool",
  "solana-version",
+ "solana-vortexor-receiver",
  "solana-vote",
  "solana-vote-program",
  "solana-wen-restart",
@@ -8509,6 +8510,14 @@ dependencies = [
  "solana-serde-varint",
 ]
 
+[[package]]
+name = "solana-vortexor-receiver"
+version = "2.3.0"
+dependencies = [
+ "solana-perf",
+ "solana-streamer",
+]
+
 [[package]]
 name = "solana-vote"
 version = "2.3.0"

validator/src/commands/run/args.rs

@@ -377,6 +377,15 @@ pub fn add_args<'a>(app: App<'a, 'a>, default_args: &'a DefaultArgs) -> App<'a,
                  --entrypoint or localhostwhen --entrypoint is not provided]",
             ),
     )
+    .arg(
+        Arg::with_name("tpu_vortexor_receiver_address")
+            .long("tpu-vortexor-receiver-address")
+            .value_name("HOST:PORT")
+            .takes_value(true)
+            .hidden(hidden_unless_forced())
+            .validator(solana_net_utils::is_host_port)
+            .help("TPU Vortexor Receiver address to which verified transaction packet will be forwarded."),
+    )
     .arg(
         Arg::with_name("public_rpc_addr")
             .long("public-rpc-address")

validator/src/commands/run/execute.rs

@@ -1140,6 +1140,19 @@ pub fn execute(
         })
         .transpose()?;
 
+    let tpu_vortexor_receiver_address =
+        matches
+            .value_of("tpu_vortexor_receiver_address")
+            .map(|tpu_vortexor_receiver_address| {
+                solana_net_utils::parse_host_port(tpu_vortexor_receiver_address).unwrap_or_else(
+                    |err| {
+                        eprintln!("Failed to parse --tpu-vortexor-receiver-address: {err}");
+                        exit(1);
+                    },
+                )
+            });
+
+    info!("tpu_vortexor_receiver_address is {tpu_vortexor_receiver_address:?}");
     let num_quic_endpoints = value_t_or_exit!(matches, "num_quic_endpoints", NonZeroUsize);
 
     let tpu_max_connections_per_peer =
@@ -1166,6 +1179,7 @@ pub fn execute(
         num_tvu_receive_sockets: tvu_receive_threads,
         num_tvu_retransmit_sockets: tvu_retransmit_threads,
         num_quic_endpoints,
+        vortexor_receiver_addr: tpu_vortexor_receiver_address,
     };
 
     let cluster_entrypoints = entrypoint_addrs

vortexor-receiver/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "solana-vortexor-receiver"
+description = "Solana TPU Vortexor Receiver"
+documentation = "https://docs.rs/solana-vortexor-receiver"
+publish = false
+version = { workspace = true }
+authors = { workspace = true }
+repository = { workspace = true }
+homepage = { workspace = true }
+license = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+solana-perf = { workspace = true }
+solana-streamer = { workspace = true }
+
+[dev-dependencies]
+assert_matches = { workspace = true }
+solana-streamer = { workspace = true, features = ["dev-context-only-utils"] }
+
+[lib]
+crate-type = ["lib"]
+name = "solana_vortexor_receiver"
+
+[package.metadata.docs.rs]
+targets = ["x86_64-unknown-linux-gnu"]

vortexor-receiver/Readme.md

@@ -0,0 +1,60 @@
+# Introduction
+The Vortexor is a service that can offload the tasks of receiving transactions
+from the public, performing signature verifications, and deduplications from the
+core validator, enabling it to focus on processing and executing the
+transactions. The verified and filtered transactions will then be forwarded to
+the validators linked with the Vortexor. This setup makes the TPU transaction
+ingestion and verification more scalable compared to a single-node solution.
+
+This module implements the VerifiedPacketReceiver in the below architecture
+which encapsulates the functionality of receiving the verified packet batches
+from the vortexor. In the first impelementation, we use UDP to receive the
+verified packets from the vortexor. It is designed to support other protocol
+option such as using QUIC.
+
+# Architecture
+Figure 1 describes the architecture diagram of the Vortexor and its
+relationship with the validator.
+
+                     +---------------------+
+                     |   Solana            |
+                     |   RPC / Web Socket  |
+                     |   Service           |
+                     +---------------------+
+                                |
+                                v
+                    +--------------------- VORTEXOR ------------------------+
+                    |           |                                           |
+                    |   +------------------+                                |
+                    |   | StakedKeyUpdater |                                |
+                    |   +------------------+                                |
+                    |           |                                           |
+                    |           v                                           |
+                    |   +-------------+        +--------------------+       |
+        TPU -->     |   | TPU Streamer| -----> | SigVerifier/Dedup  |       |
+        /QUIC       |   +-------------+        +--------------------+       |
+                    |        |                          |                   |
+                    |        v                          v                   |
+                    |  +----------------+     +------------------------+    |
+                    |  | Subscription   |<----| VerifiedPacketForwarder|    |
+                    |  | Management     |     +------------------------+    |
+                    |  +----------------+            |                      |
+                    +--------------------------------|----------------------+
+                                ^                    | (UDP/QUIC)
+    Heartbeat/subscriptions     |                    |
+                                |                    v
+                    +-------------------- AGAVE VALIDATOR ------------------+
+                    |                                                       |
+                    |  +----------------+      +-----------------------+    |
+          Config->  |  | Subscription   |      | VerifiedPacketReceiver|    |
+      Admin RPC     |  | Management     |      |                       |    |
+                    |  +----------------+      +-----------------------+    |
+                    |        |                           |                  |
+                    |        |                           v                  |
+                    |        v                      +-----------+           |
+                    |  +--------------------+       | Banking   |           |
+    Gossip <--------|--| Gossip/Contact Info|       | Stage     |           |
+                    |  +--------------------+       +-----------+           |
+                    +-------------------------------------------------------+
+
+                                       Figure 1.

vortexor-receiver/src/lib.rs

@@ -0,0 +1 @@
+pub mod receiver;

vortexor-receiver/src/receiver.rs

@@ -0,0 +1,58 @@
+/// This is responsible for receiving the verified and deduplicated transactions
+/// from the vortexor and sending down to the banking stage.
+use {
+    solana_perf::{packet::PacketBatchRecycler, recycler::Recycler},
+    solana_streamer::streamer::{self, PacketBatchSender, StreamerReceiveStats},
+    std::{
+        net::UdpSocket,
+        sync::{atomic::AtomicBool, Arc},
+        thread::{self, JoinHandle},
+        time::Duration,
+    },
+};
+
+pub struct VerifiedPacketReceiver {
+    thread_hdls: Vec<JoinHandle<()>>,
+}
+
+impl VerifiedPacketReceiver {
+    pub fn new(
+        sockets: Vec<Arc<UdpSocket>>,
+        sender: &PacketBatchSender,
+        coalesce: Duration,
+        in_vote_only_mode: Option<Arc<AtomicBool>>,
+        exit: Arc<AtomicBool>,
+    ) -> Self {
+        let recycler: PacketBatchRecycler = Recycler::warmed(1000, 1024);
+
+        let tpu_stats = Arc::new(StreamerReceiveStats::new("vortexor_receiver"));
+
+        let thread_hdls = sockets
+            .into_iter()
+            .enumerate()
+            .map(|(i, socket)| {
+                streamer::receiver(
+                    format!("solVtxRcvr{i:02}"),
+                    socket,
+                    exit.clone(),
+                    sender.clone(),
+                    recycler.clone(),
+                    tpu_stats.clone(),
+                    Some(coalesce),
+                    true,
+                    in_vote_only_mode.clone(),
+                    false, // is_staked_service
+                )
+            })
+            .collect();
+
+        Self { thread_hdls }
+    }
+
+    pub fn join(self) -> thread::Result<()> {
+        for thread_hdl in self.thread_hdls {
+            thread_hdl.join()?;
+        }
+        Ok(())
+    }
+}

vortexor/src/sender.rs

@@ -83,7 +83,9 @@ impl PacketBatchSender {
                     for batch in &packet_batches {
                         for packet_batch in batch.iter() {
                             for packet in packet_batch {
-                                packets.push(packet.data(0..).unwrap());
+                                if let Some(data) = packet.data(0..) {
+                                    packets.push(data);
+                                }
                             }
                         }
                     }