diff --git a/Cargo.lock b/Cargo.lock index eccdc93..54fb1f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1964,8 +1964,8 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "meta-feeder-sdk" -version = "1.7.0" -source = "git+https://github.com/worph/meta-feeder-sdk?tag=v1.7.0#027917c8cffe7ca43dadcf3b426cc5f69ecc443a" +version = "1.8.0" +source = "git+https://github.com/worph/meta-feeder-sdk?tag=v1.8.0#612f2aed661925cd22d51b29564832991e246e4d" dependencies = [ "anyhow", "async-trait", @@ -1990,7 +1990,7 @@ dependencies = [ [[package]] name = "meta-transport-ipfs" -version = "1.3.0" +version = "1.3.1" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index cb5a4b4..bef62b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "meta-transport-ipfs" -version = "1.3.0" +version = "1.3.1" edition = "2021" rust-version = "1.85" license = "MIT" diff --git a/src/swarm/event_loop.rs b/src/swarm/event_loop.rs index c80dcce..fe56e84 100644 --- a/src/swarm/event_loop.rs +++ b/src/swarm/event_loop.rs @@ -28,6 +28,7 @@ use super::bitswap_client::BitswapInflight; use super::identify_agent::parse_base_url; use super::kad_discovery::{is_dialable_addr, namespace_key, peer_id_from_multiaddr, KadConfig}; use super::peer_directory::PeerDirectory; +use super::provide_queue::{ProvideQueue, DEFAULT_PROVIDE_CONCURRENCY}; use super::{Behaviour, BehaviourEvent, Command, GatewayCapsConfig, PeersInfo}; /// How often [`SwarmTask::on_gateway_keepalive_tick`] re-dials dropped @@ -102,6 +103,8 @@ pub(super) struct SwarmTask { /// Gateways kept connected + routable by the keep-alive tick. See /// [`GatewayPin`]. pinned_gateways: HashMap, + /// Per-CID announcements waiting for a walk slot. See [`ProvideQueue`]. + provides: ProvideQueue, } impl SwarmTask { @@ -141,6 +144,12 @@ impl SwarmTask { http, gateway_caps, pinned_gateways: HashMap::new(), + provides: ProvideQueue::new( + std::env::var("META_SHARE_PROVIDE_CONCURRENCY") + .ok() + .and_then(|v| v.trim().parse().ok()) + .unwrap_or(DEFAULT_PROVIDE_CONCURRENCY), + ), } } @@ -291,23 +300,43 @@ impl SwarmTask { // Announce on the public IPFS DHT, keyed by the CID's // multihash bytes — that's the kubo-compatible provider key, // so a vanilla IPFS node `get_providers`ing the same content - // finds us. The StartProviding result is logged in - // `default_log`. Works in `Mode::Client` (same call the - // gateway makes). + // finds us. Queued, not fired: each announce is a DHT walk, and + // a burst of seeds must not become a burst of walks (see + // `provide_queue`). The StartProviding result is logged in + // `default_log`. Works in `Mode::Client`. let key = RecordKey::new(&cid.hash().to_bytes()); - match swarm.behaviour_mut().kad.start_providing(key) { - Ok(qid) => debug!(?qid, %cid, "kad: providing seeded cid"), - Err(e) => warn!(error = %e, %cid, "kad: start_providing(cid) failed"), + if self.provides.enqueue(key) { + debug!(%cid, pending = self.provides.pending_len(), + "kad: queued seeded cid for announce"); } + self.pump_provides(swarm); } Command::StopProviding { cid } => { let key = RecordKey::new(&cid.hash().to_bytes()); + self.provides.cancel(&key); swarm.behaviour_mut().kad.stop_providing(&key); debug!(%cid, "kad: stopped providing seeded cid"); } } } + /// Start queued announcements while walk slots are free. + fn pump_provides(&mut self, swarm: &mut Swarm) { + while let Some(key) = self.provides.next_to_start() { + match swarm.behaviour_mut().kad.start_providing(key) { + Ok(qid) => { + self.provides.started(qid); + debug!(?qid, in_flight = self.provides.in_flight_len(), + pending = self.provides.pending_len(), "kad: announcing seeded cid"); + } + Err(e) => { + self.provides.failed_to_start(); + warn!(error = %e, "kad: start_providing(cid) failed"); + } + } + } + } + /// Fired every `kad_discovery_interval` (steady-state) or /// `kad_fast_discovery_interval` (below floor). Issues a fresh /// `get_providers` query and adapts the next tick's cadence based on @@ -567,6 +596,18 @@ impl SwarmTask { trace!("bitswap: inflight query resolved"); } } + // 2d) A queued announce finished (ok or timed out): free its slot and + // start the next. Only on the query's last step — until then the + // walk is still dialing. + if let SwarmEvent::Behaviour(BehaviourEvent::Kad( + kad::Event::OutboundQueryProgressed { + id, result: QueryResult::StartProviding(_), step, .. + } + )) = &event { + if step.last && self.provides.finished(*id) { + self.pump_provides(swarm); + } + } // 3) kad providers — dial each new peer-id (subject to soft ceiling). if let SwarmEvent::Behaviour(BehaviourEvent::Kad( kad::Event::OutboundQueryProgressed { diff --git a/src/swarm/mod.rs b/src/swarm/mod.rs index 1a61e90..ede158a 100644 --- a/src/swarm/mod.rs +++ b/src/swarm/mod.rs @@ -128,6 +128,7 @@ mod event_loop; mod identify_agent; mod kad_discovery; mod peer_directory; +mod provide_queue; pub use bitswap_client::bitswap_get_block; pub use identify_agent::build_agent_version; diff --git a/src/swarm/provide_queue.rs b/src/swarm/provide_queue.rs new file mode 100644 index 0000000..206f174 --- /dev/null +++ b/src/swarm/provide_queue.rs @@ -0,0 +1,167 @@ +//! Bounded admission for per-CID DHT announcements. +//! +//! Every `Command::Provide` is a `kad.start_providing`, and every one of those is +//! a full DHT walk: find the ~20 closest peers to the key, then dial each to store +//! the record. Firing them unbounded turns a batch of seeds into a dial storm. On +//! watch.nsl.sh (2026-10-06 18:14:46) a lapsed playback-focus lease released 45 +//! queued poster seeds in one second; 45 concurrent walks took this plugin from +//! ~300 to 8,066 log lines a minute (6.8k failed dials) while a viewer was +//! mid-episode. +//! +//! So announcements queue here and at most `max_in_flight` walks run at once. +//! The queue is pure bookkeeping (no swarm, no I/O) so its rules are unit-tested; +//! [`super::event_loop`] feeds it commands and query results and starts whatever +//! [`ProvideQueue::next_to_start`] hands back. +//! +//! What it does NOT bound: libp2p-kad's own periodic republication of records +//! already provided (`provider_publication_interval`), which kad paces itself. + +use std::collections::{HashSet, VecDeque}; + +use libp2p::kad::{QueryId, RecordKey}; + +/// Default number of concurrent announce walks (`META_SHARE_PROVIDE_CONCURRENCY`). +pub(crate) const DEFAULT_PROVIDE_CONCURRENCY: usize = 2; + +pub(crate) struct ProvideQueue { + pending: VecDeque, + queued: HashSet, + in_flight: HashSet, + max_in_flight: usize, +} + +impl ProvideQueue { + /// `max_in_flight == 0` is treated as 1: a queue that never drains would + /// silently stop announcing anything. + pub(crate) fn new(max_in_flight: usize) -> Self { + Self { + pending: VecDeque::new(), + queued: HashSet::new(), + in_flight: HashSet::new(), + max_in_flight: max_in_flight.max(1), + } + } + + /// Queue `key`. `false` if it is already waiting (a re-seed of the same cid + /// before its turn costs nothing extra). + pub(crate) fn enqueue(&mut self, key: RecordKey) -> bool { + if !self.queued.insert(key.clone()) { + return false; + } + self.pending.push_back(key); + true + } + + /// Drop a still-waiting `key` (its seed was removed before its turn). A walk + /// already in flight is left alone; the caller's `stop_providing` covers it. + pub(crate) fn cancel(&mut self, key: &RecordKey) { + if self.queued.remove(key) { + self.pending.retain(|k| k != key); + } + } + + /// The next key to announce, if a slot is free. The caller must report the + /// started query with [`started`](Self::started), or [`failed_to_start`](Self::failed_to_start). + pub(crate) fn next_to_start(&mut self) -> Option { + if self.in_flight.len() >= self.max_in_flight { + return None; + } + let key = self.pending.pop_front()?; + self.queued.remove(&key); + Some(key) + } + + pub(crate) fn started(&mut self, qid: QueryId) { + self.in_flight.insert(qid); + } + + /// `start_providing` refused synchronously (e.g. the record store is full): + /// nothing is in flight, so there is nothing to record. Exists so the call + /// site reads symmetrically. + pub(crate) fn failed_to_start(&mut self) {} + + /// A `StartProviding` query finished. `true` if it was one of ours (the + /// namespace re-publish uses the same query type and is not queued). + pub(crate) fn finished(&mut self, qid: QueryId) -> bool { + self.in_flight.remove(&qid) + } + + pub(crate) fn pending_len(&self) -> usize { + self.pending.len() + } + + pub(crate) fn in_flight_len(&self) -> usize { + self.in_flight.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use libp2p::kad::{store::MemoryStore, Behaviour}; + use libp2p::{identity, PeerId}; + + fn key(n: u8) -> RecordKey { + RecordKey::new(&[n]) + } + + /// Real `QueryId`s can only be minted by a kad behaviour; start throwaway + /// queries on one (nothing is ever polled, so no I/O happens). + fn qids(n: usize) -> Vec { + let peer = PeerId::from(identity::Keypair::generate_ed25519().public()); + let mut kad = Behaviour::new(peer, MemoryStore::new(peer)); + (0..n).map(|i| kad.get_providers(key(i as u8))).collect() + } + + #[test] + fn at_most_max_in_flight_walks_run_at_once() { + let mut q = ProvideQueue::new(2); + for n in 0..45 { + assert!(q.enqueue(key(n))); + } + let ids = qids(3); + assert_eq!(q.next_to_start(), Some(key(0))); + q.started(ids[0]); + assert_eq!(q.next_to_start(), Some(key(1))); + q.started(ids[1]); + assert_eq!(q.next_to_start(), None, "both slots busy"); + assert_eq!(q.pending_len(), 43); + + assert!(q.finished(ids[0])); + assert_eq!(q.next_to_start(), Some(key(2)), "a finished walk frees its slot, FIFO"); + q.started(ids[2]); + assert_eq!(q.in_flight_len(), 2); + } + + #[test] + fn a_duplicate_waits_once_and_a_cancelled_key_never_starts() { + let mut q = ProvideQueue::new(1); + assert!(q.enqueue(key(1))); + assert!(!q.enqueue(key(1)), "already queued"); + assert!(q.enqueue(key(2))); + q.cancel(&key(1)); + assert_eq!(q.next_to_start(), Some(key(2))); + assert_eq!(q.pending_len(), 0); + // Once started (dequeued), the same cid may be queued again. + assert!(q.enqueue(key(2))); + } + + #[test] + fn a_foreign_query_does_not_free_a_slot() { + let mut q = ProvideQueue::new(1); + q.enqueue(key(1)); + q.enqueue(key(2)); + let ids = qids(2); + q.next_to_start(); + q.started(ids[0]); + assert!(!q.finished(ids[1]), "the namespace re-publish is not ours"); + assert_eq!(q.next_to_start(), None); + } + + #[test] + fn zero_concurrency_still_drains() { + let mut q = ProvideQueue::new(0); + q.enqueue(key(1)); + assert_eq!(q.next_to_start(), Some(key(1))); + } +}