diff --git a/docs/metrics.md b/docs/metrics.md index c538fbb41..94f84eab0 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -159,6 +159,14 @@ overlay.inbound.live | counter | number of live inbound c overlay.outbound-queue. | timer | time traffic sits in flow-controlled queues overlay.outbound-queue.drop- | meter | number of messages dropped from flow-controlled queues overlay.item-fetcher.next-peer | meter | ask for item past the first one +overlay.item-fetcher.claim-ask | meter | tx set fetch ask targeted a peer that claimed possession via HAVE_TX_SET +overlay.item-fetcher.claim-dropped | meter | HAVE_TX_SET dropped at admission: sending peer exceeded its budget for the current window +overlay.item-fetcher.claim-grace-wait | timer | time a tx set fetch waited from creation to its first ask (claim grace) +overlay.item-fetcher.claim-grace-satisfied | meter | tx set fetch's first ask targeted a claimed holder +overlay.item-fetcher.claim-grace-expired | meter | tx set fetch's first ask fell back to an SCP relayer or random peer +overlay.send.have-tx-set | meter | HAVE_TX_SET messages sent +overlay.recv.have-tx-set | meter | HAVE_TX_SET messages received +overlay.fetch.txset-abandoned | meter | tx set fetches abandoned by the age backstop overlay.memory.flood-known | counter | number of known flooded entries overlay.message.broadcast | meter | message broadcasted overlay.message.read | meter | message received diff --git a/overlay/src/flood/mod.rs b/overlay/src/flood/mod.rs index 9e796310e..683def73e 100644 --- a/overlay/src/flood/mod.rs +++ b/overlay/src/flood/mod.rs @@ -9,6 +9,7 @@ mod mempool; mod pending_requests; mod tx_buffer; mod txset; +mod txset_fetch; pub use inv_batcher::InvBatcher; pub use inv_messages::{GetData, InvBatch, InvEntry, TxStreamMessage}; @@ -17,3 +18,7 @@ pub use mempool::Mempool; pub use pending_requests::PendingRequests; pub use tx_buffer::TxBuffer; pub use txset::{CachedTxSet, Hash256, TxSetCache}; +pub use txset_fetch::{ + Ask, AskTier, CompletedFetch, GraceOutcome, TickResult, TxSetFetcher, TXSET_ASK_TIMEOUT, + TXSET_FETCH_GRACE, +}; diff --git a/overlay/src/flood/txset_fetch.rs b/overlay/src/flood/txset_fetch.rs new file mode 100644 index 000000000..d6cea23ae --- /dev/null +++ b/overlay/src/flood/txset_fetch.rs @@ -0,0 +1,799 @@ +//! Tx set fetch lifecycle: source tracking, pending fetches, tiered retry. +//! +//! This is the Rust analog of the C++ `ItemFetcher`/`Tracker` pair, ported for +//! the `HAVE_TX_SET` protocol change (stellar-core PR #5379). Peers that might +//! hold a tx set are tracked in two tiers: +//! +//! - **claimants**: peers that explicitly claimed possession via `HAVE_TX_SET`; +//! - **relayers**: peers that relayed an SCP envelope referencing the hash +//! (which, once the protocol allows empty-tx-set values and parallel +//! downloads, no longer implies possession). +//! +//! Fetches are retried on a timeout cadence, targeting claimants first, then +//! relayers, then any connected peer. A fetch whose ask times out drops that +//! peer's claim (the analog of the C++ `DONT_HAVE` handling — the QUIC txset +//! protocol has no negative reply, so a timeout is our miss signal). +//! +//! All methods take `now: Instant` explicitly so the logic is fully +//! deterministic under test; the driver passes `Instant::now()`. + +use libp2p::PeerId; +use lru::LruCache; +use std::collections::{HashMap, HashSet}; +use std::num::NonZeroUsize; +use std::time::{Duration, Instant}; + +use super::txset::Hash256; + +/// How long one ask may go unanswered before we retarget. Mirrors the C++ +/// `Tracker::MS_TO_WAIT_FOR_FETCH_PROGRESS`. +pub const TXSET_ASK_TIMEOUT: Duration = Duration::from_millis(1500); + +/// Claim grace period: with empty-tx-set values possible and no claimant +/// known, defer the first blind ask up to this long waiting for a +/// `HAVE_TX_SET` claim. Mirrors the C++ claim grace. +pub const TXSET_FETCH_GRACE: Duration = Duration::from_millis(1500); + +/// Backoff unit applied between candidate-list rebuilds, scaled by the number +/// of rebuilds so far (capped). Mirrors the C++ rebuild backoff. +pub const TXSET_REBUILD_BACKOFF_UNIT: Duration = Duration::from_millis(1500); +pub const TXSET_MAX_REBUILD_BACKOFF_MULT: u32 = 10; + +/// Absolute age backstop: a fetch this old is abandoned even if slot-based +/// purging never caught it (leak protection; slot purging is the normal path). +pub const TXSET_FETCH_MAX_AGE: Duration = Duration::from_secs(600); + +/// Distinct hashes for which sources (claims/relays) are remembered. Matches +/// the C++ `BUFFERED_CLAIMS_CACHE_SIZE`; also subsumes the old +/// `txset_sources` LRU (same capacity). +pub const TXSET_SOURCES_CACHE_SIZE: usize = 1000; + +/// Per-hash bound on remembered peers in each tier. +pub const MAX_SOURCES_PER_HASH: usize = 8; + +/// Which tier the selected peer came from. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AskTier { + /// The peer explicitly claimed possession via `HAVE_TX_SET`. + Claimant, + /// The peer relayed an SCP envelope referencing the hash. + Relayer, + /// No better information: any connected peer. + Blind, +} + +/// Outcome of the claim grace period, reported once per fetch on its first +/// dispatched ask (only when the grace was enabled for the fetch). +#[derive(Debug, Clone, Copy)] +pub struct GraceOutcome { + /// Time from fetch creation to the first ask. + pub waited: Duration, + /// Whether the first ask targeted a claimant. + pub satisfied: bool, +} + +/// A dispatch decision: ask `peer` for `hash`. Produced by [`TxSetFetcher::tick`] +/// and [`TxSetFetcher::dispatch_one`]; the driver performs the actual send and +/// reports back via `mark_sent` / `mark_send_failed`. +#[derive(Debug)] +pub struct Ask { + pub hash: Hash256, + pub peer: PeerId, + pub tier: AskTier, + /// Present only on the fetch's first ask when the grace was enabled. + pub grace_outcome: Option, +} + +/// A completed fetch, for latency metrics. +#[derive(Debug)] +pub struct CompletedFetch { + pub slot: u32, + pub elapsed: Duration, +} + +#[derive(Default)] +struct TxSetSources { + claimants: Vec, + relayers: Vec, +} + +fn push_bounded(list: &mut Vec, peer: PeerId) { + if list.contains(&peer) { + return; + } + if list.len() >= MAX_SOURCES_PER_HASH { + list.remove(0); + } + list.push(peer); +} + +struct PendingFetch { + /// The peer currently asked; `None` while undispatched (grace period, + /// no candidates, backoff, or send failure). + peer: Option, + /// When the current ask was (last) dispatched or confirmed on the wire. + sent_at: Instant, + created_at: Instant, + slot: u32, + /// Number of candidate-list rebuilds; scales the backoff. + rebuilds: u32, + /// Gate for the next dispatch attempt (backoff after a rebuild). + next_dispatch_at: Instant, + /// Peers asked since the last rebuild. + asked: HashSet, + /// Whether the claim grace applies to this fetch (captured at creation). + grace_enabled: bool, + /// Whether the grace outcome has been reported (first ask happened). + grace_resolved: bool, +} + +/// Result of one housekeeping tick. +#[derive(Debug, Default)] +pub struct TickResult { + /// Asks the driver should send now. + pub asks: Vec, + /// Fetches abandoned by the absolute age backstop. + pub expired: Vec, +} + +/// Tracks who might have which tx set and drives pending fetches to +/// completion. Pure state machine: no I/O, no clocks of its own. +pub struct TxSetFetcher { + sources: LruCache, + pending: HashMap, + /// Whether the current ledger protocol admits empty-tx-set values (the + /// condition under which SCP relayers may not possess referenced sets and + /// the claim grace is worth paying). Set from core via IPC. + empty_tx_sets_possible: bool, +} + +impl TxSetFetcher { + pub fn new() -> Self { + TxSetFetcher { + sources: LruCache::new(NonZeroUsize::new(TXSET_SOURCES_CACHE_SIZE).unwrap()), + pending: HashMap::new(), + empty_tx_sets_possible: false, + } + } + + pub fn set_empty_tx_sets_possible(&mut self, possible: bool) { + self.empty_tx_sets_possible = possible; + } + + /// Record that `peer` relayed an SCP envelope referencing `hash` (weak + /// evidence of possession). + pub fn record_relayer(&mut self, hash: Hash256, peer: PeerId) { + push_bounded( + &mut self + .sources + .get_or_insert_mut(hash, Default::default) + .relayers, + peer, + ); + } + + /// Record that `peer` explicitly claimed possession of `hash` via + /// `HAVE_TX_SET` (strong evidence). Returns true if the claim is relevant + /// to a pending fetch that currently has no ask outstanding, i.e. the + /// driver should run a dispatch pass promptly rather than wait for the + /// next tick. + pub fn record_claim(&mut self, hash: Hash256, peer: PeerId) -> bool { + push_bounded( + &mut self + .sources + .get_or_insert_mut(hash, Default::default) + .claimants, + peer, + ); + match self.pending.get_mut(&hash) { + Some(entry) if entry.peer.is_none() => { + // Act on the claim immediately (mirrors the C++ tracker + // canceling its retry timer on a claim): lift any rebuild + // backoff so the next dispatch pass is not gated. + entry.next_dispatch_at = entry.created_at; + true + } + _ => false, + } + } + + /// Start fetching `hash` for `slot`. Returns true if this created a new + /// pending fetch; false if one already existed (in which case its slot is + /// raised to `slot` if that is newer, so slot-based purging can never + /// strand a fetch that a newer slot still needs). + pub fn start_fetch(&mut self, hash: Hash256, slot: u32, now: Instant) -> bool { + if let Some(existing) = self.pending.get_mut(&hash) { + existing.slot = existing.slot.max(slot); + return false; + } + self.pending.insert( + hash, + PendingFetch { + peer: None, + sent_at: now, + created_at: now, + slot, + rebuilds: 0, + next_dispatch_at: now, + asked: HashSet::new(), + grace_enabled: self.empty_tx_sets_possible, + grace_resolved: false, + }, + ); + true + } + + /// Whether a fetch for `hash` is pending. + pub fn is_pending(&self, hash: &Hash256) -> bool { + self.pending.contains_key(hash) + } + + pub fn pending_len(&self) -> usize { + self.pending.len() + } + + /// The peer currently asked for `hash`, if an ask is outstanding. + pub fn asked_peer(&self, hash: &Hash256) -> Option { + self.pending.get(hash).and_then(|e| e.peer) + } + + /// The fetch completed: the set arrived (from anyone). Returns latency + /// info if a fetch was pending. + pub fn complete(&mut self, hash: &Hash256, now: Instant) -> Option { + self.pending.remove(hash).map(|entry| CompletedFetch { + slot: entry.slot, + elapsed: now.saturating_duration_since(entry.created_at), + }) + } + + /// The send for `hash`'s current ask actually reached the wire: restart + /// the response window (excludes local queueing delay from the timeout). + pub fn mark_sent(&mut self, hash: &Hash256, now: Instant) { + if let Some(entry) = self.pending.get_mut(hash) { + if entry.peer.is_some() { + entry.sent_at = now; + } + } + } + + /// The send for `hash`'s current ask failed: clear the ask so the next + /// dispatch pass retargets immediately. The failed peer stays in `asked` + /// for this round so we don't hammer it. + pub fn mark_send_failed(&mut self, hash: &Hash256) { + if let Some(entry) = self.pending.get_mut(hash) { + entry.peer = None; + } + } + + /// A peer disconnected: clear any outstanding asks to it (they will be + /// retargeted on the next dispatch pass) and forget it as a source. + pub fn peer_disconnected(&mut self, peer: &PeerId) { + for entry in self.pending.values_mut() { + if entry.peer.as_ref() == Some(peer) { + entry.peer = None; + } + } + // LruCache has no retain; walk the (bounded) entries. + for (_, sources) in self.sources.iter_mut() { + sources.claimants.retain(|p| p != peer); + sources.relayers.retain(|p| p != peer); + } + } + + /// Drop pending fetches for slots strictly below `slot` (mirrors the tx + /// set cache eviction horizon). Fetches created before slots were known + /// (slot 0) are kept. + pub fn evict_before(&mut self, slot: u32) { + self.pending + .retain(|_, entry| entry.slot == 0 || entry.slot >= slot); + } + + /// Run one housekeeping pass: time out stale asks (dropping the asked + /// peer's claim — our miss signal), dispatch undispatched fetches, and + /// abandon fetches past the absolute age backstop. + pub fn tick(&mut self, now: Instant, connected: &[PeerId]) -> TickResult { + let mut result = TickResult::default(); + + // Phase 1: expire by age, and time out stale asks. + let mut timed_out: Vec = Vec::new(); + self.pending.retain(|hash, entry| { + if now.saturating_duration_since(entry.created_at) >= TXSET_FETCH_MAX_AGE { + result.expired.push(*hash); + return false; + } + if entry.peer.is_some() + && now.saturating_duration_since(entry.sent_at) >= TXSET_ASK_TIMEOUT + { + timed_out.push(*hash); + } + true + }); + for hash in timed_out { + let entry = self.pending.get_mut(&hash).unwrap(); + let peer = entry.peer.take().unwrap(); + // The ask went unanswered: any claim this peer made was wrong. + if let Some(sources) = self.sources.peek_mut(&hash) { + sources.claimants.retain(|p| p != &peer); + } + } + + // Phase 2: dispatch whatever is undispatched and eligible. + let hashes: Vec = self + .pending + .iter() + .filter(|(_, e)| e.peer.is_none()) + .map(|(h, _)| *h) + .collect(); + for hash in hashes { + if let Some(ask) = self.try_dispatch(hash, now, connected) { + result.asks.push(ask); + } + } + result + } + + /// Attempt to dispatch the fetch for `hash` right now (used at fetch start + /// and when a claim arrives, so a claim is acted on immediately rather + /// than at the next tick). No-op if an ask is already outstanding. + pub fn dispatch_one( + &mut self, + hash: Hash256, + now: Instant, + connected: &[PeerId], + ) -> Option { + match self.pending.get(&hash) { + Some(entry) if entry.peer.is_none() => self.try_dispatch(hash, now, connected), + _ => None, + } + } + + /// Core dispatch: pick a target by tier, honoring grace and backoff. + /// Precondition: `hash` is pending and has no ask outstanding. + fn try_dispatch(&mut self, hash: Hash256, now: Instant, connected: &[PeerId]) -> Option { + let entry = self.pending.get_mut(&hash)?; + if now < entry.next_dispatch_at { + return None; // rebuild backoff + } + + let empty_sources = TxSetSources::default(); + let sources = self.sources.peek(&hash).unwrap_or(&empty_sources); + + let pick = |tier: &[PeerId], skip_asked: bool| -> Option { + tier.iter() + .find(|p| connected.contains(p) && (!skip_asked || !entry.asked.contains(p))) + .cloned() + }; + + let selected: Option<(PeerId, AskTier)> = + // Fresh claimants first, then claimants we already asked (a claim + // re-enables a peer that previously missed). + pick(&sources.claimants, true) + .or_else(|| pick(&sources.claimants, false)) + .map(|p| (p, AskTier::Claimant)) + .or_else(|| pick(&sources.relayers, true).map(|p| (p, AskTier::Relayer))) + .or_else(|| { + connected + .iter() + .find(|p| !entry.asked.contains(p)) + .cloned() + .map(|p| (p, AskTier::Blind)) + }); + + let (peer, tier) = match selected { + Some(sel) => sel, + None => { + // Every known candidate has been asked this round (or nobody + // is connected): rebuild the round with backoff. + entry.asked.clear(); + entry.rebuilds += 1; + let mult = entry.rebuilds.min(TXSET_MAX_REBUILD_BACKOFF_MULT); + entry.next_dispatch_at = now + TXSET_REBUILD_BACKOFF_UNIT * mult; + return None; + } + }; + + // Claim grace: before the first ask, with the grace enabled and no + // claimant available, hold off a bounded time in case a claim arrives. + if entry.grace_enabled + && !entry.grace_resolved + && tier != AskTier::Claimant + && now.saturating_duration_since(entry.created_at) < TXSET_FETCH_GRACE + { + return None; + } + + let grace_outcome = if entry.grace_enabled && !entry.grace_resolved { + entry.grace_resolved = true; + Some(GraceOutcome { + waited: now.saturating_duration_since(entry.created_at), + satisfied: tier == AskTier::Claimant, + }) + } else { + entry.grace_resolved = true; + None + }; + + entry.peer = Some(peer); + entry.sent_at = now; + entry.asked.insert(peer); + + Some(Ask { + hash, + peer, + tier, + grace_outcome, + }) + } +} + +impl Default for TxSetFetcher { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn hash(n: u8) -> Hash256 { + [n; 32] + } + + fn setup() -> (TxSetFetcher, Instant, PeerId, PeerId, PeerId) { + ( + TxSetFetcher::new(), + Instant::now(), + PeerId::random(), + PeerId::random(), + PeerId::random(), + ) + } + + #[test] + fn dispatch_prefers_claimant_over_relayer_over_blind() { + let (mut f, now, claimant, relayer, other) = setup(); + f.record_relayer(hash(1), relayer); + f.record_claim(hash(1), claimant); + + assert!(f.start_fetch(hash(1), 5, now)); + let connected = vec![other, relayer, claimant]; + let ask = f.dispatch_one(hash(1), now, &connected).unwrap(); + assert_eq!(ask.peer, claimant); + assert_eq!(ask.tier, AskTier::Claimant); + + // With the claimant gone, the relayer tier is next. + let mut f2 = TxSetFetcher::new(); + f2.record_relayer(hash(1), relayer); + f2.start_fetch(hash(1), 5, now); + let ask = f2.dispatch_one(hash(1), now, &connected).unwrap(); + assert_eq!(ask.peer, relayer); + assert_eq!(ask.tier, AskTier::Relayer); + + // With no sources at all, any connected peer is asked (blind). + let mut f3 = TxSetFetcher::new(); + f3.start_fetch(hash(1), 5, now); + let ask = f3.dispatch_one(hash(1), now, &connected).unwrap(); + assert_eq!(ask.tier, AskTier::Blind); + } + + #[test] + fn disconnected_sources_are_skipped() { + let (mut f, now, claimant, relayer, other) = setup(); + f.record_claim(hash(1), claimant); + f.record_relayer(hash(1), relayer); + f.start_fetch(hash(1), 5, now); + + // Neither source is connected: blind ask to the only connected peer. + let ask = f.dispatch_one(hash(1), now, &[other]).unwrap(); + assert_eq!(ask.peer, other); + assert_eq!(ask.tier, AskTier::Blind); + } + + #[test] + fn timeout_retargets_and_drops_wrong_claim() { + let (mut f, now, claimant, relayer, _) = setup(); + f.record_claim(hash(1), claimant); + f.record_relayer(hash(1), relayer); + f.start_fetch(hash(1), 5, now); + + let connected = vec![claimant, relayer]; + let ask = f.dispatch_one(hash(1), now, &connected).unwrap(); + assert_eq!(ask.peer, claimant); + + // Before the timeout: nothing happens. + let r = f.tick(now + TXSET_ASK_TIMEOUT / 2, &connected); + assert!(r.asks.is_empty()); + + // At the timeout: the claim was wrong (dropped), retarget to relayer. + let r = f.tick(now + TXSET_ASK_TIMEOUT, &connected); + assert_eq!(r.asks.len(), 1); + assert_eq!(r.asks[0].peer, relayer); + assert_eq!(r.asks[0].tier, AskTier::Relayer); + } + + #[test] + fn claim_reenables_previously_asked_peer() { + let (mut f, now, p1, p2, _) = setup(); + let connected = vec![p1, p2]; + f.start_fetch(hash(1), 5, now); + + // Blind asks burn through both peers. + let first = f.dispatch_one(hash(1), now, &connected).unwrap().peer; + let now2 = now + TXSET_ASK_TIMEOUT; + let second = f.tick(now2, &connected).asks.pop().unwrap().peer; + assert_ne!(first, second); + + // Both asked and timed out: the round is exhausted (rebuild+backoff), + // no ask goes out. + let now3 = now2 + TXSET_ASK_TIMEOUT; + assert!(f.tick(now3, &connected).asks.is_empty()); + + // The first peer now claims possession: the claim lifts the rebuild + // backoff, and the peer is re-asked in spite of having been asked + // (and missed) before. + assert!(f.record_claim(hash(1), first)); + let asks = f.tick(now3, &connected).asks; + assert_eq!(asks.len(), 1); + assert_eq!(asks[0].peer, first); + assert_eq!(asks[0].tier, AskTier::Claimant); + } + + #[test] + fn claim_with_no_pending_fetch_is_passive() { + let (mut f, now, p1, p2, _) = setup(); + assert!(!f.record_claim(hash(1), p1)); + assert_eq!(f.pending_len(), 0); + + // A claim while an ask is outstanding is recorded but not actionable + // (mirrors C++: does not interrupt the outstanding ask). + f.start_fetch(hash(2), 5, now); + f.dispatch_one(hash(2), now, &[p1]); + assert!(!f.record_claim(hash(2), p2)); + assert_eq!(f.asked_peer(&hash(2)), Some(p1)); + } + + #[test] + fn claim_for_undispatched_fetch_lifts_backoff_and_dispatches() { + let (mut f, now, p1, _, _) = setup(); + f.start_fetch(hash(1), 5, now); + // Dispatch attempt with nobody connected: rebuild + backoff. + assert!(f.dispatch_one(hash(1), now, &[]).is_none()); + + // Peer connects and claims: actionable, and the backoff no longer + // gates the dispatch — it happens immediately. + assert!(f.record_claim(hash(1), p1)); + let ask = f.dispatch_one(hash(1), now, &[p1]).unwrap(); + assert_eq!(ask.peer, p1); + assert_eq!(ask.tier, AskTier::Claimant); + } + + #[test] + fn buffered_claim_seeds_first_ask() { + // Claim arrives before any fetch: recorded passively, and the first + // ask targets the claimant once a fetch starts. + let (mut f, now, claimant, other, _) = setup(); + assert!(!f.record_claim(hash(1), claimant)); + + f.start_fetch(hash(1), 5, now); + let ask = f.dispatch_one(hash(1), now, &[other, claimant]).unwrap(); + assert_eq!(ask.peer, claimant); + assert_eq!(ask.tier, AskTier::Claimant); + } + + #[test] + fn repeated_claims_deduplicate_and_bound() { + let (mut f, _, p1, p2, _) = setup(); + f.record_claim(hash(1), p1); + f.record_claim(hash(1), p1); + f.record_claim(hash(1), p1); + f.record_claim(hash(1), p2); + let sources = f.sources.peek(&hash(1)).unwrap(); + assert_eq!(sources.claimants.len(), 2); + + // The per-hash bound holds under a flood of distinct claimants. + for _ in 0..(MAX_SOURCES_PER_HASH * 2) { + f.record_claim(hash(1), PeerId::random()); + } + assert_eq!( + f.sources.peek(&hash(1)).unwrap().claimants.len(), + MAX_SOURCES_PER_HASH + ); + } + + #[test] + fn send_failure_keeps_fetch_and_retargets() { + let (mut f, now, p1, p2, _) = setup(); + f.start_fetch(hash(1), 5, now); + let ask = f.dispatch_one(hash(1), now, &[p1, p2]).unwrap(); + let failed = ask.peer; + + f.mark_send_failed(&hash(1)); + assert!(f.is_pending(&hash(1))); + assert!(f.asked_peer(&hash(1)).is_none()); + + // Retarget goes to the other peer (failed one stays in `asked`). + let ask = f.dispatch_one(hash(1), now, &[p1, p2]).unwrap(); + assert_ne!(ask.peer, failed); + } + + #[test] + fn disconnect_clears_ask_but_keeps_fetch() { + let (mut f, now, p1, p2, _) = setup(); + f.start_fetch(hash(1), 5, now); + let ask = f.dispatch_one(hash(1), now, &[p1]).unwrap(); + assert_eq!(ask.peer, p1); + + f.peer_disconnected(&p1); + assert!(f.is_pending(&hash(1)), "fetch must survive the disconnect"); + assert!(f.asked_peer(&hash(1)).is_none()); + + // Next tick retargets to the remaining peer. + let asks = f.tick(now + Duration::from_millis(1), &[p2]).asks; + assert_eq!(asks.len(), 1); + assert_eq!(asks[0].peer, p2); + } + + #[test] + fn fetch_with_no_peers_parks_until_a_peer_appears() { + let (mut f, now, p1, _, _) = setup(); + f.start_fetch(hash(1), 5, now); + assert!(f.dispatch_one(hash(1), now, &[]).is_none()); + assert!(f.is_pending(&hash(1)), "fetch must not be lost"); + + // A peer connects; after the rebuild backoff the fetch dispatches. + let later = now + TXSET_REBUILD_BACKOFF_UNIT; + let asks = f.tick(later, &[p1]).asks; + assert_eq!(asks.len(), 1); + assert_eq!(asks[0].peer, p1); + } + + #[test] + fn rebuild_backoff_scales_with_attempts() { + let (mut f, now, _, _, _) = setup(); + f.start_fetch(hash(1), 5, now); + + // Two empty dispatch attempts (rebuilds 1 and 2). The second is only + // permitted once the first backoff (1 * unit) expired. + assert!(f.dispatch_one(hash(1), now, &[]).is_none()); + let after_first = now + TXSET_REBUILD_BACKOFF_UNIT; + assert!(f.tick(after_first, &[]).asks.is_empty()); // rebuild #2 + + // Backoff is now 2 * unit: a peer connecting at 1 * unit later is not + // asked yet; at 2 * unit it is. + let p = PeerId::random(); + let too_soon = after_first + TXSET_REBUILD_BACKOFF_UNIT - Duration::from_millis(1); + assert!(f.tick(too_soon, &[p]).asks.is_empty()); + let due = after_first + 2 * TXSET_REBUILD_BACKOFF_UNIT; + assert_eq!(f.tick(due, &[p]).asks.len(), 1); + } + + #[test] + fn complete_reports_slot_and_latency() { + let (mut f, now, p1, _, _) = setup(); + f.start_fetch(hash(1), 42, now); + f.dispatch_one(hash(1), now, &[p1]); + + let done = f + .complete(&hash(1), now + Duration::from_millis(300)) + .unwrap(); + assert_eq!(done.slot, 42); + assert_eq!(done.elapsed, Duration::from_millis(300)); + assert!(!f.is_pending(&hash(1))); + assert!(f.complete(&hash(1), now).is_none()); + } + + #[test] + fn duplicate_start_fetch_raises_slot_only() { + let (mut f, now, _, _, _) = setup(); + assert!(f.start_fetch(hash(1), 5, now)); + assert!(!f.start_fetch(hash(1), 9, now)); + assert!(!f.start_fetch(hash(1), 3, now)); + assert_eq!(f.pending.get(&hash(1)).unwrap().slot, 9); + assert_eq!(f.pending_len(), 1); + } + + #[test] + fn evict_before_purges_old_slots_keeps_new_and_unknown() { + let (mut f, now, _, _, _) = setup(); + f.start_fetch(hash(1), 10, now); + f.start_fetch(hash(2), 100, now); + f.start_fetch(hash(3), 0, now); // slot unknown + + f.evict_before(50); + assert!(!f.is_pending(&hash(1))); + assert!(f.is_pending(&hash(2))); + assert!(f.is_pending(&hash(3))); + } + + #[test] + fn age_backstop_expires_ancient_fetches() { + let (mut f, now, p1, _, _) = setup(); + f.start_fetch(hash(1), 5, now); + let r = f.tick(now + TXSET_FETCH_MAX_AGE, &[p1]); + assert_eq!(r.expired, vec![hash(1)]); + assert!(!f.is_pending(&hash(1))); + } + + // --- grace period --- + + #[test] + fn grace_defers_blind_ask_until_claim_or_expiry() { + let (mut f, now, relayer, claimant, _) = setup(); + f.set_empty_tx_sets_possible(true); + f.record_relayer(hash(1), relayer); + f.start_fetch(hash(1), 5, now); + + // No claimant: the first ask is deferred, even with a relayer ready. + assert!(f.dispatch_one(hash(1), now, &[relayer]).is_none()); + assert!(f + .tick(now + TXSET_FETCH_GRACE / 2, &[relayer]) + .asks + .is_empty()); + + // A claim preempts the wait immediately. + assert!(f.record_claim(hash(1), claimant)); + let asks = f + .tick(now + TXSET_FETCH_GRACE / 2, &[relayer, claimant]) + .asks; + assert_eq!(asks.len(), 1); + assert_eq!(asks[0].peer, claimant); + let outcome = asks[0].grace_outcome.expect("first ask reports grace"); + assert!(outcome.satisfied); + assert_eq!(outcome.waited, TXSET_FETCH_GRACE / 2); + } + + #[test] + fn grace_expiry_falls_back_to_relayer() { + let (mut f, now, relayer, _, _) = setup(); + f.set_empty_tx_sets_possible(true); + f.record_relayer(hash(1), relayer); + f.start_fetch(hash(1), 5, now); + + assert!(f.dispatch_one(hash(1), now, &[relayer]).is_none()); + let asks = f.tick(now + TXSET_FETCH_GRACE, &[relayer]).asks; + assert_eq!(asks.len(), 1); + assert_eq!(asks[0].peer, relayer); + let outcome = asks[0].grace_outcome.expect("first ask reports grace"); + assert!(!outcome.satisfied); + assert_eq!(outcome.waited, TXSET_FETCH_GRACE); + } + + #[test] + fn grace_reported_once_then_never_again() { + let (mut f, now, p1, p2, _) = setup(); + f.set_empty_tx_sets_possible(true); + f.start_fetch(hash(1), 5, now); + + let t1 = now + TXSET_FETCH_GRACE; + let asks = f.tick(t1, &[p1, p2]).asks; + assert!(asks[0].grace_outcome.is_some()); + + // Retarget after timeout: no second grace outcome. + let t2 = t1 + TXSET_ASK_TIMEOUT; + let asks = f.tick(t2, &[p1, p2]).asks; + assert_eq!(asks.len(), 1); + assert!(asks[0].grace_outcome.is_none()); + } + + #[test] + fn grace_disabled_means_immediate_first_ask_and_no_outcome() { + let (mut f, now, relayer, _, _) = setup(); + // empty_tx_sets_possible defaults to false. + f.record_relayer(hash(1), relayer); + f.start_fetch(hash(1), 5, now); + + let ask = f.dispatch_one(hash(1), now, &[relayer]).unwrap(); + assert_eq!(ask.peer, relayer); + assert!(ask.grace_outcome.is_none()); + } + + #[test] + fn grace_flag_captured_at_fetch_creation() { + let (mut f, now, p1, _, _) = setup(); + f.start_fetch(hash(1), 5, now); + // Turning the flag on later must not retroactively delay this fetch. + f.set_empty_tx_sets_possible(true); + assert!(f.dispatch_one(hash(1), now, &[p1]).is_some()); + } +} diff --git a/overlay/src/libp2p_overlay.rs b/overlay/src/libp2p_overlay.rs index f5e4396e0..649f439da 100644 --- a/overlay/src/libp2p_overlay.rs +++ b/overlay/src/libp2p_overlay.rs @@ -12,7 +12,8 @@ //! QUIC provides independent loss recovery per stream. use crate::flood::{ - GetData, InvBatch, InvBatcher, InvEntry, InvTracker, PendingRequests, TxBuffer, TxStreamMessage, + Ask, AskTier, GetData, InvBatch, InvBatcher, InvEntry, InvTracker, PendingRequests, TxBuffer, + TxSetFetcher, TxStreamMessage, }; use crate::metrics::OverlayMetrics; use crate::wire::ValidatedTx; @@ -48,6 +49,14 @@ const MAX_MESSAGE_SIZE: usize = 16 * 1024 * 1024; /// TXs that can't be queued are dropped - they'll be re-requested if needed. const TX_EVENT_CHANNEL_CAPACITY: usize = 10_000; +/// Max HAVE_TX_SET messages admitted per peer per admission window. Mirrors +/// the C++ `Peer::HAVE_TX_SET_MAX_PER_PERIOD`. +pub const HAVE_TX_SET_MAX_PER_PERIOD: u32 = 32; + +/// The HAVE_TX_SET admission window. Mirrors the C++ per-peer recurrent-timer +/// period (a single global reset tick here, equivalent in effect). +pub const HAVE_TX_SET_PERIOD: Duration = Duration::from_secs(5); + /// Events from the overlay to the application #[derive(Debug, Clone)] pub enum OverlayEvent { @@ -99,6 +108,16 @@ pub enum OverlayCommand { }, /// Record that a peer has a specific TX set (learned from SCP message) RecordTxSetSource { hash: [u8; 32], peer: PeerId }, + /// Announce to all connected peers that we possess a TX set (HAVE_TX_SET) + AnnounceTxSet { hash: [u8; 32] }, + /// Send HAVE_TX_SET for one hash to one peer (claims accompanying SCP + /// state sent to a catching-up peer) + SendHaveTxSetToPeer { hash: [u8; 32], to: PeerId }, + /// Drop pending TX set fetches for slots below the given one + PurgeTxSetFetchesBelow { slot: u32 }, + /// Whether the ledger protocol admits empty-tx-set values (enables the + /// claim grace period on new fetches) + SetEmptyTxSetsPossible(bool), /// Connect to a peer by address (bootstrap — PeerId unknown) Dial(Multiaddr), /// Connect to a known peer by PeerId (reconnect — deduplicates automatically) @@ -228,6 +247,62 @@ impl OverlayHandle { } } + /// Send HAVE_TX_SET for `hash` to a single peer + pub async fn send_have_txset(&self, hash: [u8; 32], to: PeerId) { + if let Err(e) = self + .cmd_tx + .send(OverlayCommand::SendHaveTxSetToPeer { hash, to }) + .await + { + warn!( + "Overlay command channel closed, failed to send SendHaveTxSetToPeer: {}", + e + ); + } + } + + /// Announce possession of a TX set to all connected peers via HAVE_TX_SET + pub async fn announce_txset(&self, hash: [u8; 32]) { + if let Err(e) = self + .cmd_tx + .send(OverlayCommand::AnnounceTxSet { hash }) + .await + { + warn!( + "Overlay command channel closed, failed to send AnnounceTxSet: {}", + e + ); + } + } + + /// Drop pending TX set fetches for slots below `slot` (ledger closed) + pub async fn purge_txset_fetches_below(&self, slot: u32) { + if let Err(e) = self + .cmd_tx + .send(OverlayCommand::PurgeTxSetFetchesBelow { slot }) + .await + { + warn!( + "Overlay command channel closed, failed to send PurgeTxSetFetchesBelow: {}", + e + ); + } + } + + /// Set whether the ledger protocol admits empty-tx-set values + pub async fn set_empty_tx_sets_possible(&self, possible: bool) { + if let Err(e) = self + .cmd_tx + .send(OverlayCommand::SetEmptyTxSetsPossible(possible)) + .await + { + warn!( + "Overlay command channel closed, failed to send SetEmptyTxSetsPossible: {}", + e + ); + } + } + pub async fn dial(&self, addr: Multiaddr) { if let Err(e) = self.cmd_tx.send(OverlayCommand::Dial(addr)).await { warn!("Overlay command channel closed, failed to send Dial: {}", e); @@ -310,11 +385,12 @@ struct SharedState { tx_seen: RwLock>, /// Track which peers we've sent each SCP message to (prevent duplicate sends) scp_sent_to: RwLock>>, - /// TX set sources: which peer has which TX set (learned from SCP messages) - txset_sources: RwLock>, - /// Pending TX set requests: hash -> (peer, request_time) to avoid duplicate fetches and track latency - /// hash -> (peer asked, request time, slot the set is for) - pending_txset_requests: RwLock>, + /// TX set fetch state: who might have which set (HAVE_TX_SET claimants + /// and SCP relayers) plus pending fetches with tiered retry. + txset_fetcher: RwLock, + /// HAVE_TX_SET admission budget consumed per peer in the current window + /// (cleared every HAVE_TX_SET_PERIOD by the housekeeping task). + have_txset_admitted: RwLock>, /// Event sender for non-TX events (SCP, TxSet - critical path, unbounded) event_tx: mpsc::UnboundedSender, /// Bounded TX event sender (backpressure - drops allowed) @@ -355,10 +431,8 @@ impl SharedState { scp_sent_to: RwLock::new(lru::LruCache::new( std::num::NonZeroUsize::new(10000).unwrap(), )), - txset_sources: RwLock::new(lru::LruCache::new( - std::num::NonZeroUsize::new(1000).unwrap(), - )), - pending_txset_requests: RwLock::new(HashMap::new()), + txset_fetcher: RwLock::new(TxSetFetcher::new()), + have_txset_admitted: RwLock::new(HashMap::new()), event_tx, tx_event_tx, tx_dropped_count: AtomicU64::new(0), @@ -535,9 +609,25 @@ impl StellarOverlay { self.send_txset_response(to, hash, data).await; } OverlayCommand::RecordTxSetSource { hash, peer } => { - let mut sources = self.state.txset_sources.write().await; - sources.put(hash, peer); - debug!("Recorded peer {} as source for TX set {:02x?}...", peer, &hash[..4]); + let mut fetcher = self.state.txset_fetcher.write().await; + fetcher.record_relayer(hash, peer); + debug!("Recorded peer {} as SCP relayer for TX set {:02x?}...", peer, &hash[..4]); + } + OverlayCommand::AnnounceTxSet { hash } => { + self.announce_txset(hash).await; + } + OverlayCommand::SendHaveTxSetToPeer { hash, to } => { + self.send_have_txset_to_peer(hash, to).await; + } + OverlayCommand::PurgeTxSetFetchesBelow { slot } => { + self.state.txset_fetcher.write().await.evict_before(slot); + } + OverlayCommand::SetEmptyTxSetsPossible(possible) => { + self.state + .txset_fetcher + .write() + .await + .set_empty_tx_sets_possible(possible); } OverlayCommand::Dial(addr) => { info!("Dialing peer at {}", addr); @@ -687,19 +777,19 @@ impl StellarOverlay { let mut streams = self.state.peer_streams.write().await; streams.remove(&peer_id); } - // Clean up pending txset requests for this peer - { - let mut pending = self.state.pending_txset_requests.write().await; - let before_len = pending.len(); - pending.retain(|_hash, (p, _, _)| p != &peer_id); - let removed = before_len - pending.len(); - if removed > 0 { - info!( - "Removed {} pending txset requests for disconnected peer {}", - removed, peer_id - ); - } - } + // Clear outstanding tx set asks to this peer (the fetches + // survive and retarget on the next housekeeping tick) and + // forget it as a claimant/relayer. + self.state + .txset_fetcher + .write() + .await + .peer_disconnected(&peer_id); + self.state + .have_txset_admitted + .write() + .await + .remove(&peer_id); // Notify main loop to clean up any pending requests for this peer if let Err(e) = self.state.event_tx.send(OverlayEvent::PeerDisconnected { peer_id: peer_id.clone(), @@ -896,108 +986,59 @@ impl StellarOverlay { } } - /// Fetch TX set from a peer - preferring the peer who sent us the SCP message referencing it + /// Start (or join) a TX set fetch. Targeting prefers HAVE_TX_SET + /// claimants over SCP relayers over blind asks; if nothing can be + /// dispatched now (duplicate, grace period, or no peers) the fetch stays + /// pending and the housekeeping task drives it to completion. async fn fetch_txset(&mut self, hash: [u8; 32], slot: u32) { - // Check if we're already fetching this TxSet from a connected peer (dedup) - { - let pending = self.state.pending_txset_requests.read().await; - if let Some((pending_peer, _, _)) = pending.get(&hash) { - // Check if that peer is still connected - let streams = self.state.peer_streams.read().await; - if streams.contains_key(pending_peer) { - debug!( - "TXSET_FETCH_SKIP: TxSet {:02x?}... already being fetched from {}, skipping duplicate", - &hash[..4], pending_peer - ); - return; - } - // Otherwise, peer disconnected - we'll re-request below - } - } - - // First check if we know which peer has this TX set (from SCP message) - let known_source = { - let sources = self.state.txset_sources.read().await; - sources.peek(&hash).cloned() + let now = Instant::now(); + let connected = connected_peers(&self.state).await; + let (created, ask) = { + let mut fetcher = self.state.txset_fetcher.write().await; + let created = fetcher.start_fetch(hash, slot, now); + let ask = fetcher.dispatch_one(hash, now, &connected); + (created, ask) }; - - let peer = if let Some(source_peer) = known_source { - // Verify this peer is still connected - let streams = self.state.peer_streams.read().await; - if streams.contains_key(&source_peer) { + if !created { + debug!( + "TXSET_FETCH_SKIP: TxSet {:02x?}... already being fetched", + &hash[..4] + ); + } + match ask { + Some(ask) => dispatch_txset_asks(&self.state, vec![ask]).await, + None if created => { info!( - "TXSET_FETCH: Fetching TX set {:02x?}... from known source {}", - &hash[..4], - source_peer + "TXSET_FETCH_PARKED: TxSet {:02x?}... waiting (grace period or no peers)", + &hash[..4] ); - source_peer - } else { - // Source peer disconnected, fall back to any peer - match streams.keys().next().cloned() { - Some(p) => { - info!("TXSET_FETCH: Fetching TX set {:02x?}... from fallback peer {} (source {} disconnected)", - &hash[..4], p, source_peer); - p - } - None => { - warn!( - "TXSET_FETCH_FAIL: No peers to fetch TX set {:02x?}... from", - &hash[..4] - ); - return; - } - } - } - } else { - // No known source, pick any connected peer - let streams = self.state.peer_streams.read().await; - match streams.keys().next().cloned() { - Some(p) => { - info!( - "TXSET_FETCH: Fetching TX set {:02x?}... from random peer {} (no known source)", - &hash[..4], - p - ); - p - } - None => { - warn!( - "TXSET_FETCH_FAIL: No peers to fetch TX set {:02x?}... from", - &hash[..4] - ); - return; - } } - }; - - // Record this pending request with timestamp for latency tracking - self.state - .pending_txset_requests - .write() - .await - .insert(hash, (peer.clone(), Instant::now(), slot)); + None => {} + } + } - let request = crate::xdr::frame_get_tx_set(hash); + /// Send HAVE_TX_SET for `hash` to one peer. The write happens in its own + /// task: a claim is 36 bytes but the target's txset stream mutex may be + /// held by a multi-MB tx set write, and the event loop must never wait on + /// that. + async fn send_have_txset_to_peer(&mut self, hash: [u8; 32], to: PeerId) { + spawn_have_txset_send(&self.state, hash, to); + } - match send_to_peer_stream(&self.state, peer.clone(), StreamType::TxSet, &request).await { - Ok(_) => info!( - "TXSET_FETCH_SENT: Sent request for TxSet {:02x?}... to {}", - &hash[..4], - peer - ), - Err(e) => { - warn!( - "TXSET_FETCH_FAIL: Failed to send TxSet request {:02x?}... to {}: {}", - &hash[..4], - peer, - e - ); - self.state - .pending_txset_requests - .write() - .await - .remove(&hash); - } + /// Broadcast HAVE_TX_SET for `hash` to all connected peers, one send task + /// per peer (see `send_have_txset_to_peer` for why). + async fn announce_txset(&mut self, hash: [u8; 32]) { + let peers = connected_peers(&self.state).await; + if peers.is_empty() { + return; + } + info!( + "TXSET_ANNOUNCE: Announcing TX set {:02x?}... to {} peers", + &hash[..4], + peers.len() + ); + for peer in peers { + spawn_have_txset_send(&self.state, hash, peer); } } @@ -1902,6 +1943,53 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc< .metrics .byte_read .fetch_add(data.len() as u64, Ordering::Relaxed); + + // HAVE_TX_SET is codec'd manually (the pinned + // stellar-xdr crate predates it), so it must be tried + // before the typed decode below, which would reject + // its discriminant. + if let Some(claimed) = crate::xdr::parse_have_tx_set(&data) { + state + .metrics + .recv_have_txset + .fetch_add(1, Ordering::Relaxed); + if !admit_have_tx_set(&state, &peer_id).await { + state.metrics.claim_dropped.fetch_add(1, Ordering::Relaxed); + debug!( + "CLAIM_DROPPED: HAVE_TX_SET {:02x?}... from {} over budget", + &claimed[..4], + peer_id + ); + continue; + } + debug!( + "CLAIM_RECV: Peer {} claims TX set {:02x?}...", + peer_id, + &claimed[..4] + ); + let actionable = state + .txset_fetcher + .write() + .await + .record_claim(claimed, peer_id); + if actionable { + // A fetch is pending with no ask outstanding + // (grace wait, parked, or between retries): + // act on the claim immediately. + let now = Instant::now(); + let connected = connected_peers(&state).await; + let ask = state + .txset_fetcher + .write() + .await + .dispatch_one(claimed, now, &connected); + if let Some(ask) = ask { + dispatch_txset_asks(&state, vec![ask]).await; + } + } + continue; + } + let message = match crate::xdr::parse_stellar_message(&data) { Ok(message) => message, Err(e) => { @@ -1940,23 +2028,21 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc< let txset_data = data[4..].to_vec(); let hash = crate::xdr::sha256_hash(&txset_data); - // Clear pending request flag and measure fetch latency + // Complete the pending fetch (from whichever + // peer answered) and measure fetch latency. let slot = { - let mut pending = state.pending_txset_requests.write().await; - if let Some((_, request_time, slot)) = pending.remove(&hash) { - let fetch_us = request_time.elapsed().as_micros() as u64; - state - .metrics - .fetch_txset_sum_us - .fetch_add(fetch_us, Ordering::Relaxed); + let mut fetcher = state.txset_fetcher.write().await; + fetcher.complete(&hash, Instant::now()).map(|done| { + state.metrics.fetch_txset_sum_us.fetch_add( + done.elapsed.as_micros() as u64, + Ordering::Relaxed, + ); state .metrics .fetch_txset_count .fetch_add(1, Ordering::Relaxed); - Some(slot) - } else { - None - } + done.slot + }) }; info!( @@ -1999,19 +2085,170 @@ async fn handle_inbound_txset_streams(mut incoming: IncomingStreams, state: Arc< } } +/// Snapshot of currently connected peers (those with open streams). +async fn connected_peers(state: &Arc) -> Vec { + state.peer_streams.read().await.keys().cloned().collect() +} + +/// Consume one unit of `peer`'s HAVE_TX_SET admission budget. Returns false +/// (drop the message) once the per-window cap is hit; the budget map is +/// cleared every HAVE_TX_SET_PERIOD by the housekeeping task. +async fn admit_have_tx_set(state: &Arc, peer: &PeerId) -> bool { + let mut admitted = state.have_txset_admitted.write().await; + let count = admitted.entry(*peer).or_insert(0); + if *count >= HAVE_TX_SET_MAX_PER_PERIOD { + return false; + } + *count += 1; + true +} + +/// Send one HAVE_TX_SET to one peer in a dedicated task, so no caller (event +/// loop, reader task, housekeeping) ever waits on the target's txset stream +/// mutex, which may be held by a multi-MB tx set write. +fn spawn_have_txset_send(state: &Arc, hash: [u8; 32], to: PeerId) { + let state = Arc::clone(state); + tokio::spawn(async move { + let msg = crate::xdr::frame_have_tx_set(hash); + match send_to_peer_stream(&state, to, StreamType::TxSet, &msg).await { + Ok(_) => { + state + .metrics + .send_have_txset + .fetch_add(1, Ordering::Relaxed); + state.metrics.message_write.fetch_add(1, Ordering::Relaxed); + state + .metrics + .byte_write + .fetch_add(msg.len() as u64, Ordering::Relaxed); + debug!("CLAIM_SENT: HAVE_TX_SET {:02x?}... to {}", &hash[..4], to); + } + Err(e) => { + debug!( + "CLAIM_SEND_FAIL: HAVE_TX_SET {:02x?}... to {}: {}", + &hash[..4], + to, + e + ); + } + } + }); +} + +/// Send GET_TX_SET for each dispatched ask and record the outcome with the +/// fetcher (wire timestamp on success, retarget-on-next-pass on failure). +/// Also records the claim/grace metrics carried on the ask. Each send runs in +/// its own task so a busy or backpressured peer stream never stalls the +/// caller (`sent_at` was stamped at dispatch, so the housekeeping loop won't +/// re-dispatch the hash while the send is in flight; the peer's response +/// window restarts at `mark_sent` once the request is actually on the wire). +async fn dispatch_txset_asks(state: &Arc, asks: Vec) { + for ask in asks { + let metrics = &state.metrics; + if ask.tier == AskTier::Claimant { + metrics.claim_ask.fetch_add(1, Ordering::Relaxed); + } + if let Some(grace) = &ask.grace_outcome { + metrics + .claim_grace_wait_sum_us + .fetch_add(grace.waited.as_micros() as u64, Ordering::Relaxed); + metrics + .claim_grace_wait_count + .fetch_add(1, Ordering::Relaxed); + if grace.satisfied { + metrics + .claim_grace_satisfied + .fetch_add(1, Ordering::Relaxed); + } else { + metrics.claim_grace_expired.fetch_add(1, Ordering::Relaxed); + } + } + + let state = Arc::clone(state); + tokio::spawn(async move { + let request = crate::xdr::frame_get_tx_set(ask.hash); + match send_to_peer_stream(&state, ask.peer, StreamType::TxSet, &request).await { + Ok(_) => { + state + .txset_fetcher + .write() + .await + .mark_sent(&ask.hash, Instant::now()); + info!( + "TXSET_FETCH_SENT: Sent request for TxSet {:02x?}... to {} (tier {:?})", + &ask.hash[..4], + ask.peer, + ask.tier + ); + } + Err(e) => { + state + .txset_fetcher + .write() + .await + .mark_send_failed(&ask.hash); + warn!( + "TXSET_FETCH_FAIL: Failed to send TxSet request {:02x?}... to {}: {} (will retarget)", + &ask.hash[..4], + ask.peer, + e + ); + } + } + }); + } +} + /// INV/GETDATA housekeeping task. /// /// Periodically: /// 1. Flushes INV batches that have timed out (100ms) /// 2. Checks GETDATA timeouts and retries to other peers +/// 3. Drives pending TX set fetches (timeouts, retargeting, grace expiry) +/// 4. Resets the per-peer HAVE_TX_SET admission budgets every 5s async fn inv_getdata_housekeeping_task(state: Arc) { // Run every 50ms (half the batch timeout for responsiveness) let mut interval = tokio::time::interval(Duration::from_millis(50)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut last_admission_reset = Instant::now(); + loop { interval.tick().await; + // 0a. Drive pending TX set fetches: time out stale asks (dropping + // wrong claims), dispatch parked/retargeted fetches, expire ancient + // ones. Cheap when nothing is pending. + { + let now = Instant::now(); + let connected = connected_peers(&state).await; + let result = { + let mut fetcher = state.txset_fetcher.write().await; + fetcher.tick(now, &connected) + }; + if !result.expired.is_empty() { + state + .metrics + .fetch_txset_abandoned + .fetch_add(result.expired.len() as u64, Ordering::Relaxed); + for hash in &result.expired { + warn!( + "TXSET_FETCH_ABANDONED: Gave up on TxSet {:02x?}... (age backstop)", + &hash[..4] + ); + } + } + if !result.asks.is_empty() { + dispatch_txset_asks(&state, result.asks).await; + } + } + + // 0b. Reset per-peer HAVE_TX_SET admission budgets once per window. + if last_admission_reset.elapsed() >= HAVE_TX_SET_PERIOD { + state.have_txset_admitted.write().await.clear(); + last_admission_reset = Instant::now(); + } + // 1. Flush expired INV batches let expired_peers = { let batcher = state.inv_batcher.read().await; @@ -5096,3 +5333,207 @@ async fn test_20_node_mesh_with_dedup() { let _ = tokio::time::timeout(Duration::from_secs(2), task).await; } } + +// --- HAVE_TX_SET integration tests (stellar-core PR #5379) ----------------- + +/// Dial `from` -> `to` and wait until both ends see the peer's streams. +#[cfg(test)] +async fn test_connect(from: &OverlayHandle, to_port: u16) { + let addr: Multiaddr = format!("/ip4/127.0.0.1/udp/{}/quic-v1", to_port) + .parse() + .unwrap(); + from.dial(addr).await; + tokio::time::sleep(Duration::from_millis(500)).await; +} + +/// HAVE_TX_SET admission cap: per-peer budget, drops beyond it, isolation +/// between peers, and the periodic reset. Exercised directly against +/// SharedState (no wire), mirroring the C++ "HAVE_TX_SET admission cap" test. +#[tokio::test] +async fn test_have_tx_set_admission_cap() { + let keypair = Keypair::generate_ed25519(); + let (_handle, _events, _tx_events, overlay) = + create_overlay(keypair, Arc::new(OverlayMetrics::new())).unwrap(); + let state = overlay.state.clone(); + + let peer1 = PeerId::random(); + let peer2 = PeerId::random(); + + // Exactly the budget is admitted; the rest is dropped. + for _ in 0..HAVE_TX_SET_MAX_PER_PERIOD { + assert!(admit_have_tx_set(&state, &peer1).await); + } + for _ in 0..8 { + assert!(!admit_have_tx_set(&state, &peer1).await); + } + + // The budget is per peer: another peer is unaffected. + assert!(admit_have_tx_set(&state, &peer2).await); + + // The periodic reset restores the budget. + state.have_txset_admitted.write().await.clear(); + assert!(admit_have_tx_set(&state, &peer1).await); + + // Disconnect cleanup drops the peer's entry entirely. + state.have_txset_admitted.write().await.remove(&peer2); + assert_eq!( + state + .have_txset_admitted + .read() + .await + .get(&peer2) + .copied() + .unwrap_or(0), + 0 + ); +} + +/// An announced HAVE_TX_SET steers a later fetch to the announcer: node C is +/// connected to A and B; only A announces possession, and C's fetch must go +/// to A (claim tier), not blind-ask B. +#[tokio::test] +async fn test_announce_claim_steers_fetch() { + let key_a = Keypair::generate_ed25519(); + let key_b = Keypair::generate_ed25519(); + let key_c = Keypair::generate_ed25519(); + let peer_a = PeerId::from_public_key(&key_a.public()); + + let metrics_c = Arc::new(OverlayMetrics::new()); + let (handle_a, mut events_a, _txa, overlay_a) = + create_overlay(key_a, Arc::new(OverlayMetrics::new())).unwrap(); + let (_handle_b, mut events_b, _txb, overlay_b) = + create_overlay(key_b, Arc::new(OverlayMetrics::new())).unwrap(); + let (handle_c, mut events_c, _txc, overlay_c) = + create_overlay(key_c, Arc::clone(&metrics_c)).unwrap(); + + tokio::spawn(async move { overlay_a.run("127.0.0.1", 25101).await }); + tokio::spawn(async move { overlay_b.run("127.0.0.1", 25102).await }); + tokio::spawn(async move { overlay_c.run("127.0.0.1", 25103).await }); + tokio::time::sleep(Duration::from_millis(200)).await; + + test_connect(&handle_c, 25101).await; // C -> A + test_connect(&handle_c, 25102).await; // C -> B + + let (hash, data) = test_txset_xdr(0x51); + + // A announces possession; give the claim time to arrive at C. + handle_a.announce_txset(hash).await; + tokio::time::sleep(Duration::from_millis(400)).await; + + // C fetches: the ask must target A. + handle_c.fetch_txset(hash, 7).await; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let mut requester: Option = None; + while tokio::time::Instant::now() < deadline && requester.is_none() { + tokio::select! { + Some(event) = events_a.recv() => { + if let OverlayEvent::TxSetRequested { hash: h, from } = event { + assert_eq!(h, hash); + requester = Some(from); + } + } + _ = tokio::time::sleep(Duration::from_millis(10)) => {} + } + } + let requester = requester.expect("The fetch must target the announcing peer"); + assert_eq!( + metrics_c.claim_ask.load(Ordering::Relaxed), + 1, + "The ask must be recorded as claim-tier" + ); + + // B must not have been asked (the ask went straight to the claimant). + let mut asked_b = false; + let deadline = tokio::time::Instant::now() + Duration::from_millis(300); + while tokio::time::Instant::now() < deadline { + tokio::select! { + Some(event) = events_b.recv() => { + if matches!(event, OverlayEvent::TxSetRequested { .. }) { + asked_b = true; + } + } + _ = tokio::time::sleep(Duration::from_millis(10)) => {} + } + } + assert!(!asked_b, "No blind ask should reach the non-claimant"); + + // A serves the set; C completes the fetch (stream still healthy after + // carrying HAVE_TX_SET traffic). + let _ = peer_a; + handle_a.send_txset(hash, data, requester).await; + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let mut got_set = false; + while tokio::time::Instant::now() < deadline && !got_set { + tokio::select! { + Some(event) = events_c.recv() => { + if let OverlayEvent::TxSetReceived { hash: h, slot, .. } = event { + assert_eq!(h, hash); + assert_eq!(slot, Some(7), "completion must carry the requested slot"); + got_set = true; + } + } + _ = tokio::time::sleep(Duration::from_millis(10)) => {} + } + } + assert!(got_set, "The fetch must complete with the tx set"); +} + +/// With no claims at all, a fetch blind-asks one peer, and when that peer +/// stays silent the housekeeping retry retargets to the other peer within the +/// ask timeout. End-to-end regression for the pre-existing "fetch stuck on a +/// connected-but-empty peer" gap. +#[tokio::test] +async fn test_fetch_retries_to_second_peer() { + let key_a = Keypair::generate_ed25519(); + let key_b = Keypair::generate_ed25519(); + let key_c = Keypair::generate_ed25519(); + + let (_handle_a, mut events_a, _txa, overlay_a) = + create_overlay(key_a, Arc::new(OverlayMetrics::new())).unwrap(); + let (_handle_b, mut events_b, _txb, overlay_b) = + create_overlay(key_b, Arc::new(OverlayMetrics::new())).unwrap(); + let (handle_c, _events_c, _txc, overlay_c) = + create_overlay(key_c, Arc::new(OverlayMetrics::new())).unwrap(); + + tokio::spawn(async move { overlay_a.run("127.0.0.1", 25201).await }); + tokio::spawn(async move { overlay_b.run("127.0.0.1", 25202).await }); + tokio::spawn(async move { overlay_c.run("127.0.0.1", 25203).await }); + tokio::time::sleep(Duration::from_millis(200)).await; + + test_connect(&handle_c, 25201).await; + test_connect(&handle_c, 25202).await; + + let (hash, _) = test_txset_xdr(0x61); + handle_c.fetch_txset(hash, 3).await; + + // Neither peer serves the set. Within one ask timeout + housekeeping + // slack, BOTH peers must have been asked (first blind ask + retarget). + // Generous slack: the full test suite runs many network tests in + // parallel and can starve the housekeeping tick. + let deadline = + tokio::time::Instant::now() + crate::flood::TXSET_ASK_TIMEOUT + Duration::from_secs(6); + let (mut asked_a, mut asked_b) = (false, false); + while tokio::time::Instant::now() < deadline && !(asked_a && asked_b) { + tokio::select! { + Some(event) = events_a.recv() => { + if matches!(event, OverlayEvent::TxSetRequested { .. }) { + asked_a = true; + } + } + Some(event) = events_b.recv() => { + if matches!(event, OverlayEvent::TxSetRequested { .. }) { + asked_b = true; + } + } + _ = tokio::time::sleep(Duration::from_millis(20)) => {} + } + } + assert!( + asked_a && asked_b, + "Fetch must retarget to the second peer after the first stays silent \ + (asked_a={}, asked_b={})", + asked_a, + asked_b + ); +} diff --git a/overlay/src/main.rs b/overlay/src/main.rs index e4386246e..83ce1ee51 100644 --- a/overlay/src/main.rs +++ b/overlay/src/main.rs @@ -388,6 +388,28 @@ fn cache_tx_set_xdr( }); } +/// Tx set hashes referenced by `envelopes` (raw ScpEnvelope XDR) that are +/// held in `cache`, deduplicated in first-reference order. These are the +/// HAVE_TX_SET claims to send ahead of SCP state (PR #5379: with parallel +/// tx set downloading, relaying an envelope no longer implies possessing the +/// sets it references, so possession is claimed explicitly). Unheld sets and +/// undecodable envelopes contribute nothing. +fn claims_for_envelopes(envelopes: &[Vec], cache: &TxSetCache) -> Vec { + use stellar_xdr::curr::{Limits, ReadXdr, ScpEnvelope}; + let mut claims = Vec::new(); + for env_bytes in envelopes { + let Ok(envelope) = ScpEnvelope::from_xdr(env_bytes.as_slice(), Limits::none()) else { + continue; + }; + for hash in xdr::extract_txset_hashes_from_envelope(&envelope) { + if cache.get(&hash).is_some() && !claims.contains(&hash) { + claims.push(hash); + } + } + } + claims +} + /// Application state struct App { core_ipc: CoreIpc, @@ -418,10 +440,16 @@ struct App { /// PeerId → configured hostname, so targeted reconnect can re-resolve DNS /// after a pod restart changes the peer's IP address. peer_hostnames: Arc>>, + /// TX set hashes already announced via HAVE_TX_SET (announce-once). + announced_txsets: lru::LruCache, /// Shared metrics counters for the overlay metrics: Arc, } +/// Capacity of the announce-once dedup cache. Matches the C++ +/// `PendingEnvelopes::mAnnouncedTxSets` sizing (TXSET_CACHE_SIZE-scale). +const ANNOUNCED_TXSETS_CACHE_SIZE: usize = 1000; + /// Peer addresses configured via SetPeerConfig, used for reconnection. struct ConfiguredPeers { /// All peer address strings (known + preferred) @@ -498,6 +526,9 @@ impl App { })), known_peers: Arc::new(RwLock::new(HashMap::new())), peer_hostnames: Arc::new(RwLock::new(HashMap::new())), + announced_txsets: lru::LruCache::new( + std::num::NonZeroUsize::new(ANNOUNCED_TXSETS_CACHE_SIZE).unwrap(), + ), metrics, }) } @@ -654,6 +685,24 @@ impl App { info!("Overlay shutting down"); } + /// Announce possession of a newly cached TX set to all peers via + /// HAVE_TX_SET, at most once per hash (mirrors the C++ + /// `PendingEnvelopes::maybeAnnounceHaveTxSet`). Sets stamped slot 0 — the + /// sentinel for data restored from the DB rather than obtained on a live + /// consensus path — are not announced. Returns whether an announce was + /// issued (observable in tests). + fn maybe_announce_txset(&mut self, hash: Hash256, slot: u32) -> bool { + if slot == 0 || self.announced_txsets.contains(&hash) { + return false; + } + self.announced_txsets.put(hash, ()); + let handle = self.libp2p_handle.clone(); + tokio::spawn(async move { + handle.announce_txset(hash).await; + }); + true + } + /// Handle an event from the libp2p QUIC overlay (SCP + TX) async fn handle_libp2p_event(&mut self, event: LibP2pOverlayEvent) { match event { @@ -748,12 +797,12 @@ impl App { // This ensures the TxSet is available when SCP processing resumes // Stamp with the slot the set was requested for so eviction is // exact; for an unsolicited set fall back to the next slot. - cache_tx_set_xdr( - &mut self.tx_set_cache, - slot.unwrap_or(self.current_ledger_seq + 1), - hash, - data.clone(), - ); + let stamp_slot = slot.unwrap_or(self.current_ledger_seq + 1); + cache_tx_set_xdr(&mut self.tx_set_cache, stamp_slot, hash, data.clone()); + + // We now possess the set: tell peers (HAVE_TX_SET), so their + // fetches can target us instead of blind-asking. + self.maybe_announce_txset(hash, stamp_slot); // Always push TX set to Core (Core handles dedup) info!( @@ -1072,6 +1121,9 @@ impl App { ); cache_tx_set_xdr(&mut self.tx_set_cache, slot, hash, tx_set_xdr.to_vec()); + + // Announce the locally-built set to peers (HAVE_TX_SET). + self.maybe_announce_txset(hash, slot); } MessageType::SubmitTx => { @@ -1130,7 +1182,10 @@ impl App { } MessageType::LedgerClosed => { - // Parse payload: [ledgerSeq:4][ledgerHash:32] + // Parse payload: [ledgerSeq:4][ledgerHash:32][flags:1] + // The flags byte (bit 0 = ledger protocol admits empty-tx-set + // values, PR #5379 claim-grace gating) is a newer extension: + // a legacy 36-byte payload is accepted and leaves the flag off. if msg.payload.len() >= 4 { let ledger_seq = u32::from_le_bytes(msg.payload[0..4].try_into().unwrap()); info!("Ledger {} closed", ledger_seq); @@ -1138,9 +1193,24 @@ impl App { // Update current ledger self.current_ledger_seq = ledger_seq; + if let Some(flags) = msg.payload.get(36) { + let possible = flags & 0x1 != 0; + let handle = self.libp2p_handle.clone(); + tokio::spawn(async move { + handle.set_empty_tx_sets_possible(possible).await; + }); + } + // Evict old TX sets from cache - self.tx_set_cache - .evict_before(ledger_seq.saturating_sub(12)); + let horizon = ledger_seq.saturating_sub(12); + self.tx_set_cache.evict_before(horizon); + + // Drop pending tx set fetches for slots past the same + // horizon (fetches retry until here, not forever). + let handle = self.libp2p_handle.clone(); + tokio::spawn(async move { + handle.purge_txset_fetches_below(horizon).await; + }); } } @@ -1222,36 +1292,54 @@ impl App { num_envelopes, peer_id, request_id ); - // Parse and forward each envelope to the requesting peer - let handle = self.libp2p_handle.clone(); - let payload = msg.payload.clone(); - tokio::spawn(async move { - let mut offset = 12; // Skip request_id (8) + count (4) - for _ in 0..num_envelopes { - if offset + 4 > payload.len() { - warn!("ScpStateResponse truncated at envelope length"); - break; - } - let env_len = - u32::from_le_bytes(payload[offset..offset + 4].try_into().unwrap()) - as usize; - offset += 4; + // Parse the envelopes up front (also needed for the claim + // computation against the local tx set cache). + let mut envelopes: Vec> = Vec::with_capacity(num_envelopes); + let payload = &msg.payload; + let mut offset = 12; // Skip request_id (8) + count (4) + for _ in 0..num_envelopes { + if offset + 4 > payload.len() { + warn!("ScpStateResponse truncated at envelope length"); + break; + } + let env_len = + u32::from_le_bytes(payload[offset..offset + 4].try_into().unwrap()) + as usize; + offset += 4; + + if offset + env_len > payload.len() { + warn!("ScpStateResponse truncated at envelope data"); + break; + } + envelopes.push(payload[offset..offset + env_len].to_vec()); + offset += env_len; + } - if offset + env_len > payload.len() { - warn!("ScpStateResponse truncated at envelope data"); - break; - } - let envelope = &payload[offset..offset + env_len]; - offset += env_len; + // Claims first: tell the catching-up peer which referenced + // tx sets we hold, so its fetches target us instead of + // blind-asking (PR #5379). Claims travel on the txset stream + // and envelopes on the SCP stream — cross-stream ordering is + // not guaranteed, but a late claim still redirects the + // peer's fetch via its claim/retry logic. + let claims = claims_for_envelopes(&envelopes, &self.tx_set_cache); + let handle = self.libp2p_handle.clone(); + tokio::spawn(async move { + for hash in &claims { + handle.send_have_txset(*hash, peer_id).await; + } + let count = envelopes.len(); + for envelope in envelopes { // Send envelope to requesting peer over SCP stream - if let Err(e) = handle.send_scp_to_peer(peer_id.clone(), envelope).await { + if let Err(e) = handle.send_scp_to_peer(peer_id, &envelope).await { warn!("Failed to send SCP envelope to {}: {:?}", peer_id, e); } } info!( - "Finished forwarding {} SCP envelopes to {}", - num_envelopes, peer_id + "Finished forwarding {} SCP envelopes ({} claims) to {}", + count, + claims.len(), + peer_id ); }); } @@ -1938,6 +2026,9 @@ mod tests { })), known_peers: Arc::new(RwLock::new(HashMap::new())), peer_hostnames: Arc::new(RwLock::new(HashMap::new())), + announced_txsets: lru::LruCache::new( + std::num::NonZeroUsize::new(ANNOUNCED_TXSETS_CACHE_SIZE).unwrap(), + ), metrics, }; (app, core_side) @@ -2042,6 +2133,168 @@ mod tests { assert_eq!(&resp.payload[32..], &xdr_bytes[..]); } + /// LedgerClosed accepts both the legacy 36-byte payload and the extended + /// one carrying the flags byte (empty-tx-sets-possible, PR #5379). + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_ledger_closed_payload_flag_extension() { + let (mut app, _core) = test_app(); + + // Legacy payload (36 bytes): handled, seq updated. + assert!( + app.handle_core_message(Message::new( + MessageType::LedgerClosed, + ledger_closed_payload(41) + )) + .await + ); + assert_eq!(app.current_ledger_seq, 41); + + // Extended payload (37 bytes, flag set): handled, seq updated. + let mut extended = ledger_closed_payload(42); + extended.push(0x1); + assert!( + app.handle_core_message(Message::new(MessageType::LedgerClosed, extended)) + .await + ); + assert_eq!(app.current_ledger_seq, 42); + } + + // --- claims accompanying SCP state (stellar-core PR #5379) --- + + /// A nomination envelope whose votes reference the given tx set hashes. + fn nominate_envelope_xdr(tx_set_hashes: &[[u8; 32]]) -> Vec { + use stellar_xdr::curr::{ + Hash, ScpEnvelope, ScpNomination, ScpStatementPledges, StellarValue, StellarValueExt, + TimePoint, Value, VecM, + }; + + let votes: Vec = tx_set_hashes + .iter() + .map(|h| { + let sv = StellarValue { + tx_set_hash: Hash(*h), + close_time: TimePoint(1), + upgrades: VecM::default(), + ext: StellarValueExt::Basic, + }; + Value::try_from(sv.to_xdr(Limits::none()).unwrap()).unwrap() + }) + .collect(); + + let mut envelope = ScpEnvelope::default(); + envelope.statement.pledges = ScpStatementPledges::Nominate(ScpNomination { + quorum_set_hash: Hash([0; 32]), + votes: VecM::try_from(votes).unwrap(), + accepted: VecM::default(), + }); + envelope.to_xdr(Limits::none()).unwrap() + } + + /// Mirrors the C++ "HAVE_TX_SET claims accompany SCP state" test: only + /// held sets are claimed, each at most once per batch; unheld sets and + /// undecodable envelopes contribute nothing. + #[test] + fn test_claims_for_envelopes() { + let mut cache = TxSetCache::new(10); + let (held, held_xdr) = test_txset_xdr(31); + let unheld = [0x99u8; 32]; + cache.insert(CachedTxSet { + hash: held, + xdr: held_xdr, + ledger_seq: 5, + }); + + let env1 = nominate_envelope_xdr(&[held, unheld]); + // A second envelope referencing the held set again: no duplicate claim. + let env2 = nominate_envelope_xdr(&[held]); + let garbage = vec![0xffu8; 7]; + + let claims = claims_for_envelopes(&[env1, env2, garbage], &cache); + assert_eq!(claims, vec![held]); + + // No envelopes, or none referencing held sets: no claims. + assert!(claims_for_envelopes(&[], &cache).is_empty()); + let env3 = nominate_envelope_xdr(&[unheld]); + assert!(claims_for_envelopes(&[env3], &cache).is_empty()); + } + + // --- HAVE_TX_SET announce (stellar-core PR #5379) --- + + /// Announce-once semantics: the first acquisition of a set announces it, + /// re-acquisitions do not, and slot-0 (DB-restored sentinel) sets are + /// never announced. Mirrors the C++ "HAVE_TX_SET announce" test. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_announce_once_and_slot0_suppressed() { + let (mut app, _core) = test_app(); + let (hash, _) = test_txset_xdr(21); + + assert!(app.maybe_announce_txset(hash, 10), "first announce fires"); + assert!( + !app.maybe_announce_txset(hash, 10), + "re-announce is suppressed" + ); + assert!( + !app.maybe_announce_txset(hash, 11), + "suppression is per hash, not per slot" + ); + + let (restored_hash, _) = test_txset_xdr(22); + assert!( + !app.maybe_announce_txset(restored_hash, 0), + "slot-0 (restored) sets are never announced" + ); + assert!( + !app.announced_txsets.contains(&restored_hash), + "a suppressed announce must not poison the dedup cache" + ); + assert!( + app.maybe_announce_txset(restored_hash, 12), + "the same set announced later via a live path still fires" + ); + } + + /// Core caching a locally-built set (CacheTxSet) announces it. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_cache_tx_set_announces() { + let (mut app, _core) = test_app(); + let (hash, xdr_bytes) = test_txset_xdr(23); + + let mut payload = request_tx_set_payload(&hash, 100); + payload.extend_from_slice(&xdr_bytes); + assert!( + app.handle_core_message(Message::new(MessageType::CacheTxSet, payload)) + .await + ); + assert!( + app.announced_txsets.contains(&hash), + "caching a locally-built set must announce it" + ); + } + + /// A set fetched from a peer (TxSetReceived) is announced too — we can + /// now serve it, and other nodes' fetches should know that. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_txset_received_announces() { + let (mut app, mut core) = test_app(); + let (hash, xdr_bytes) = test_txset_xdr(24); + + app.handle_libp2p_event(LibP2pOverlayEvent::TxSetReceived { + hash, + data: xdr_bytes, + from: PeerId::random(), + slot: Some(50), + }) + .await; + // Drain the TxSetAvailable push to core. + core.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + let _ = MessageCodec::read(&mut core).unwrap(); + + assert!( + app.announced_txsets.contains(&hash), + "a set fetched from the network must be announced" + ); + } + /// Same property for sets fetched from peers: a TxSetReceived event is /// cached under the slot the set was requested for, so it survives /// eviction until that slot is actually past. diff --git a/overlay/src/metrics.rs b/overlay/src/metrics.rs index 9371beae3..ee63dd3bd 100644 --- a/overlay/src/metrics.rs +++ b/overlay/src/metrics.rs @@ -97,6 +97,25 @@ pub struct OverlayMetrics { /// overlay.send.txset — TX set messages sent pub send_txset: AtomicU64, + // HAVE_TX_SET / tx set claim metrics (stellar-core PR #5379) + /// overlay.send.have-tx-set — HAVE_TX_SET messages sent + pub send_have_txset: AtomicU64, + /// overlay.recv.have-tx-set — HAVE_TX_SET messages received (pre-admission) + pub recv_have_txset: AtomicU64, + /// overlay.item-fetcher.claim-ask — fetch ask targeted a claimed holder + pub claim_ask: AtomicU64, + /// overlay.item-fetcher.claim-dropped — HAVE_TX_SET dropped at admission + pub claim_dropped: AtomicU64, + /// overlay.item-fetcher.claim-grace-wait — wait from fetch creation to first ask + pub claim_grace_wait_sum_us: AtomicU64, + pub claim_grace_wait_count: AtomicU64, + /// overlay.item-fetcher.claim-grace-satisfied — first ask targeted a claimant + pub claim_grace_satisfied: AtomicU64, + /// overlay.item-fetcher.claim-grace-expired — first ask fell back to relayer/blind + pub claim_grace_expired: AtomicU64, + /// overlay.fetch.txset-abandoned — tx set fetches dropped by the age backstop + pub fetch_txset_abandoned: AtomicU64, + // Receive timers (per message type, tracked as sum_us + count) /// overlay.recv.scp-message — time processing SCP messages pub recv_scp_sum_us: AtomicU64, @@ -152,6 +171,15 @@ impl Default for OverlayMetrics { send_scp_message: AtomicU64::new(0), send_transaction: AtomicU64::new(0), send_txset: AtomicU64::new(0), + send_have_txset: AtomicU64::new(0), + recv_have_txset: AtomicU64::new(0), + claim_ask: AtomicU64::new(0), + claim_dropped: AtomicU64::new(0), + claim_grace_wait_sum_us: AtomicU64::new(0), + claim_grace_wait_count: AtomicU64::new(0), + claim_grace_satisfied: AtomicU64::new(0), + claim_grace_expired: AtomicU64::new(0), + fetch_txset_abandoned: AtomicU64::new(0), recv_scp_sum_us: AtomicU64::new(0), recv_scp_count: AtomicU64::new(0), fetch_txset_sum_us: AtomicU64::new(0), @@ -213,6 +241,15 @@ impl OverlayMetrics { send_scp_message: self.send_scp_message.load(ORD), send_transaction: self.send_transaction.load(ORD), send_txset: self.send_txset.load(ORD), + send_have_txset: self.send_have_txset.load(ORD), + recv_have_txset: self.recv_have_txset.load(ORD), + claim_ask: self.claim_ask.load(ORD), + claim_dropped: self.claim_dropped.load(ORD), + claim_grace_wait_sum_us: self.claim_grace_wait_sum_us.load(ORD), + claim_grace_wait_count: self.claim_grace_wait_count.load(ORD), + claim_grace_satisfied: self.claim_grace_satisfied.load(ORD), + claim_grace_expired: self.claim_grace_expired.load(ORD), + fetch_txset_abandoned: self.fetch_txset_abandoned.load(ORD), recv_scp_sum_us: self.recv_scp_sum_us.load(ORD), recv_scp_count: self.recv_scp_count.load(ORD), fetch_txset_sum_us: self.fetch_txset_sum_us.load(ORD), @@ -282,6 +319,15 @@ pub struct MetricsSnapshot { pub send_scp_message: u64, pub send_transaction: u64, pub send_txset: u64, + pub send_have_txset: u64, + pub recv_have_txset: u64, + pub claim_ask: u64, + pub claim_dropped: u64, + pub claim_grace_wait_sum_us: u64, + pub claim_grace_wait_count: u64, + pub claim_grace_satisfied: u64, + pub claim_grace_expired: u64, + pub fetch_txset_abandoned: u64, pub recv_scp_sum_us: u64, pub recv_scp_count: u64, pub fetch_txset_sum_us: u64, diff --git a/overlay/src/xdr.rs b/overlay/src/xdr.rs index 763a443b5..465502ca5 100644 --- a/overlay/src/xdr.rs +++ b/overlay/src/xdr.rs @@ -94,6 +94,36 @@ pub(crate) fn frame_get_scp_state(ledger_seq: u32) -> Vec { frame(MessageType::GetScpState, &ledger_seq.to_be_bytes()) } +// --- HAVE_TX_SET (stellar-core PR #5379) ---------------------------------- +// +// `HAVE_TX_SET = 25` with body `struct HaveTxSet { Hash txSetHash; }` is not +// yet in the pinned `stellar-xdr` crate (26.0.0), so it is codec'd manually: +// 4-byte big-endian discriminant + 32-byte hash, exactly 36 bytes. The +// `have_tx_set_absent_from_crate` canary test below fails the moment the +// crate gains the arm, prompting a switch to the typed codec. + +/// The `MessageType::HAVE_TX_SET` XDR union discriminant. +pub(crate) const HAVE_TX_SET_DISCRIMINANT: i32 = 25; + +/// `StellarMessage::HaveTxSet(HaveTxSet { txSetHash })` framing. +pub(crate) fn frame_have_tx_set(hash: [u8; 32]) -> Vec { + let mut out = Vec::with_capacity(36); + out.extend_from_slice(&HAVE_TX_SET_DISCRIMINANT.to_be_bytes()); + out.extend_from_slice(&hash); + out +} + +/// Strict parse of a `HAVE_TX_SET` wire message: correct discriminant and +/// exact length, nothing else. Returns the claimed tx set hash. +pub(crate) fn parse_have_tx_set(data: &[u8]) -> Option<[u8; 32]> { + if data.len() != 36 || data[0..4] != HAVE_TX_SET_DISCRIMINANT.to_be_bytes() { + return None; + } + let mut hash = [0u8; 32]; + hash.copy_from_slice(&data[4..36]); + Some(hash) +} + /// Cheap content-hash check with no decode — used for tx sets built by our /// trusted local core, where we skip decoding but still guard against a /// hash/bytes mismatch that would make the set unfetchable network-wide. @@ -108,8 +138,9 @@ pub(crate) fn parse_stellar_message(bytes: &[u8]) -> Result Vec<[u8; 32]> { +/// stream reader reuse the single decode it already performed. Also used by +/// the app to send HAVE_TX_SET claims alongside SCP state (PR #5379). +pub fn extract_txset_hashes_from_envelope(envelope: &ScpEnvelope) -> Vec<[u8; 32]> { let mut hashes = Vec::new(); match &envelope.statement.pledges { ScpStatementPledges::Prepare(prepare) => { @@ -234,6 +265,48 @@ pub(crate) mod tests { assert_eq!(frame_get_scp_state(12345), expected); } + // --- HAVE_TX_SET manual codec ------------------------------------------- + + #[test] + fn have_tx_set_round_trips() { + let hash = [0x7c; 32]; + let framed = frame_have_tx_set(hash); + assert_eq!(framed.len(), 36); + assert_eq!(parse_have_tx_set(&framed), Some(hash)); + } + + #[test] + fn parse_have_tx_set_rejects_malformed() { + let hash = [0x7c; 32]; + let framed = frame_have_tx_set(hash); + + // Wrong discriminant + let mut wrong_type = framed.clone(); + wrong_type[3] = 6; // GET_TX_SET + assert_eq!(parse_have_tx_set(&wrong_type), None); + + // Truncated / oversized / empty + assert_eq!(parse_have_tx_set(&framed[..35]), None); + let mut long = framed.clone(); + long.push(0); + assert_eq!(parse_have_tx_set(&long), None); + assert_eq!(parse_have_tx_set(&[]), None); + } + + /// Canary: the pinned stellar-xdr crate does not know HAVE_TX_SET, which + /// is why the manual codec above exists (and why receivers must try + /// `parse_have_tx_set` before `parse_stellar_message`). When the crate + /// gains the arm this test fails — switch to the typed codec and add a + /// `frames_match_typed_have_tx_set` equivalence test like the other arms. + #[test] + fn have_tx_set_absent_from_crate() { + let framed = frame_have_tx_set([0x7c; 32]); + assert!( + parse_stellar_message(&framed).is_err(), + "stellar-xdr now decodes HAVE_TX_SET: retire the manual codec" + ); + } + // --- content-hash guard ------------------------------------------------- #[test] diff --git a/src/herder/HerderSCPDriver.cpp b/src/herder/HerderSCPDriver.cpp index 76db8d2d2..a2ac3a2ce 100644 --- a/src/herder/HerderSCPDriver.cpp +++ b/src/herder/HerderSCPDriver.cpp @@ -1319,6 +1319,14 @@ HerderSCPDriver::recordBallotBlockedOnTxSet(uint64_t slotIndex, timing.mBallotBlockedOnTxSetStart.end()) { timing.mBallotBlockedOnTxSetStart[value] = mApp.getClock().now(); + + if (StellarValue sv; + isParallelTxSetDownloadEnabled() && toStellarValue(value, sv)) + { + // Remember that `value` is stalled waiting for the tx set + // with hash `sv.txSetHash`. + mStallingByTxSet[sv.txSetHash].emplace_back(slotIndex, value); + } } } @@ -1337,6 +1345,27 @@ HerderSCPDriver::measureAndRecordBallotBlockedOnTxSet(uint64_t slotIndex, std::chrono::duration_cast( mApp.getClock().now() - valueIt->second); mSCPMetrics.mBallotBlockedOnTxSet.Update(elapsed); + + if (StellarValue sv; toStellarValue(value, sv)) + { + // This value is no longer stalled. Remove it from + // `mStallingByTxSet` + auto sIt = mStallingByTxSet.find(sv.txSetHash); + if (sIt != mStallingByTxSet.end()) + { + auto& vec = sIt->second; + vec.erase(std::remove_if(vec.begin(), vec.end(), + [&](auto const& p) { + return p.first == slotIndex && + p.second == value; + }), + vec.end()); + if (vec.empty()) + { + mStallingByTxSet.erase(sIt); + } + } + } return; } } @@ -1695,6 +1724,23 @@ HerderSCPDriver::purgeSlotsOutsideRange(std::optional minSlotIndex, // Clean up expired weak_ptrs from the pending tx set registries. purgeExpiredWeakPtrs(mPendingTxSetWrappers); purgeExpiredWeakPtrs(mPendingTxSetEnvelopeWrappers); + + // Drop stalled-ballot resume entries whose slots fall outside the retained + // range. + for (auto it = mStallingByTxSet.begin(); it != mStallingByTxSet.end();) + { + auto& stalling = it->second; + stalling.erase( + std::remove_if(stalling.begin(), stalling.end(), + [&](auto const& slotAndValue) { + auto const slot = slotAndValue.first; + return slot != slotToKeep && + ((minSlotIndex && slot < *minSlotIndex) || + (maxSlotIndex && slot > *maxSlotIndex)); + }), + stalling.end()); + it = stalling.empty() ? mStallingByTxSet.erase(it) : std::next(it); + } } void @@ -1728,6 +1774,34 @@ HerderSCPDriver::onTxSetReceived(Hash const& txSetHash, } mPendingTxSetEnvelopeWrappers.erase(envIt); } + + // Resume any slot that stalled waiting for this tx set + maybeResumeBalloting(txSetHash); +} + +void +HerderSCPDriver::maybeResumeBalloting(Hash const& txSetHash) +{ + if (!isParallelTxSetDownloadEnabled()) + { + return; + } + + auto it = mStallingByTxSet.find(txSetHash); + if (it == mStallingByTxSet.end()) + { + return; + } + + // Remove entry from `mStallingByTxSet` and iterate over stalling slots + // Remove prior to iterating because the `receivedTxSet` flow may itself + // modify `mStallingByTxSet`. + auto const stalling = std::move(it->second); + mStallingByTxSet.erase(it); + for (auto const& slotAndValue : stalling) + { + mSCP.receivedTxSet(slotAndValue.first, slotAndValue.second); + } } void diff --git a/src/herder/HerderSCPDriver.h b/src/herder/HerderSCPDriver.h index 81f9114e4..b0cb83c0d 100644 --- a/src/herder/HerderSCPDriver.h +++ b/src/herder/HerderSCPDriver.h @@ -175,6 +175,10 @@ class HerderSCPDriver : public SCPDriver // downloading). void onTxSetReceived(Hash const& txSetHash, TxSetXDRFrameConstPtr txSet); + // If balloting is stalled waiting for txSetHash, then resume balloting from + // the stall point. Otherwise, do nothing. + void maybeResumeBalloting(Hash const& txSetHash); + double getExternalizeLag(NodeID const& id) const; Json::Value getQsetLagInfo(bool summary, bool fullKeys); @@ -313,6 +317,11 @@ class HerderSCPDriver : public SCPDriver // * first prepare to externalize std::map mSCPExecutionTimes; + // Values stalled at the ballot commit gate waiting for a tx set. + // Mapping from to pairs of (, ). + UnorderedMap>> + mStallingByTxSet; + uint32_t mLedgerSeqNominating; ValueWrapperPtr mCurrentValue; diff --git a/src/overlay/OverlayIPC.cpp b/src/overlay/OverlayIPC.cpp index 9547d4a0e..3d68d644e 100644 --- a/src/overlay/OverlayIPC.cpp +++ b/src/overlay/OverlayIPC.cpp @@ -458,10 +458,15 @@ OverlayIPC::notifyLedgerClosed(uint32_t ledgerSeq, Hash const& ledgerHash) IPCMessage msg; msg.type = IPCMessageType::LEDGER_CLOSED; - // Payload: [ledgerSeq:4][ledgerHash:32] - msg.payload.resize(4 + 32); + // Payload: [ledgerSeq:4][ledgerHash:32][flags:1] + // flags bit 0: the ledger protocol admits empty-tx-set values, which + // enables the HAVE_TX_SET claim grace period on tx set fetches in the + // overlay (PR #5379). Always 0 until the CAP-0083 features are ported to + // this branch; the overlay also accepts the legacy 36-byte payload. + msg.payload.resize(4 + 32 + 1); std::memcpy(msg.payload.data(), &ledgerSeq, 4); std::memcpy(msg.payload.data() + 4, ledgerHash.data(), 32); + msg.payload[36] = 0; std::lock_guard lock(mSendMutex); mChannel->send(msg); diff --git a/src/overlay/OverlayMetrics.cpp b/src/overlay/OverlayMetrics.cpp index 7b4d3e055..be8f73696 100644 --- a/src/overlay/OverlayMetrics.cpp +++ b/src/overlay/OverlayMetrics.cpp @@ -57,6 +57,22 @@ OverlayMetrics::OverlayMetrics(Application& app) , mAuthenticatedPeersSize(app.getMetrics().NewCounter( {"overlay", "connection", "authenticated"})) , mFetchTxSetTimer(app.getMetrics().NewTimer({"overlay", "fetch", "txset"})) + , mSendHaveTxSetMeter(app.getMetrics().NewMeter( + {"overlay", "send", "have-tx-set"}, "message")) + , mRecvHaveTxSetMeter(app.getMetrics().NewMeter( + {"overlay", "recv", "have-tx-set"}, "message")) + , mItemFetcherClaimAsk(app.getMetrics().NewMeter( + {"overlay", "item-fetcher", "claim-ask"}, "item-fetcher")) + , mItemFetcherClaimDropped(app.getMetrics().NewMeter( + {"overlay", "item-fetcher", "claim-dropped"}, "item-fetcher")) + , mItemFetcherClaimGraceWait(app.getMetrics().NewTimer( + {"overlay", "item-fetcher", "claim-grace-wait"})) + , mItemFetcherClaimGraceSatisfied(app.getMetrics().NewMeter( + {"overlay", "item-fetcher", "claim-grace-satisfied"}, "item-fetcher")) + , mItemFetcherClaimGraceExpired(app.getMetrics().NewMeter( + {"overlay", "item-fetcher", "claim-grace-expired"}, "item-fetcher")) + , mAbandonedTxSetFetches(app.getMetrics().NewMeter( + {"overlay", "fetch", "txset-abandoned"}, "message")) { } } diff --git a/src/overlay/OverlayMetrics.h b/src/overlay/OverlayMetrics.h index 5fa27c9f1..aa285b4c0 100644 --- a/src/overlay/OverlayMetrics.h +++ b/src/overlay/OverlayMetrics.h @@ -68,5 +68,15 @@ struct OverlayMetrics // ── TxSet fetch latency ── medida::Timer& mFetchTxSetTimer; + + // ── HAVE_TX_SET / tx set claim metrics (PR #5379) ── + medida::Meter& mSendHaveTxSetMeter; + medida::Meter& mRecvHaveTxSetMeter; + medida::Meter& mItemFetcherClaimAsk; + medida::Meter& mItemFetcherClaimDropped; + medida::Timer& mItemFetcherClaimGraceWait; + medida::Meter& mItemFetcherClaimGraceSatisfied; + medida::Meter& mItemFetcherClaimGraceExpired; + medida::Meter& mAbandonedTxSetFetches; }; } diff --git a/src/overlay/RustOverlayManager.cpp b/src/overlay/RustOverlayManager.cpp index 94cde991f..4c6dfa194 100644 --- a/src/overlay/RustOverlayManager.cpp +++ b/src/overlay/RustOverlayManager.cpp @@ -325,6 +325,15 @@ RustOverlayManager::syncOverlayMetrics() markDelta(m.mSendTransactionMeter, "send_transaction"); markDelta(m.mSendTxSetMeter, "send_txset"); + // HAVE_TX_SET / tx set claim metrics (PR #5379) + markDelta(m.mSendHaveTxSetMeter, "send_have_txset"); + markDelta(m.mRecvHaveTxSetMeter, "recv_have_txset"); + markDelta(m.mItemFetcherClaimAsk, "claim_ask"); + markDelta(m.mItemFetcherClaimDropped, "claim_dropped"); + markDelta(m.mItemFetcherClaimGraceSatisfied, "claim_grace_satisfied"); + markDelta(m.mItemFetcherClaimGraceExpired, "claim_grace_expired"); + markDelta(m.mAbandonedTxSetFetches, "fetch_txset_abandoned"); + // Connection lifecycle — these aren't registered as medida meters on // the C++ side yet, so they'll just be tracked by the existing counters. // The inbound/outbound attempt/establish/drop are already covered @@ -377,6 +386,31 @@ RustOverlayManager::syncOverlayMetrics() mLastSyncedValues["flood_tx_batch_size_count"] = count; } + // ── Claim grace wait timer (PR #5379) ── + if (root.isMember("claim_grace_wait_sum_us") && + root.isMember("claim_grace_wait_count")) + { + auto sum = + static_cast(root["claim_grace_wait_sum_us"].asUInt64()); + auto count = + static_cast(root["claim_grace_wait_count"].asUInt64()); + auto lastSum = mLastSyncedValues["claim_grace_wait_sum_us"]; + auto lastCount = mLastSyncedValues["claim_grace_wait_count"]; + auto deltaSum = sum - lastSum; + auto deltaCount = count - lastCount; + if (deltaCount > 0 && deltaSum > 0) + { + auto avgUs = deltaSum / deltaCount; + for (int64_t i = 0; i < deltaCount; ++i) + { + m.mItemFetcherClaimGraceWait.Update( + std::chrono::microseconds{avgUs}); + } + } + mLastSyncedValues["claim_grace_wait_sum_us"] = sum; + mLastSyncedValues["claim_grace_wait_count"] = count; + } + // ── Fetch TxSet timer ── if (root.isMember("fetch_txset_sum_us") && root.isMember("fetch_txset_count")) diff --git a/src/scp/BallotProtocol.cpp b/src/scp/BallotProtocol.cpp index b69f5c2a8..0495928cb 100644 --- a/src/scp/BallotProtocol.cpp +++ b/src/scp/BallotProtocol.cpp @@ -54,7 +54,7 @@ BallotProtocol::isNewerStatement(NodeID const& nodeID, SCPStatement const& st) bool BallotProtocol::isNewerStatement(SCPStatement const& oldst, - SCPStatement const& st) + SCPStatement const& st) const { bool res = false; @@ -98,7 +98,7 @@ BallotProtocol::isNewerStatement(SCPStatement const& oldst, else { // Lexicographical order between PREPARE statements: - // (b, p, p', h) + // (b, p, p', h, c) auto const& oldPrep = oldst.pledges.prepare(); auto const& prep = st.pledges.prepare(); @@ -124,7 +124,16 @@ BallotProtocol::isNewerStatement(SCPStatement const& oldst, } else if (compBallot == 0) { - res = (oldPrep.nH < prep.nH); + if (mSlot.getSCPDriver() + .protocolAllowsEmptyTxSetValues() && + oldPrep.nH == prep.nH) + { + res = (oldPrep.nC < prep.nC); + } + else + { + res = (oldPrep.nH < prep.nH); + } } } } @@ -672,7 +681,7 @@ BallotProtocol::createStatement(SCPStatementType const& type) return statement; } -void +SCPStatement BallotProtocol::emitCurrentStateStatement() { ZoneScoped; @@ -726,6 +735,11 @@ BallotProtocol::emitCurrentStateStatement() throw std::runtime_error("moved to a bad state (ballot protocol)"); } } + + // Return the statement this call generated. Intentionally does not return + // the statement generated by recursion so that the caller may reason about + // what this call specifically produced. + return envelope.statement; } void @@ -1141,6 +1155,7 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH) mSlot.getSlotIndex(), mSlot.getSCP().ballotToStr(newH)); bool didWork = false; + bool stalled = false; // remember newH's value mValueOverride = mSlot.getSCPDriver().wrapValue(newH.value); @@ -1176,6 +1191,8 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH) mSlot.getSCPDriver().recordBallotBlockedOnTxSet( mSlot.getSlotIndex(), newC.value); + stalled = true; + CLOG_TRACE( SCP, "BallotProtocol::setConfirmPrepared slot:{} " @@ -1214,12 +1231,53 @@ BallotProtocol::setConfirmPrepared(SCPBallot const& newC, SCPBallot const& newH) if (didWork) { - emitCurrentStateStatement(); + auto const emitted = emitCurrentStateStatement(); + + if (stalled) + { + // Stalled waiting for the tx set corresponding to newC.value. + // Remember the state that existed at the stall point so that we can + // evaluate whether it's safe to resume (skipping a ballot timeout) + // if the tx set arrives. + mStalledCommit = StalledCommit{newC, newH, emitted}; + } + else + { + mStalledCommit.reset(); + } } return didWork; } +void +BallotProtocol::receivedTxSet(Value const& value) +{ + ZoneScoped; + // Only act if this slot stalled waiting for exactly this value's tx set. + if (!mStalledCommit || !(mStalledCommit->mCommitBallot.value == value)) + { + return; + } + + // It should not be possible to end up here prior to the protocol supporting + // kStructurallyValidValue. + releaseAssert(mSlot.getSCPDriver().protocolAllowsEmptyTxSetValues()); + + auto const stalled = *mStalledCommit; + mStalledCommit.reset(); + + // Resume only if the node has done no balloting work since the stall. + auto const* selfEnv = getLatestMessage(mSlot.getSCP().getLocalNodeID()); + if (selfEnv == nullptr || !(selfEnv->statement == stalled.mStallStatement)) + { + return; + } + + // Re-run the commit step setConfirmPrepared deferred. + setConfirmPrepared(stalled.mCommitBallot, stalled.mHighBallot); +} + void BallotProtocol::findExtendedInterval(Interval& candidate, std::set const& boundaries, diff --git a/src/scp/BallotProtocol.h b/src/scp/BallotProtocol.h index 2ff2cb120..1df9acd49 100644 --- a/src/scp/BallotProtocol.h +++ b/src/scp/BallotProtocol.h @@ -9,6 +9,7 @@ #include "util/GlobalChecks.h" #include #include +#include #include #include #include @@ -95,6 +96,16 @@ class BallotProtocol SCPEnvelopeWrapperPtr mLastEnvelopeEmit; // last envelope emitted by this node + // Information about the state of balloting upon stalling when attempting to + // set `c` to a value the node has not successfully fetched. + struct StalledCommit + { + SCPBallot mCommitBallot; // c (deferred; not in the emitted stmt) + SCPBallot mHighBallot; // h + SCPStatement mStallStatement; // self statement emitted at the stall + }; + std::optional mStalledCommit; + public: BallotProtocol(Slot& slot); @@ -119,6 +130,12 @@ class BallotProtocol // flavor that takes the actual desired counter value bool bumpState(Value const& value, uint32 n); + // Called when the tx set referenced by @p value arrives. + // If balloting stalled waiting for this tx set, AND the node's state has + // not changed since hitting the stall point, this function will resume + // balloting for this slot immediately. Otherwise, it does nothing. + void receivedTxSet(Value const& value); + // ** status methods // returns information about the local state in JSON format @@ -166,8 +183,8 @@ class BallotProtocol static std::set getStatementValues(SCPStatement const& st); // returns true if st is newer than oldst - static bool isNewerStatement(SCPStatement const& oldst, - SCPStatement const& st); + bool isNewerStatement(SCPStatement const& oldst, + SCPStatement const& st) const; private: // attempts to make progress using the latest statement as a hint @@ -305,9 +322,11 @@ class BallotProtocol // we have. bool updateCurrentValue(SCPBallot const& ballot); - // emits a statement reflecting the nodes' current state - // and attempts to make progress - void emitCurrentStateStatement(); + // Emits a statement reflecting the node's current state and attempts to + // make progress. Returns the statement it generated (built from current + // state, *before* the self-processing recursion that may advance the ballot + // further), so callers can capture exactly what this call produced. + SCPStatement emitCurrentStateStatement(); // verifies that the internal state is consistent void checkInvariants(); diff --git a/src/scp/SCP.cpp b/src/scp/SCP.cpp index 6bb7edce3..8b391896c 100644 --- a/src/scp/SCP.cpp +++ b/src/scp/SCP.cpp @@ -51,6 +51,16 @@ SCP::stopNomination(uint64 slotIndex) } } +void +SCP::receivedTxSet(uint64 slotIndex, Value const& value) +{ + auto s = getSlot(slotIndex, false); + if (s) + { + s->receivedTxSet(value); + } +} + void SCP::updateLocalQuorumSet(SCPQuorumSet const& qSet) { diff --git a/src/scp/SCP.h b/src/scp/SCP.h index f06980937..c99fb86be 100644 --- a/src/scp/SCP.h +++ b/src/scp/SCP.h @@ -59,6 +59,10 @@ class SCP // stops nomination for a slot void stopNomination(uint64 slotIndex); + // Notify SCP that the tx set referenced by @p value has arrived so that it + // may resume balloting if stalled waiting for this tx set. + void receivedTxSet(uint64 slotIndex, Value const& value); + // Local QuorumSet interface (can be dynamically updated) void updateLocalQuorumSet(SCPQuorumSet const& qSet); SCPQuorumSet const& getLocalQuorumSet(); diff --git a/src/scp/Slot.cpp b/src/scp/Slot.cpp index bff055757..ba15e214f 100644 --- a/src/scp/Slot.cpp +++ b/src/scp/Slot.cpp @@ -131,7 +131,7 @@ Slot::isNewerNominationOrBallotSt(SCPStatement const& oldSt, } else { - if (BallotProtocol::isNewerStatement(oldSt, newSt)) + if (mBallotProtocol.isNewerStatement(oldSt, newSt)) { replace = true; } @@ -217,6 +217,12 @@ Slot::abandonBallot() return mBallotProtocol.abandonBallot(0); } +void +Slot::receivedTxSet(Value const& value) +{ + mBallotProtocol.receivedTxSet(value); +} + bool Slot::bumpState(Value const& value, bool force) { diff --git a/src/scp/Slot.h b/src/scp/Slot.h index ecf1e8fee..e43f0b857 100644 --- a/src/scp/Slot.h +++ b/src/scp/Slot.h @@ -117,6 +117,10 @@ class Slot : public std::enable_shared_from_this bool abandonBallot(); + // Notify this slot that the tx set referenced by @p value has arrived so + // that it may resume balloting if stalled waiting for this tx set. + void receivedTxSet(Value const& value); + // bumps the ballot based on the local state and the value passed in: // in prepare phase, attempts to take value // otherwise, no-ops diff --git a/src/scp/test/SCPTests.cpp b/src/scp/test/SCPTests.cpp index f9ded04b0..f6ea5f36e 100644 --- a/src/scp/test/SCPTests.cpp +++ b/src/scp/test/SCPTests.cpp @@ -43,10 +43,12 @@ class TestSCP : public SCPDriver uint32_t mIncrementBallotTimeoutMS = 1000; uint32_t mInitialNominationTimeoutMS = 1000; uint32_t mIncrementNominationTimeoutMS = 1000; + bool const mProtocolAllowsEmptyTxSetValues; TestSCP(NodeID const& nodeID, SCPQuorumSet const& qSetLocal, - bool isValidator = true) + bool isValidator = true, bool protocolAllowsEmptyTxSetValues = true) : mSCP(*this, nodeID, isValidator, qSetLocal) + , mProtocolAllowsEmptyTxSetValues(protocolAllowsEmptyTxSetValues) { mPriorityLookup = [&](NodeID const& n) { return (n == mSCP.getLocalNodeID()) ? 1000 : 1; @@ -163,13 +165,17 @@ class TestSCP : public SCPDriver bool isParallelTxSetDownloadEnabled() const override { - return true; + // Leave unimplemented. A node's parallel downloading setting only + // affects higher level systems (such as PendingEnvelopes). + // NominationProtocol and BallotProtocol only reason about whether the + // protocol supports empty-tx-set values + releaseAssert(false); } bool protocolAllowsEmptyTxSetValues() const override { - return true; + return mProtocolAllowsEmptyTxSetValues; } void @@ -185,6 +191,12 @@ class TestSCP : public SCPDriver return mSCP.getSlot(slotIndex, true)->bumpState(v, true); } + void + receivedTxSet(uint64 slotIndex, Value const& v) + { + mSCP.receivedTxSet(slotIndex, v); + } + bool nominate(uint64 slotIndex, Value const& value, bool timedout) { @@ -853,7 +865,9 @@ TEST_CASE("ballot protocol core5", "[scp][ballotprotocol]") uint256 qSetHash = sha256(xdr::xdr_to_opaque(qSet)); - TestSCP scp(v0SecretKey.getPublicKey(), qSet); + bool const protocolAllowsEmptyTxSetValues = GENERATE(false, true); + TestSCP scp(v0SecretKey.getPublicKey(), qSet, /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); auto test = [&](TestSCP& scp) { scp.storeQuorumSet(std::make_shared(qSet)); @@ -2643,7 +2657,8 @@ TEST_CASE("ballot protocol core5", "[scp][ballotprotocol]") SECTION("non validator watching the network") { SIMULATION_CREATE_NODE(NV); - TestSCP scpNV(vNVSecretKey.getPublicKey(), qSet, false); + TestSCP scpNV(vNVSecretKey.getPublicKey(), qSet, false, + protocolAllowsEmptyTxSetValues); scpNV.storeQuorumSet(std::make_shared(qSet)); uint256 qSetHashNV = scpNV.mSCP.getLocalNode()->getQuorumSetHash(); @@ -2672,7 +2687,8 @@ TEST_CASE("ballot protocol core5", "[scp][ballotprotocol]") SECTION("restore ballot protocol") { - TestSCP scp2(v0SecretKey.getPublicKey(), qSet); + TestSCP scp2(v0SecretKey.getPublicKey(), qSet, /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); scp2.storeQuorumSet(std::make_shared(qSet)); SCPBallot b(2, xValue); SECTION("prepare") @@ -2717,7 +2733,9 @@ TEST_CASE("ballot protocol core3", "[scp][ballotprotocol]") uint256 qSetHash = sha256(xdr::xdr_to_opaque(qSet)); - TestSCP scp(v0SecretKey.getPublicKey(), qSet); + bool const protocolAllowsEmptyTxSetValues = GENERATE(false, true); + TestSCP scp(v0SecretKey.getPublicKey(), qSet, /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); auto test = [&](TestSCP& scp) { scp.storeQuorumSet(std::make_shared(qSet)); @@ -2864,7 +2882,9 @@ TEST_CASE("ballot protocol core3", "[scp][ballotprotocol]") SECTION("node without self - quorum timeout") { SIMULATION_CREATE_NODE(NodeNS); - TestSCP scpNNS(vNodeNSSecretKey.getPublicKey(), qSet); + TestSCP scpNNS(vNodeNSSecretKey.getPublicKey(), qSet, + /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); scpNNS.storeQuorumSet(std::make_shared(qSet)); uint256 qSetHashNodeNS = scpNNS.mSCP.getLocalNode()->getQuorumSetHash(); @@ -2921,9 +2941,11 @@ TEST_CASE("nomination tests core5", "[scp][nominationprotocol]") expectedLeaders.end())); }; + bool const protocolAllowsEmptyTxSetValues = GENERATE(false, true); SECTION("nomination - v0 is top") { - TestSCP scp(v0SecretKey.getPublicKey(), qSet); + TestSCP scp(v0SecretKey.getPublicKey(), qSet, /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); auto test = [&](TestSCP& scp) { uint256 qSetHash0 = scp.mSCP.getLocalNode()->getQuorumSetHash(); @@ -3037,7 +3059,9 @@ TEST_CASE("nomination tests core5", "[scp][nominationprotocol]") } SECTION("nomination - restored state") { - TestSCP scp2(v0SecretKey.getPublicKey(), qSet); + TestSCP scp2(v0SecretKey.getPublicKey(), qSet, + /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); scp2.storeQuorumSet( std::make_shared(qSet)); @@ -3274,7 +3298,8 @@ TEST_CASE("nomination tests core5", "[scp][nominationprotocol]") } SECTION("v1 is top node") { - TestSCP scp(v0SecretKey.getPublicKey(), qSet); + TestSCP scp(v0SecretKey.getPublicKey(), qSet, /*isValidator*/ true, + protocolAllowsEmptyTxSetValues); auto test = [&](TestSCP& scp) { uint256 qSetHash0 = scp.mSCP.getLocalNode()->getQuorumSetHash(); @@ -3483,7 +3508,7 @@ TEST_CASE("nomination tests core5", "[scp][nominationprotocol]") } } -#ifdef CAP_0087 +#ifdef CAP_0083 TEST_CASE("nomination times out structurally-valid value into empty tx set", "[scp][nomination]") { @@ -3865,6 +3890,114 @@ TEST_CASE("setConfirmPrepared stalls on kStructurallyValidValue value", } REQUIRE(foundC); } + + SECTION("resumes immediately on receivedTxSet") + { + auto const envsBeforeClear = scp.mEnvs.size(); + + // Simulate tx set arrival: the stalled value becomes fully validated. + scp.clearDownload(xValue); + + // Resume the deferred commit directly — no ballot bump, no new + // envelopes, no timer. + scp.receivedTxSet(0, xValue); + + // The commit now completes at the SAME counter it stalled on (1) — not + // a bumped counter — proving the resume, not a re-drive. + REQUIRE(scp.mEnvs.size() > envsBeforeClear); + auto const& lastPrep = scp.mEnvs.back().statement.pledges.prepare(); + REQUIRE(lastPrep.nC == 1); + REQUIRE(lastPrep.nH == 1); + + // A repeat delivery no-ops. + auto const envsAfterResume = scp.mEnvs.size(); + scp.receivedTxSet(0, xValue); + REQUIRE(scp.mEnvs.size() == envsAfterResume); + } + + SECTION("receivedTxSet no-ops for a value the slot is not stalled on") + { + auto const envsBefore = scp.mEnvs.size(); + + scp.clearDownload(xValue); + // yValue is not what balloting stalled on, so the stash does not match. + scp.receivedTxSet(0, yValue); + + REQUIRE(scp.mEnvs.size() == envsBefore); + } + + SECTION("receivedTxSet with bad tx set stays stalled") + { + auto const envsBefore = scp.mEnvs.size(); + + // No clearDownload: xValue is still only kStructurallyValidValue, so + // re-running the commit step must re-stall rather than commit. + // Simulates a bad (only structurally valid) tx set arriving. + scp.receivedTxSet(0, xValue); + + REQUIRE(scp.mEnvs.size() == envsBefore); + } + + SECTION("receivedTxSet declines after a local ballot bump (state changed)") + { + // The race the guard exists for: the ballot timer fires (bumping the + // ballot) just before the tx set arrives, so the stash is stale. + REQUIRE(scp.bumpState(0, xValue)); + auto const envsAfterBump = scp.mEnvs.size(); + { + auto const& prep = scp.mEnvs.back().statement.pledges.prepare(); + REQUIRE(prep.ballot.counter == 2); + REQUIRE(prep.nC == 0); + REQUIRE(prep.nH == 1); + } + + scp.clearDownload(xValue); + scp.receivedTxSet(0, xValue); + + // The self statement changed since the stall (b bumped 1 -> 2), so + // the resume declines: no emission, no commit. + REQUIRE(scp.mEnvs.size() == envsAfterBump); + + // The normal path still completes once the network confirms prepared + // at the bumped counter (tx set present -> no re-stall). + SCPBallot xB2(2, xValue); + REQUIRE(scp.receiveEnvelope( + makePrepare(v1SecretKey, qSetHash, 0, xB2, &xB2)) == + SCP::EnvelopeState::VALID); + REQUIRE(scp.receiveEnvelope( + makePrepare(v2SecretKey, qSetHash, 0, xB2, &xB2)) == + SCP::EnvelopeState::VALID); + auto const& lastPrep = scp.mEnvs.back().statement.pledges.prepare(); + REQUIRE(lastPrep.nC == 2); + REQUIRE(lastPrep.nH == 2); + } + + SECTION("receivedTxSet declines after accepting a higher incompatible " + "prepared") + { + // {v1, v2} is a quorum voting prepare (2, y): the node accepts it as + // prepared (p = (2,y), p' = (1,x)), which invalidates the resume path. + SCPBallot yB2(2, yValue); + REQUIRE( + scp.receiveEnvelope(makePrepare(v1SecretKey, qSetHash, 0, yB2)) == + SCP::EnvelopeState::VALID); + REQUIRE( + scp.receiveEnvelope(makePrepare(v2SecretKey, qSetHash, 0, yB2)) == + SCP::EnvelopeState::VALID); + + auto const envsAfterPrepared = scp.mEnvs.size(); + auto const& prep = scp.mEnvs.back().statement.pledges.prepare(); + REQUIRE(prep.prepared); + REQUIRE(*prep.prepared == yB2); + REQUIRE(prep.nC == 0); + REQUIRE(prep.nH == 1); + + scp.clearDownload(xValue); + scp.receivedTxSet(0, xValue); + + // Statement changed since the stall -> resume declines, no emission. + REQUIRE(scp.mEnvs.size() == envsAfterPrepared); + } } TEST_CASE("incoming PREPARE with structurally valid prepared value is accepted", @@ -3937,6 +4070,6 @@ TEST_CASE("incoming PREPARE with non-tx-set-invalid value is dropped", // No local emit triggered. REQUIRE(scp.mEnvs.empty()); } -#endif // CAP_0087 +#endif // CAP_0083 }