commit 5fa7f11e2b76fc8e3ec194b633b7f455ac9b5317
parent 8d3e2db98b1091017586d748a64d5b915c4f0405
Author: Joris Hartog <jorishartog@hotmail.com>
Date: Thu, 30 Jul 2026 16:00:43 +0200
Split transaction gossip batches
Diffstat:
3 files changed, 170 insertions(+), 18 deletions(-)
diff --git a/src/adapters/p2p.rs b/src/adapters/p2p.rs
@@ -25,7 +25,7 @@ use crate::{
app::{
BlockInventory, GossipEnvelope, MEMPOOL_STATUS_LIMIT, NETWORK_ID, NodeCore,
PROTOCOL_VERSION, PeerDirection, ProtocolHello, SharedNode, SharedPeerBook,
- TransactionRejection, debug_logging_enabled, now_ms,
+ TRANSACTION_BATCH_LIMIT, TransactionRejection, debug_logging_enabled, now_ms,
},
domain::{Block, ChainSnapshot, Ledger, Transaction, TransactionSubmitOutcome, verify_vdf},
};
@@ -632,7 +632,7 @@ impl GossipNetwork {
.map(|tx| tx.signature().to_string())
.collect::<Vec<_>>(),
);
- passthrough.push(GossipEnvelope::Transactions { transactions });
+ passthrough.extend(transaction_batch_envelopes(transactions));
}
GossipEnvelope::Block(block) => blocks.push(BlockInventory {
height: block.height,
@@ -656,9 +656,7 @@ impl GossipNetwork {
}
if !full_transactions.is_empty() {
- passthrough.push(GossipEnvelope::Transactions {
- transactions: full_transactions,
- });
+ passthrough.extend(transaction_batch_envelopes(full_transactions));
}
txs.sort();
@@ -1027,10 +1025,17 @@ async fn session_loop(
if let Some(peer) = &known_peer {
let transactions = network.pending_transactions_for_retry(peer).await;
if !transactions.is_empty() {
- let retry = GossipEnvelope::Transactions { transactions };
- write_envelope(&mut writer, &retry).await?;
+ let retry_envelopes = transaction_batch_envelopes(transactions);
+ for retry in &retry_envelopes {
+ write_envelope(&mut writer, retry).await?;
+ }
P2pMetricsCounters::inc(&network.inner.metrics.transaction_retries_sent);
- network.inner.peers.lock().await.record_sent(peer, 1);
+ network
+ .inner
+ .peers
+ .lock()
+ .await
+ .record_sent(peer, retry_envelopes.len() as u64);
}
}
}
@@ -1126,8 +1131,8 @@ async fn process_envelope(
.lock()
.await
.transactions_by_signature(&signatures);
- if !transactions.is_empty() {
- write_envelope(writer, &GossipEnvelope::Transactions { transactions }).await?;
+ for envelope in transaction_batch_envelopes(transactions) {
+ write_envelope(writer, &envelope).await?;
}
}
GossipEnvelope::BlockRequest { hashes } => {
@@ -1744,7 +1749,11 @@ fn validate_envelope_limits(envelope: &GossipEnvelope) -> Result<()> {
)?;
}
GossipEnvelope::Transactions { transactions } => {
- ensure_len("transaction batch", transactions.len(), MAX_OBJECT_REQUESTS)?;
+ ensure_len(
+ "transaction batch",
+ transactions.len(),
+ TRANSACTION_BATCH_LIMIT,
+ )?;
}
GossipEnvelope::Blocks { blocks } => {
ensure_len("block batch", blocks.len(), MAX_BLOCK_BATCH)?;
@@ -1767,6 +1776,15 @@ fn validate_envelope_limits(envelope: &GossipEnvelope) -> Result<()> {
Ok(())
}
+fn transaction_batch_envelopes(transactions: Vec<Transaction>) -> Vec<GossipEnvelope> {
+ transactions
+ .chunks(TRANSACTION_BATCH_LIMIT)
+ .map(|chunk| GossipEnvelope::Transactions {
+ transactions: chunk.to_vec(),
+ })
+ .collect()
+}
+
fn ensure_len(label: &str, len: usize, max: usize) -> Result<()> {
if len > max {
anyhow::bail!("{label} has {len} items, exceeding limit {max}");
@@ -2442,7 +2460,7 @@ mod tests {
use crate::{
app::{
BlockInventory, GossipEnvelope, NETWORK_ID, NodeCore, PROTOCOL_VERSION, PeerBook,
- PeerDirection, ProtocolHello,
+ PeerDirection, ProtocolHello, TRANSACTION_BATCH_LIMIT,
},
domain::{Amount, GenesisBurn, Ledger, Wallet},
};
@@ -2907,6 +2925,101 @@ mod tests {
}
#[tokio::test]
+ async fn prepare_gossip_splits_transaction_batches_at_receiver_limit() {
+ let alice = Wallet::from_seed("mempool-repair-batch-alice");
+ let allocations = allocations(std::slice::from_ref(&alice), 1_000);
+ let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations)));
+ let network = super::GossipNetwork {
+ inner: Arc::new(super::GossipNetworkInner {
+ node: Arc::clone(&node),
+ peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())),
+ listen_addr: "127.0.0.1:9544".parse().unwrap(),
+ node_id: super::new_node_id(),
+ sessions: tokio::sync::Mutex::new(BTreeMap::new()),
+ tx_delivery: tokio::sync::Mutex::new(BTreeMap::new()),
+ inbound_limiter: Arc::new(
+ StdMutex::new(super::InboundConnectionLimiter::default()),
+ ),
+ metrics: super::P2pMetricsCounters::default(),
+ }),
+ };
+
+ let transactions = {
+ let mut node = node.lock().await;
+ (0..(TRANSACTION_BATCH_LIMIT + 1))
+ .map(|_| node.burn(1).unwrap())
+ .collect::<Vec<_>>()
+ };
+
+ let prepared = network
+ .prepare_gossip(vec![GossipEnvelope::Transactions { transactions }])
+ .await;
+
+ let batch_sizes = prepared
+ .iter()
+ .filter_map(|envelope| match envelope {
+ GossipEnvelope::Transactions { transactions } => Some(transactions.len()),
+ _ => None,
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(batch_sizes, vec![TRANSACTION_BATCH_LIMIT, 1]);
+ assert!(prepared.iter().all(|envelope| match envelope {
+ GossipEnvelope::Transactions { transactions } =>
+ transactions.len() <= TRANSACTION_BATCH_LIMIT,
+ _ => true,
+ }));
+ assert!(matches!(
+ prepared.last(),
+ Some(GossipEnvelope::Inventory { txs, blocks })
+ if txs.len() == TRANSACTION_BATCH_LIMIT + 1 && blocks.is_empty()
+ ));
+ }
+
+ #[tokio::test]
+ async fn retry_transaction_batches_are_split_at_receiver_limit() {
+ let alice = Wallet::from_seed("tx-ack-retry-batch-alice");
+ let allocations = allocations(std::slice::from_ref(&alice), 1_000);
+ let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations)));
+ let network = super::GossipNetwork {
+ inner: Arc::new(super::GossipNetworkInner {
+ node: Arc::clone(&node),
+ peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())),
+ listen_addr: "127.0.0.1:9544".parse().unwrap(),
+ node_id: super::new_node_id(),
+ sessions: tokio::sync::Mutex::new(BTreeMap::new()),
+ tx_delivery: tokio::sync::Mutex::new(BTreeMap::new()),
+ inbound_limiter: Arc::new(
+ StdMutex::new(super::InboundConnectionLimiter::default()),
+ ),
+ metrics: super::P2pMetricsCounters::default(),
+ }),
+ };
+ let peer = "127.0.0.1:9545";
+ {
+ let mut node = node.lock().await;
+ for _ in 0..(TRANSACTION_BATCH_LIMIT + 1) {
+ node.burn(1).unwrap();
+ }
+ }
+
+ let retry = network.pending_transactions_for_retry(peer).await;
+ assert_eq!(retry.len(), TRANSACTION_BATCH_LIMIT + 1);
+
+ let retry_batches = super::transaction_batch_envelopes(retry);
+ let batch_sizes = retry_batches
+ .iter()
+ .map(|envelope| match envelope {
+ GossipEnvelope::Transactions { transactions } => {
+ assert!(transactions.len() <= TRANSACTION_BATCH_LIMIT);
+ transactions.len()
+ }
+ other => panic!("expected transaction batch, got {other:?}"),
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(batch_sizes, vec![TRANSACTION_BATCH_LIMIT, 1]);
+ }
+
+ #[tokio::test]
async fn unacked_pending_transactions_retry_until_peer_accepts() {
let alice = Wallet::from_seed("tx-ack-retry-alice");
let allocations = allocations(std::slice::from_ref(&alice), 1_000);
diff --git a/src/app.rs b/src/app.rs
@@ -27,6 +27,7 @@ pub const DEFAULT_VDF_ROUNDS: u32 = 67_000_000;
pub const PROTOCOL_VERSION: u32 = 1;
pub const NETWORK_ID: &str = "iuna-devnet-v2";
pub const BLOCK_REQUEST_LIMIT: usize = 128;
+pub const TRANSACTION_BATCH_LIMIT: usize = 128;
pub const MEMPOOL_STATUS_LIMIT: usize = MAX_PENDING_TRANSACTIONS;
const IMPORT_REBROADCAST_LIMIT: usize = 128;
pub const PEER_MISBEHAVIOR_BAN_SCORE: u32 = 3;
@@ -418,11 +419,12 @@ impl NodeCore {
pub fn mempool_gossip(&self) -> Vec<GossipEnvelope> {
let transactions = self.ledger.pending().to_vec();
- if transactions.is_empty() {
- Vec::new()
- } else {
- vec![GossipEnvelope::Transactions { transactions }]
- }
+ transactions
+ .chunks(TRANSACTION_BATCH_LIMIT)
+ .map(|chunk| GossipEnvelope::Transactions {
+ transactions: chunk.to_vec(),
+ })
+ .collect()
}
pub fn chain_snapshot(&self) -> ChainSnapshot {
diff --git a/tests/iuna.rs b/tests/iuna.rs
@@ -7,7 +7,7 @@ use iuna::{
adapters::chain_store::SqliteChainStore,
app::{
DEFAULT_BURN_PER_BLOCK, GossipEnvelope, InMemoryNetwork, NodeConfig, NodeCore, PeerBook,
- PeerDirection,
+ PeerDirection, TRANSACTION_BATCH_LIMIT,
},
domain::{
Amount, BLOCK_REWARD, DEFAULT_FEE_PER_BYTE, GenesisBurn, Ledger, MAX_BLOCK_BYTES,
@@ -1720,6 +1720,43 @@ fn mempool_gossip_repairs_future_nonce_gap_without_networking() {
}
#[test]
+fn mempool_gossip_splits_transaction_batches_at_receiver_limit() {
+ let alice = Wallet::from_seed("mempool-batch-alice");
+ let bob = Wallet::from_seed("mempool-batch-bob");
+ let wallets = vec![alice.clone(), bob.clone()];
+ let allocations = allocations(&wallets, 10_000);
+ let mut alice_node = node("alice", alice, allocations.clone());
+ let mut bob_node = node("bob", bob, allocations);
+
+ for _ in 0..(TRANSACTION_BATCH_LIMIT + 1) {
+ alice_node.burn(1).unwrap();
+ }
+ alice_node.drain_outbox();
+
+ let gossip = alice_node.mempool_gossip();
+ assert_eq!(gossip.len(), 2);
+ let total_transactions = gossip
+ .iter()
+ .map(|envelope| match envelope {
+ GossipEnvelope::Transactions { transactions } => {
+ assert!(transactions.len() <= TRANSACTION_BATCH_LIMIT);
+ transactions.len()
+ }
+ other => panic!("expected transaction batch, got {other:?}"),
+ })
+ .sum::<usize>();
+ assert_eq!(total_transactions, TRANSACTION_BATCH_LIMIT + 1);
+
+ for envelope in gossip {
+ bob_node.receive(envelope).unwrap();
+ }
+ assert_eq!(
+ bob_node.ledger().pending().len(),
+ TRANSACTION_BATCH_LIMIT + 1
+ );
+}
+
+#[test]
fn peer_status_advertises_mempool_and_drives_missing_transaction_request() {
let alice = Wallet::from_seed("mempool-status-alice");
let bob = Wallet::from_seed("mempool-status-bob");
