commit 1d5c0bb2d85228eba9efcc69f9ce190bb4dcaf3f
parent c1cdd4d481b432cae99412f048c339872ecd5d15
Author: Joris Hartog <jorishartog@hotmail.com>
Date: Tue, 28 Jul 2026 20:07:08 +0200
Sync mempools through peer status
Diffstat:
| M | src/adapters/http.rs | | | 76 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--- |
| M | src/adapters/p2p.rs | | | 285 | ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- |
| M | src/app.rs | | | 144 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-- |
| M | tests/iuna.rs | | | 49 | ++++++++++++++++++++++++++++++++++++++++++++++++- |
4 files changed, 525 insertions(+), 29 deletions(-)
diff --git a/src/adapters/http.rs b/src/adapters/http.rs
@@ -95,6 +95,9 @@ struct NetworkHealthResponse {
stale_peers: usize,
banned_peers: usize,
pending_transactions: usize,
+ mempool_known_peers: usize,
+ mempool_divergent_peers: usize,
+ mempool_missing_transactions: usize,
last_error: Option<String>,
}
@@ -814,6 +817,18 @@ fn network_health_at(
.iter()
.filter(|peer| peer.is_banned_at(now_ms))
.count();
+ let mempool_known_peers = peers
+ .iter()
+ .filter(|peer| peer.last_known_mempool_count.is_some())
+ .count();
+ let mempool_divergent_peers = peers
+ .iter()
+ .filter(|peer| peer.last_known_mempool_missing.unwrap_or(0) > 0)
+ .count();
+ let mempool_missing_transactions = peers
+ .iter()
+ .map(|peer| peer.last_known_mempool_missing.unwrap_or(0))
+ .sum();
let lag_blocks = best_known_height.saturating_sub(local_height);
let last_error = peers.iter().rev().find_map(|peer| {
peer.last_error
@@ -828,6 +843,8 @@ fn network_health_at(
"banned"
} else if lag_blocks > 0 {
"syncing"
+ } else if mempool_missing_transactions > 0 {
+ "mempool syncing"
} else if failed_peers > 0 && healthy_peers == 0 {
"peer errors"
} else if stale_peers > 0 && healthy_peers == stale_peers {
@@ -840,7 +857,10 @@ fn network_health_at(
.to_string();
NetworkHealthResponse {
- ok: !peers.is_empty() && lag_blocks == 0 && healthy_peers > stale_peers,
+ ok: !peers.is_empty()
+ && lag_blocks == 0
+ && mempool_missing_transactions == 0
+ && healthy_peers > stale_peers,
state,
local_height,
best_known_height,
@@ -853,6 +873,9 @@ fn network_health_at(
stale_peers,
banned_peers,
pending_transactions: status.chain.pending_transactions,
+ mempool_known_peers,
+ mempool_divergent_peers,
+ mempool_missing_transactions,
last_error,
}
}
@@ -2148,6 +2171,9 @@ const INDEX_HTML: &str = r#"<!doctype html>
<div class="peer-summary-item"><div class="peer-summary-label">Stale</div><div class="peer-summary-value" x-text="networkHealth.stale_peers ?? '-'"></div></div>
<div class="peer-summary-item"><div class="peer-summary-label">Banned</div><div class="peer-summary-value" x-text="networkHealth.banned_peers ?? '-'"></div></div>
<div class="peer-summary-item"><div class="peer-summary-label">Mempool</div><div class="peer-summary-value" x-text="networkHealth.pending_transactions ?? '-'"></div></div>
+ <div class="peer-summary-item"><div class="peer-summary-label">Peer Mempools</div><div class="peer-summary-value" x-text="networkHealth.mempool_known_peers ?? '-'"></div></div>
+ <div class="peer-summary-item"><div class="peer-summary-label">Divergent</div><div class="peer-summary-value" x-text="networkHealth.mempool_divergent_peers ?? '-'"></div></div>
+ <div class="peer-summary-item"><div class="peer-summary-label">Missing Tx</div><div class="peer-summary-value" x-text="networkHealth.mempool_missing_transactions ?? '-'"></div></div>
</div>
</div>
<div class="peer-summary">
@@ -2159,7 +2185,7 @@ const INDEX_HTML: &str = r#"<!doctype html>
</div>
<div class="table-wrap">
<table>
- <thead><tr><th>Status</th><th>Address</th><th>Direction</th><th>Last Contact</th><th>Ban</th><th>Score</th><th>Height</th><th>Delta</th><th>Tip</th><th>Sent</th><th>Received</th><th>Last Error</th><th>Actions</th></tr></thead>
+ <thead><tr><th>Status</th><th>Address</th><th>Direction</th><th>Last Contact</th><th>Ban</th><th>Score</th><th>Height</th><th>Delta</th><th>Tip</th><th>Mempool</th><th>Shared</th><th>Missing</th><th>Root</th><th>Sent</th><th>Received</th><th>Last Error</th><th>Actions</th></tr></thead>
<tbody>
<template x-for="peer in peers" :key="peer.address">
<tr>
@@ -2172,13 +2198,17 @@ const INDEX_HTML: &str = r#"<!doctype html>
<td x-text="peer.last_known_height ?? '-'"></td>
<td x-text="peerHeightDelta(peer)"></td>
<td><code x-text="short(peer.last_known_tip_hash)"></code></td>
+ <td x-text="peer.last_known_mempool_count ?? '-'"></td>
+ <td x-text="peer.last_known_mempool_shared ?? '-'"></td>
+ <td x-text="peer.last_known_mempool_missing ?? '-'"></td>
+ <td><code x-text="short(peer.last_known_mempool_root)"></code></td>
<td x-text="peer.messages_sent"></td>
<td x-text="peer.messages_received"></td>
<td x-text="peer.last_error || peer.last_transaction_rejection || ''"></td>
<td><div class="peer-actions"><button class="peer-remove" type="button" x-show="canRemovePeer(peer)" @click="removePeer(peer)">Remove</button><span class="muted" x-show="!canRemovePeer(peer)">Observed</span></div></td>
</tr>
</template>
- <tr x-show="peers.length === 0"><td colspan="13">No peers</td></tr>
+ <tr x-show="peers.length === 0"><td colspan="17">No peers</td></tr>
</tbody>
</table>
</div>
@@ -2209,6 +2239,11 @@ const INDEX_HTML: &str = r#"<!doctype html>
<div class="metric"><div class="label">Tx Rejected Rx</div><div class="value" x-text="p2pMetrics.transactions_rejected_received ?? 0"></div></div>
<div class="metric"><div class="label">Tx Retries</div><div class="value" x-text="p2pMetrics.transaction_retries_sent ?? 0"></div></div>
<div class="metric"><div class="label">Tx Ack Pending</div><div class="value" x-text="p2pMetrics.transaction_ack_pending ?? 0"></div></div>
+ <div class="metric"><div class="label">Mempool Status Rx</div><div class="value" x-text="p2pMetrics.mempool_statuses_received ?? 0"></div></div>
+ <div class="metric"><div class="label">Mempool Tx Seen</div><div class="value" x-text="p2pMetrics.mempool_status_transactions_received ?? 0"></div></div>
+ <div class="metric"><div class="label">Mempool Mismatch</div><div class="value" x-text="p2pMetrics.mempool_status_mismatches ?? 0"></div></div>
+ <div class="metric"><div class="label">Mempool Requests</div><div class="value" x-text="p2pMetrics.mempool_transaction_requests_sent ?? 0"></div></div>
+ <div class="metric"><div class="label">Mempool Requested Tx</div><div class="value" x-text="p2pMetrics.mempool_transaction_request_signatures_sent ?? 0"></div></div>
</div>
<div class="metric-context">
<div class="tx-field"><span class="tx-label">Last Failure</span><span class="tx-value text" x-text="p2pMetrics.last_session_failure || '-'"></span></div>
@@ -2770,6 +2805,16 @@ mod tests {
assert_eq!(isolated.local_height, 0);
assert_eq!(isolated.best_known_height, 0);
+ let mut mempool_peers = PeerBook::from_addresses(vec!["127.0.0.1:9444".to_string()]);
+ mempool_peers.record_status("127.0.0.1:9444", 0, "tip".to_string());
+ mempool_peers.record_mempool_status("127.0.0.1:9444", 2, "remote-root".to_string(), 1, 1);
+ let mempool_syncing = super::network_health(&status, &mempool_peers.list());
+ assert!(!mempool_syncing.ok);
+ assert_eq!(mempool_syncing.state, "mempool syncing");
+ assert_eq!(mempool_syncing.mempool_known_peers, 1);
+ assert_eq!(mempool_syncing.mempool_divergent_peers, 1);
+ assert_eq!(mempool_syncing.mempool_missing_transactions, 1);
+
let syncing = super::network_health(
&status,
&[PeerInfo {
@@ -2779,6 +2824,11 @@ mod tests {
messages_received: 1,
last_known_height: Some(3),
last_known_tip_hash: Some("remote-tip".to_string()),
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: None,
last_transaction_rejection: None,
last_contact_ms: Some(10_000),
@@ -2804,6 +2854,11 @@ mod tests {
messages_received: 0,
last_known_height: None,
last_known_tip_hash: None,
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: Some("connection refused".to_string()),
last_transaction_rejection: None,
last_contact_ms: Some(10_000),
@@ -2831,6 +2886,11 @@ mod tests {
messages_received: 1,
last_known_height: Some(0),
last_known_tip_hash: Some("tip".to_string()),
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: None,
last_transaction_rejection: Some(
"peer rejected transaction abc: conflict".to_string(),
@@ -2858,6 +2918,11 @@ mod tests {
messages_received: 1,
last_known_height: Some(0),
last_known_tip_hash: Some("tip".to_string()),
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: None,
last_transaction_rejection: None,
last_contact_ms: Some(1),
@@ -2883,6 +2948,11 @@ mod tests {
messages_received: 0,
last_known_height: None,
last_known_tip_hash: None,
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: Some("invalid block".to_string()),
last_transaction_rejection: None,
last_contact_ms: Some(10),
diff --git a/src/adapters/p2p.rs b/src/adapters/p2p.rs
@@ -23,8 +23,8 @@ use tokio::{
use crate::{
app::{
- BlockInventory, GossipEnvelope, NETWORK_ID, NodeCore, PROTOCOL_VERSION, ProtocolHello,
- SharedNode, SharedPeerBook, TransactionRejection,
+ BlockInventory, GossipEnvelope, MEMPOOL_STATUS_LIMIT, NETWORK_ID, NodeCore,
+ PROTOCOL_VERSION, ProtocolHello, SharedNode, SharedPeerBook, TransactionRejection,
},
domain::{Block, ChainSnapshot, Ledger, Transaction, TransactionSubmitOutcome, verify_vdf},
};
@@ -50,6 +50,9 @@ static NEXT_NODE_ID: AtomicU64 = AtomicU64::new(1);
struct PeerStatus {
height: u64,
tip_hash: String,
+ mempool_count: usize,
+ mempool_root: String,
+ mempool_txs: Vec<String>,
request_snapshot: bool,
push_snapshot: bool,
}
@@ -59,6 +62,27 @@ impl PeerStatus {
Self {
height,
tip_hash,
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
+ request_snapshot: false,
+ push_snapshot: false,
+ }
+ }
+
+ fn from_envelope(
+ height: u64,
+ tip_hash: String,
+ mempool_count: usize,
+ mempool_root: String,
+ mempool_txs: Vec<String>,
+ ) -> Self {
+ Self {
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
request_snapshot: false,
push_snapshot: false,
}
@@ -68,6 +92,9 @@ impl PeerStatus {
Self {
height,
tip_hash,
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
request_snapshot: true,
push_snapshot: false,
}
@@ -77,6 +104,9 @@ impl PeerStatus {
Self {
height,
tip_hash,
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
request_snapshot: false,
push_snapshot: true,
}
@@ -137,6 +167,11 @@ struct P2pMetricsCounters {
transactions_rejected_sent: AtomicU64,
transactions_rejected_received: AtomicU64,
transaction_retries_sent: AtomicU64,
+ mempool_statuses_received: AtomicU64,
+ mempool_status_transactions_received: AtomicU64,
+ mempool_status_mismatches: AtomicU64,
+ mempool_transaction_requests_sent: AtomicU64,
+ mempool_transaction_request_signatures_sent: AtomicU64,
last_session_failure: StdMutex<Option<String>>,
last_empty_frame_remote: StdMutex<Option<String>>,
last_parse_error: StdMutex<Option<String>>,
@@ -172,6 +207,11 @@ pub struct P2pMetrics {
pub transactions_rejected_sent: u64,
pub transactions_rejected_received: u64,
pub transaction_retries_sent: u64,
+ pub mempool_statuses_received: u64,
+ pub mempool_status_transactions_received: u64,
+ pub mempool_status_mismatches: u64,
+ pub mempool_transaction_requests_sent: u64,
+ pub mempool_transaction_request_signatures_sent: u64,
pub transaction_ack_pending: u64,
pub last_session_failure: Option<String>,
pub last_empty_frame_remote: Option<String>,
@@ -233,6 +273,17 @@ impl P2pMetricsCounters {
.transactions_rejected_received
.load(Ordering::Relaxed),
transaction_retries_sent: self.transaction_retries_sent.load(Ordering::Relaxed),
+ mempool_statuses_received: self.mempool_statuses_received.load(Ordering::Relaxed),
+ mempool_status_transactions_received: self
+ .mempool_status_transactions_received
+ .load(Ordering::Relaxed),
+ mempool_status_mismatches: self.mempool_status_mismatches.load(Ordering::Relaxed),
+ mempool_transaction_requests_sent: self
+ .mempool_transaction_requests_sent
+ .load(Ordering::Relaxed),
+ mempool_transaction_request_signatures_sent: self
+ .mempool_transaction_request_signatures_sent
+ .load(Ordering::Relaxed),
transaction_ack_pending: 0,
last_session_failure: self
.last_session_failure
@@ -745,9 +796,31 @@ async fn session_loop(
return Ok(());
}
maybe_request_catchup(&network, &mut writer, peer_status.as_ref().unwrap()).await?;
- } else if let GossipEnvelope::PeerStatus { height, tip_hash } = envelope {
- peer_status = Some(PeerStatus::new(height, tip_hash.clone()));
- record_peer_status(&network, &known_peer, remote_addr, height, tip_hash).await;
+ } else if let GossipEnvelope::PeerStatus {
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ } = envelope
+ {
+ let status = PeerStatus::from_envelope(
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ );
+ record_peer_status(&network, &known_peer, remote_addr, &status).await;
+ maybe_request_mempool_catchup(
+ &network,
+ &mut writer,
+ &known_peer,
+ remote_addr,
+ &status,
+ )
+ .await?;
+ peer_status = Some(status);
maybe_request_catchup(&network, &mut writer, peer_status.as_ref().unwrap()).await?;
} else {
process_envelope(
@@ -814,9 +887,25 @@ async fn session_loop(
maybe_request_catchup(&network, &mut writer, peer_status.as_ref().unwrap()).await?;
continue;
}
- if let GossipEnvelope::PeerStatus { height, tip_hash } = &envelope {
- peer_status = Some(PeerStatus::new(*height, tip_hash.clone()));
- record_peer_status(&network, &known_peer, remote_addr, *height, tip_hash.clone()).await;
+ if let GossipEnvelope::PeerStatus {
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ } = &envelope
+ {
+ let status = PeerStatus::from_envelope(
+ *height,
+ tip_hash.clone(),
+ *mempool_count,
+ mempool_root.clone(),
+ mempool_txs.clone(),
+ );
+ record_peer_status(&network, &known_peer, remote_addr, &status).await;
+ maybe_request_mempool_catchup(&network, &mut writer, &known_peer, remote_addr, &status)
+ .await?;
+ peer_status = Some(status);
maybe_request_catchup(&network, &mut writer, peer_status.as_ref().unwrap()).await?;
continue;
}
@@ -1109,6 +1198,77 @@ async fn maybe_request_catchup(
Ok(())
}
+async fn maybe_request_mempool_catchup(
+ network: &GossipNetwork,
+ writer: &mut OwnedWriteHalf,
+ known_peer: &Option<String>,
+ remote_addr: SocketAddr,
+ peer_status: &PeerStatus,
+) -> Result<()> {
+ P2pMetricsCounters::inc(&network.inner.metrics.mempool_statuses_received);
+ P2pMetricsCounters::add(
+ &network.inner.metrics.mempool_status_transactions_received,
+ peer_status.mempool_txs.len() as u64,
+ );
+
+ let (local_root, local_inventory, requests) = {
+ let node = network.inner.node.lock().await;
+ (
+ node.mempool_root(),
+ node.mempool_inventory(MEMPOOL_STATUS_LIMIT),
+ node.missing_inventory_requests(&peer_status.mempool_txs, &[]),
+ )
+ };
+ let local_txs = local_inventory.into_iter().collect::<BTreeSet<_>>();
+ let shared = peer_status
+ .mempool_txs
+ .iter()
+ .filter(|signature| local_txs.contains(*signature))
+ .count();
+ let missing = peer_status.mempool_txs.len().saturating_sub(shared);
+
+ if peer_status.mempool_root != local_root {
+ P2pMetricsCounters::inc(&network.inner.metrics.mempool_status_mismatches);
+ }
+ record_peer_mempool_status(
+ network,
+ known_peer,
+ remote_addr,
+ peer_status,
+ shared,
+ missing,
+ )
+ .await;
+
+ let requested_signatures = requests
+ .iter()
+ .map(|envelope| match envelope {
+ GossipEnvelope::TransactionRequest { signatures } => signatures.len(),
+ _ => 0,
+ })
+ .sum::<usize>();
+ if requested_signatures > 0 {
+ write_payload(writer, &requests).await?;
+ P2pMetricsCounters::inc(&network.inner.metrics.mempool_transaction_requests_sent);
+ P2pMetricsCounters::add(
+ &network
+ .inner
+ .metrics
+ .mempool_transaction_request_signatures_sent,
+ requested_signatures as u64,
+ );
+ if let Some(peer) = known_peer {
+ network
+ .inner
+ .peers
+ .lock()
+ .await
+ .record_sent(peer, requests.len() as u64);
+ }
+ }
+ Ok(())
+}
+
async fn push_catchup_to_peer(
network: &GossipNetwork,
writer: &mut OwnedWriteHalf,
@@ -1429,8 +1589,10 @@ fn validate_envelope_limits(envelope: &GossipEnvelope) -> Result<()> {
GossipEnvelope::PeerList { peers } => {
ensure_len("peer list", peers.len(), MAX_PEER_LIST)?;
}
+ GossipEnvelope::PeerStatus { mempool_txs, .. } => {
+ ensure_len("mempool status", mempool_txs.len(), MAX_INVENTORY_ITEMS)?;
+ }
GossipEnvelope::Hello(_)
- | GossipEnvelope::PeerStatus { .. }
| GossipEnvelope::ChainSnapshotRequest
| GossipEnvelope::Transaction(_)
| GossipEnvelope::Block(_)
@@ -1551,7 +1713,19 @@ async fn fetch_peer_status(peer: &str) -> Result<PeerStatus> {
}
Ok(PeerStatus::new(hello.height, hello.tip_hash))
}
- GossipEnvelope::PeerStatus { height, tip_hash } => Ok(PeerStatus::new(height, tip_hash)),
+ GossipEnvelope::PeerStatus {
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ } => Ok(PeerStatus::from_envelope(
+ height,
+ tip_hash,
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ )),
other => anyhow::bail!("peer {peer} sent {other:?} instead of peer status"),
}
}
@@ -1714,16 +1888,14 @@ async fn record_peer_status(
network: &GossipNetwork,
known_peer: &Option<String>,
remote_addr: SocketAddr,
- height: u64,
- tip_hash: String,
+ peer_status: &PeerStatus,
) {
if let Some(peer) = known_peer {
- network
- .inner
- .peers
- .lock()
- .await
- .record_status(peer, height, tip_hash);
+ network.inner.peers.lock().await.record_status(
+ peer,
+ peer_status.height,
+ peer_status.tip_hash.clone(),
+ );
} else {
network
.inner
@@ -1734,6 +1906,34 @@ async fn record_peer_status(
}
}
+async fn record_peer_mempool_status(
+ network: &GossipNetwork,
+ known_peer: &Option<String>,
+ remote_addr: SocketAddr,
+ peer_status: &PeerStatus,
+ shared: usize,
+ missing: usize,
+) {
+ let mut peers = network.inner.peers.lock().await;
+ if let Some(peer) = known_peer {
+ peers.record_mempool_status(
+ peer,
+ peer_status.mempool_count,
+ peer_status.mempool_root.clone(),
+ shared,
+ missing,
+ );
+ } else {
+ peers.record_inbound_mempool_status(
+ &remote_addr.to_string(),
+ peer_status.mempool_count,
+ peer_status.mempool_root.clone(),
+ shared,
+ missing,
+ );
+ }
+}
+
async fn process_hello(
network: &GossipNetwork,
remote_addr: SocketAddr,
@@ -1795,8 +1995,7 @@ async fn process_hello(
network,
known_peer,
remote_addr,
- hello.height,
- hello.tip_hash.clone(),
+ &PeerStatus::new(hello.height, hello.tip_hash.clone()),
)
.await;
if request_snapshot {
@@ -2132,6 +2331,38 @@ mod tests {
}
#[test]
+ fn parser_accepts_legacy_peer_status_without_mempool_fields() {
+ let envelope =
+ parse_envelope(r#"{"type":"peer_status","height":7,"tip_hash":"tip"}"#).unwrap();
+
+ assert_eq!(
+ envelope,
+ GossipEnvelope::PeerStatus {
+ height: 7,
+ tip_hash: "tip".to_string(),
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
+ }
+ );
+ }
+
+ #[test]
+ fn oversized_mempool_status_is_rejected_before_processing() {
+ let envelope = GossipEnvelope::PeerStatus {
+ height: 7,
+ tip_hash: "tip".to_string(),
+ mempool_count: MAX_INVENTORY_ITEMS + 1,
+ mempool_root: "root".to_string(),
+ mempool_txs: vec!["sig".to_string(); MAX_INVENTORY_ITEMS + 1],
+ };
+
+ let error = validate_envelope_limits(&envelope).unwrap_err();
+
+ assert!(error.to_string().contains("mempool status"));
+ }
+
+ #[test]
fn received_envelope_metrics_are_categorized() {
let metrics = super::P2pMetricsCounters::default();
@@ -2140,6 +2371,9 @@ mod tests {
&GossipEnvelope::PeerStatus {
height: 7,
tip_hash: "tip".to_string(),
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
},
);
super::record_received_envelope_kind(
@@ -2176,6 +2410,9 @@ mod tests {
let line = serde_json::to_string(&GossipEnvelope::PeerStatus {
height: 7,
tip_hash: "tip".to_string(),
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
})
.unwrap();
let split_at = line.len() / 2;
@@ -2883,8 +3120,7 @@ mod tests {
&network,
&None,
"127.0.0.1:51729".parse().unwrap(),
- 4,
- "tip".to_string(),
+ &super::PeerStatus::new(4, "tip".to_string()),
)
.await;
@@ -2902,7 +3138,10 @@ mod tests {
"127.0.0.1:9544",
GossipEnvelope::PeerStatus {
height: 0,
- tip_hash: "tip".to_string()
+ tip_hash: "tip".to_string(),
+ mempool_count: 0,
+ mempool_root: String::new(),
+ mempool_txs: Vec::new(),
}
)
.unwrap()
diff --git a/src/app.rs b/src/app.rs
@@ -1,5 +1,5 @@
use std::{
- collections::BTreeMap,
+ collections::{BTreeMap, BTreeSet},
sync::Arc,
time::{SystemTime, UNIX_EPOCH},
};
@@ -12,7 +12,7 @@ use tokio::sync::Mutex;
use crate::domain::{
Amount, Block, ChainSnapshot, ChainStatus, DEFAULT_FEE_PER_BYTE, DEFAULT_TRANSACTION_FEE,
Ledger, MINE_FINALIZER_FEE, OutPoint, PreparedBlock, StratumMineShare, StratumMineTemplate,
- Transaction, TransactionSubmitOutcome, VDF_TARGET_BLOCK_MS, Wallet, run_vdf,
+ Transaction, TransactionSubmitOutcome, VDF_TARGET_BLOCK_MS, Wallet, hex_hash, run_vdf,
};
pub type SharedNode = Arc<Mutex<NodeCore>>;
@@ -23,6 +23,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 MEMPOOL_STATUS_LIMIT: usize = 512;
const IMPORT_REBROADCAST_LIMIT: usize = 128;
pub const PEER_MISBEHAVIOR_BAN_SCORE: u32 = 3;
pub const PEER_MISBEHAVIOR_BAN_MS: u64 = 10 * 60 * 1_000;
@@ -81,6 +82,12 @@ pub enum GossipEnvelope {
PeerStatus {
height: u64,
tip_hash: String,
+ #[serde(default)]
+ mempool_count: usize,
+ #[serde(default)]
+ mempool_root: String,
+ #[serde(default)]
+ mempool_txs: Vec<String>,
},
ChainSnapshotRequest,
BlockRangeRequest {
@@ -360,6 +367,36 @@ impl NodeCore {
self.ledger.pending().to_vec()
}
+ pub fn mempool_inventory(&self, limit: usize) -> Vec<String> {
+ let mut signatures = self
+ .ledger
+ .pending()
+ .iter()
+ .map(|tx| tx.signature().to_string())
+ .collect::<BTreeSet<_>>()
+ .into_iter()
+ .collect::<Vec<_>>();
+ signatures.truncate(limit);
+ signatures
+ }
+
+ pub fn mempool_count(&self) -> usize {
+ self.ledger.pending().len()
+ }
+
+ pub fn mempool_root(&self) -> String {
+ mempool_root_for_signatures(
+ &self
+ .ledger
+ .pending()
+ .iter()
+ .map(|tx| tx.signature().to_string())
+ .collect::<BTreeSet<_>>()
+ .into_iter()
+ .collect::<Vec<_>>(),
+ )
+ }
+
pub fn mempool_gossip(&self) -> Vec<GossipEnvelope> {
let transactions = self.ledger.pending().to_vec();
if transactions.is_empty() {
@@ -388,9 +425,13 @@ impl NodeCore {
pub fn peer_status(&self) -> GossipEnvelope {
let status = self.ledger.status();
+ let mempool_txs = self.mempool_inventory(MEMPOOL_STATUS_LIMIT);
GossipEnvelope::PeerStatus {
height: status.height,
tip_hash: status.tip_hash,
+ mempool_count: self.mempool_count(),
+ mempool_root: self.mempool_root(),
+ mempool_txs,
}
}
@@ -1198,6 +1239,22 @@ impl PeerBook {
.last_known_tip_hash
.clone()
.or(from_peer.last_known_tip_hash);
+ to_peer.last_known_mempool_count = to_peer
+ .last_known_mempool_count
+ .or(from_peer.last_known_mempool_count);
+ to_peer.last_known_mempool_root = to_peer
+ .last_known_mempool_root
+ .clone()
+ .or(from_peer.last_known_mempool_root);
+ to_peer.last_known_mempool_shared = to_peer
+ .last_known_mempool_shared
+ .or(from_peer.last_known_mempool_shared);
+ to_peer.last_known_mempool_missing = to_peer
+ .last_known_mempool_missing
+ .or(from_peer.last_known_mempool_missing);
+ to_peer.last_mempool_status_ms = to_peer
+ .last_mempool_status_ms
+ .max(from_peer.last_mempool_status_ms);
to_peer.last_contact_ms = to_peer.last_contact_ms.max(from_peer.last_contact_ms);
to_peer.last_success_ms = to_peer.last_success_ms.max(from_peer.last_success_ms);
to_peer.last_error_ms = to_peer.last_error_ms.max(from_peer.last_error_ms);
@@ -1291,6 +1348,66 @@ impl PeerBook {
}
}
+ pub fn record_mempool_status(
+ &mut self,
+ address: &str,
+ mempool_count: usize,
+ mempool_root: String,
+ mempool_shared: usize,
+ mempool_missing: usize,
+ ) {
+ self.record_mempool_status_with_direction(
+ address,
+ PeerDirection::Outbound,
+ mempool_count,
+ mempool_root,
+ mempool_shared,
+ mempool_missing,
+ );
+ }
+
+ pub fn record_inbound_mempool_status(
+ &mut self,
+ address: &str,
+ mempool_count: usize,
+ mempool_root: String,
+ mempool_shared: usize,
+ mempool_missing: usize,
+ ) {
+ self.record_mempool_status_with_direction(
+ address,
+ PeerDirection::Inbound,
+ mempool_count,
+ mempool_root,
+ mempool_shared,
+ mempool_missing,
+ );
+ }
+
+ fn record_mempool_status_with_direction(
+ &mut self,
+ address: &str,
+ direction: PeerDirection,
+ mempool_count: usize,
+ mempool_root: String,
+ mempool_shared: usize,
+ mempool_missing: usize,
+ ) {
+ let now = now_ms();
+ let peer = self.ensure(address, direction);
+ peer.last_known_mempool_count = Some(mempool_count);
+ peer.last_known_mempool_root = Some(mempool_root);
+ peer.last_known_mempool_shared = Some(mempool_shared);
+ peer.last_known_mempool_missing = Some(mempool_missing);
+ peer.last_mempool_status_ms = Some(now);
+ peer.last_contact_ms = Some(now);
+ peer.last_success_ms = Some(now);
+ if !peer.is_banned_at(now) {
+ peer.last_error = None;
+ peer.clear_misbehavior();
+ }
+ }
+
pub fn record_error(&mut self, address: &str, error: impl Into<String>) {
let now = now_ms();
let peer = self.ensure(address, PeerDirection::Outbound);
@@ -1395,6 +1512,16 @@ pub struct PeerInfo {
pub messages_received: u64,
pub last_known_height: Option<u64>,
pub last_known_tip_hash: Option<String>,
+ #[serde(default)]
+ pub last_known_mempool_count: Option<usize>,
+ #[serde(default)]
+ pub last_known_mempool_root: Option<String>,
+ #[serde(default)]
+ pub last_known_mempool_shared: Option<usize>,
+ #[serde(default)]
+ pub last_known_mempool_missing: Option<usize>,
+ #[serde(default)]
+ pub last_mempool_status_ms: Option<u64>,
pub last_error: Option<String>,
pub last_transaction_rejection: Option<String>,
pub last_contact_ms: Option<u64>,
@@ -1415,6 +1542,11 @@ impl PeerInfo {
messages_received: 0,
last_known_height: None,
last_known_tip_hash: None,
+ last_known_mempool_count: None,
+ last_known_mempool_root: None,
+ last_known_mempool_shared: None,
+ last_known_mempool_missing: None,
+ last_mempool_status_ms: None,
last_error: None,
last_transaction_rejection: None,
last_contact_ms: None,
@@ -1453,6 +1585,14 @@ pub fn now_ms() -> u64 {
.as_millis() as u64
}
+pub fn mempool_root_for_signatures(signatures: &[String]) -> String {
+ if signatures.is_empty() {
+ String::new()
+ } else {
+ hex_hash(format!("iuna-mempool-root:{}", signatures.join("|")))
+ }
+}
+
fn auto_pow_salt(wallet_address: &str, anchor: &str) -> u64 {
let digest = Sha256::digest(format!("iuna-auto-pow:{wallet_address}:{anchor}").as_bytes());
let mut bytes = [0_u8; 8];
diff --git a/tests/iuna.rs b/tests/iuna.rs
@@ -2,7 +2,10 @@ use std::collections::BTreeMap;
use iuna::{
adapters::chain_store::SqliteChainStore,
- app::{DEFAULT_BURN_PER_BLOCK, InMemoryNetwork, NodeConfig, NodeCore, PeerBook, PeerDirection},
+ app::{
+ DEFAULT_BURN_PER_BLOCK, GossipEnvelope, InMemoryNetwork, NodeConfig, NodeCore, PeerBook,
+ PeerDirection,
+ },
domain::{
Amount, BLOCK_REWARD, DEFAULT_FEE_PER_BYTE, GenesisBurn, Ledger, MAX_BLOCK_BYTES,
MICRO_IUNA, TransactionSubmitOutcome, VDF_TARGET_BLOCK_MS, Wallet, run_vdf, verify_vdf,
@@ -1663,6 +1666,41 @@ fn mempool_gossip_repairs_future_nonce_gap_without_networking() {
}
#[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");
+ let wallets = vec![alice.clone(), bob.clone()];
+ let allocations = allocations(&wallets, 1_000);
+ let mut alice_node = node("alice", alice, allocations.clone());
+ let bob_node = node("bob", bob, allocations);
+
+ let tx = alice_node.burn(1).unwrap();
+ let signature = tx.signature().to_string();
+
+ let GossipEnvelope::PeerStatus {
+ mempool_count,
+ mempool_root,
+ mempool_txs,
+ ..
+ } = alice_node.peer_status()
+ else {
+ panic!("expected peer status");
+ };
+
+ assert_eq!(mempool_count, 1);
+ assert!(!mempool_root.is_empty());
+ assert_eq!(mempool_txs, vec![signature.clone()]);
+
+ let requests = bob_node.missing_inventory_requests(&mempool_txs, &[]);
+ assert_eq!(
+ requests,
+ vec![GossipEnvelope::TransactionRequest {
+ signatures: vec![signature]
+ }]
+ );
+}
+
+#[test]
fn received_block_is_rebroadcast_to_other_peers_without_networking() {
let names = ["alice", "bob", "carol"];
let wallets = wallets(&names);
@@ -1776,6 +1814,7 @@ fn peer_book_tracks_multiple_peers_without_networking() {
peers.record_sent("127.0.0.1:9444", 2);
peers.record_status("127.0.0.1:9444", 12, "tip-hash".to_string());
+ peers.record_mempool_status("127.0.0.1:9444", 3, "mempool-root".to_string(), 2, 1);
peers.record_error("127.0.0.1:9445", "connection refused");
peers.record_received("127.0.0.1:9555", 1);
peers.record_inbound_error("127.0.0.1:56666", "invalid nonce");
@@ -1797,6 +1836,14 @@ fn peer_book_tracks_multiple_peers_without_networking() {
assert_eq!(sent_peer.messages_sent, 2);
assert_eq!(sent_peer.last_known_height, Some(12));
assert_eq!(sent_peer.last_known_tip_hash.as_deref(), Some("tip-hash"));
+ assert_eq!(sent_peer.last_known_mempool_count, Some(3));
+ assert_eq!(
+ sent_peer.last_known_mempool_root.as_deref(),
+ Some("mempool-root")
+ );
+ assert_eq!(sent_peer.last_known_mempool_shared, Some(2));
+ assert_eq!(sent_peer.last_known_mempool_missing, Some(1));
+ assert!(sent_peer.last_mempool_status_ms.is_some());
assert_eq!(sent_peer.last_error, None);
assert!(sent_peer.last_contact_ms.is_some());
assert!(sent_peer.last_success_ms.is_some());
