From 0350bbe445f4b182f9a341349e5036ce329c22bc Mon Sep 17 00:00:00 2001 From: Michael Taylor Date: Tue, 28 Jul 2026 03:51:10 -0700 Subject: [PATCH] fix(holdings): re-announce holdings when the peer pool rises from zero (#1734) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A node that pinned content while its peer pool was empty never announced it to any peer, and never re-announced when peers arrived — the leading candidate cause of the live network showing zero providers. The inventory-change reaction floods an opcode-222 announcement only for a NON-EMPTY delta diffed against the node's OWN local DHT provider records. Those records answer "what changed here", never "what do my peers know", and the two diverge silently the moment an inventory change happens with nobody connected: the pin moves the local records, the flood reaches zero peers, and every later reconcile of the same inventory is a no-op. A restart is worse than a repeat of it — bring-up seeds the remembered inventory from the on-disk cache before any peer connects, so a restarted node with content already cached has a permanently empty delta and never floods at all, even with peers connected. The fix removes the divergence rather than patching one of its symptoms: the node re-states its holdings IN FULL — every held content id as an `Add`, no diff — whenever its pool rises from zero peers to some, including the first such observation after bring-up. Re-stating carries no state to be wrong about, and is cheap because an `Add` is idempotent at every receiver under an advancing `seq`. A pool that merely grows while already peered does not re-flood. Version: minor (0.65.0) — new announce behaviour, no API or wire change. Co-Authored-By: Claude --- Cargo.lock | 4 +- Cargo.toml | 2 +- SPEC.md | 14 ++ crates/dig-node-core/Cargo.toml | 2 +- crates/dig-node-core/src/peer.rs | 24 +++ .../src/seams/dig_peer/holdings.rs | 129 ++++++++++++++ crates/dig-node-core/tests/holdings_wire.rs | 167 +++++++++++++++++- 7 files changed, 336 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index def367c..55764dd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2211,7 +2211,7 @@ dependencies = [ [[package]] name = "dig-node-core" -version = "0.22.0" +version = "0.23.0" dependencies = [ "async-trait", "axum", @@ -2266,7 +2266,7 @@ dependencies = [ [[package]] name = "dig-node-service" -version = "0.64.0" +version = "0.65.0" dependencies = [ "async-trait", "axum", diff --git a/Cargo.toml b/Cargo.toml index 5ff5e3a..5937d8e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,7 @@ edition = "2021" # the ROOT manifest (`[workspace.package].version`), so it MUST be set here for a # release to fire (§3.6). The library crates (dig-node-core/dig-runtime/dig-wallet) # keep their own independent versions — only the released binary tracks the workspace version. -version = "0.64.0" +version = "0.65.0" # Release hardening, matching digstore: keep integer-overflow checks ON in release. # The node parses untrusted serialized input and does offset/length arithmetic over diff --git a/SPEC.md b/SPEC.md index 90baf25..84c40cf 100644 --- a/SPEC.md +++ b/SPEC.md @@ -3152,6 +3152,20 @@ retracts and leave this node advertising content it does not hold. - A degraded node (no signable leaf, no inbound receiver) MUST remain discoverable through the durable DHT records — the real-time layer is additive and MUST NOT be a hard dependency of discoverability. +**Peer-presence re-announce (MUST, #1734).** A node MUST re-announce its CURRENT holdings in full — every +held content id as an `Add`, with no diff — whenever its connected peer pool rises from ZERO peers to one +or more, including the first such observation after bring-up. It MUST NOT re-announce merely because an +already-peered pool grew. + +The reconcile delta above is computed against this node's OWN local provider records, so it answers "what +changed here", never "what do my peers know". Those two diverge silently as soon as an inventory change +happens with nobody connected: the local records move, the flood reaches zero peers, and every later +reconcile of the same inventory is a no-op. Without this rule a node that pinned before its first peer — or +that RESTARTED with content already cached, where the remembered inventory is seeded from disk before any +peer connects — holds the content, has recorded it as announced, and is invisible to every peer it later +connects to, with a restart re-entering the same state. Re-stating the whole inventory is safe because an +`Add` is idempotent at every receiver under an advancing `seq`. + **Ingress (MUST).** An inbound announcement is applied ONLY after all of the following, fail-closed, at a single chokepoint. dig-dht is crypto-free by design, and `ingest_verified_provider` is the sole sanctioned bypass of its mTLS self-announce check, so these are the whole of the authentication: diff --git a/crates/dig-node-core/Cargo.toml b/crates/dig-node-core/Cargo.toml index f37e595..1fba63b 100644 --- a/crates/dig-node-core/Cargo.toml +++ b/crates/dig-node-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dig-node-core" -version = "0.22.0" +version = "0.23.0" edition = "2021" license = "GPL-2.0-only" description = "The canonical DIG node ENGINE library (crate `dig_node_core`): the JSON-RPC dispatch (`handle_rpc`, the same contract as rpc.dig.net), local-first content serve/fetch/redirect from LOCAL .dig store modules (via digstore_host::serve_blind), chain-anchored-root resolution, chain-watch + subscriptions + generation gap-fill, the LRU cache, and the full P2P stack. Shared UNCHANGED by both host shells: the `dig-node` OS-service binary (dig-node-service) and the DIG Browser's in-process cdylib (dig-runtime). Native Rust so the compiled-module serve path works." diff --git a/crates/dig-node-core/src/peer.rs b/crates/dig-node-core/src/peer.rs index d54f8a8..cc609c7 100644 --- a/crates/dig-node-core/src/peer.rs +++ b/crates/dig-node-core/src/peer.rs @@ -859,6 +859,16 @@ impl crate::seams::dig_peer::AnnounceHolder for DhtInventoryAnnouncer { } } +/// The node's cache list, as the holdings layer's peer-presence announcer reads it (#1734). +struct NodeHoldingsInventory(Arc); + +#[async_trait::async_trait] +impl crate::seams::dig_peer::holdings::HoldingsInventory for NodeHoldingsInventory { + async fn current(&self) -> Vec { + self.0.cache_list_cached().await + } +} + /// The signer plus the pool it floods to — everything needed to emit an opcode-222 announcement. #[derive(Clone)] struct HoldingsFlood { @@ -2269,6 +2279,20 @@ async fn run_peer_network(node: Arc) -> Result<(), String> { pool: handle.clone(), }); + // Re-state the node's holdings the first time it sees a connected peer, and on every later + // return-from-zero (#1734). Without this a node is only ever heard when its inventory CHANGES while + // a peer happens to be listening: a pin at zero peers — and a restart, whose remembered inventory is + // seeded from disk before any peer connects — moves the local records and leaves every later + // reconcile a no-op, so the node holds the content, believes it announced, and is invisible. + if let Some(flood) = holdings_flood.as_ref() { + tokio::spawn(crate::seams::dig_peer::holdings::run_first_peer_announcer( + handle.clone(), + Arc::new(NodeHoldingsInventory(node.clone())) + as Arc, + flood.broadcaster.clone(), + )); + } + // 4a. Feed the DHT routing table LIVE from the gossip pool (#1574): bring_up_dht's one-shot // bootstrap runs BEFORE any peer connects, so in a freshly-formed network routing starts empty // and find_providers finds nobody. This forwards every PoolEvent (add → insert, remove → diff --git a/crates/dig-node-core/src/seams/dig_peer/holdings.rs b/crates/dig-node-core/src/seams/dig_peer/holdings.rs index 44e7e30..4058a31 100644 --- a/crates/dig-node-core/src/seams/dig_peer/holdings.rs +++ b/crates/dig-node-core/src/seams/dig_peer/holdings.rs @@ -797,6 +797,135 @@ pub async fn reconcile_and_announce( delta } +/// Announce this node's ENTIRE current holdings as `Add` deltas, ignoring any diff. Returns how many +/// frames were sent. +/// +/// Distinct from [`reconcile_and_announce`] on purpose, and the distinction is the #1734 fix. That +/// function announces a DELTA computed against the node's OWN local DHT records, which answers "what +/// changed here", not "what do my peers know". Those two diverge silently the moment an inventory +/// change happens with nobody listening: the local records move, the flood reaches zero peers, and +/// every later reconcile of the same inventory is a no-op — so a node that pinned before its first peer +/// (or restarted with content already cached, where the remembered set is seeded from disk) holds the +/// capsule, believes it announced, and is invisible to every peer it later connects to. +/// +/// This function has no such state to be wrong about: it re-states the whole truth. Re-stating is safe +/// and cheap because an `Add` is idempotent at every receiver — a re-ingested record refreshes the same +/// provider entry under an advancing `seq` — so the repair costs one frame per inventory batch and can +/// never contradict the durable records it mirrors. +pub async fn announce_all_holdings( + broadcaster: &HoldingsBroadcaster, + transport: &dyn AnnounceTransport, + cached: &[crate::CachedCapsule], + now: u64, +) -> usize { + let held = crate::dht::inventory_content_ids(cached); + if held.is_empty() { + return 0; // nothing to state; a node holding nothing has nothing to be invisible about + } + broadcaster + .announce_change(transport, &held, &[], now) + .await +} + +/// The node's current cached inventory, as the holdings layer needs to read it. +/// +/// A trait rather than a `&Node` so the peer-presence announcer below is drivable over a real gossip +/// wire without a node or a disk — the wiring across two nodes is the only place the #1734 defect is +/// visible, so that path has to be testable. +#[async_trait::async_trait] +pub trait HoldingsInventory: Send + Sync { + /// The capsules this node currently holds. + async fn current(&self) -> Vec; +} + +/// Whether this node currently observes any connected peer — the edge the announce hangs on. +/// +/// Kept as a tiny explicit state machine, separate from the task that drives it, because the EDGE +/// definition is the whole policy: it must fire on `0 -> N` (the node was unheard, so its holdings must +/// be re-stated) and must NOT fire when an already-peered pool merely grows (that would re-flood the +/// full inventory on every pool addition). A total loss of peers re-arms it, since a node whose only +/// peer went away is invisible again the moment one returns. +#[derive(Debug, Default)] +pub struct PoolPresence { + has_peers: bool, +} + +impl PoolPresence { + /// Fold in an observed connected-peer count; `true` means "announce now". + pub fn observe(&mut self, connected: usize) -> bool { + let rising = connected > 0 && !self.has_peers; + self.has_peers = connected > 0; + rising + } +} + +/// Re-state this node's holdings to its peers whenever the pool rises from zero peers to some (#1734). +/// +/// Spawned once at bring-up, and the reason "I hold X" and "peers know I hold X" cannot drift apart for +/// long: the very first non-empty pool observation announces the current inventory in full, so a pin +/// that happened with nobody connected — and a restart whose remembered inventory came off disk — both +/// repair themselves the moment a peer arrives, with no unpin/repin dance. Ends when the pool's event +/// channel closes. +/// +/// Best-effort by construction: a failed subscribe logs and returns, leaving the durable DHT provider +/// records as the discovery path (a freshness degradation, never an outage). +pub async fn run_first_peer_announcer( + pool: dig_gossip::GossipHandle, + inventory: Arc, + broadcaster: Arc, +) { + // Subscribed BEFORE the first count is read, so a peer that connects between the two is still seen + // as an event rather than silently falling into the gap. + let mut events = match pool.subscribe_pool_events() { + Ok(events) => events, + Err(e) => { + tracing::warn!( + error = %e, + "holdings announce: no pool events; holdings pinned before the first peer will only \ + be discoverable through the durable DHT records" + ); + return; + } + }; + let mut presence = PoolPresence::default(); + let mut announce_if_rising = |connected: usize| presence.observe(connected); + + if announce_if_rising(pool.connected_pool_peers().len()) { + restate_holdings(&pool, inventory.as_ref(), broadcaster.as_ref()).await; + } + loop { + match events.recv().await { + // The count is re-read from the pool rather than tracked from the event stream: the pool is + // the authority on who is connected, and a lagged receiver would otherwise leave a + // reconstructed count permanently wrong. + Ok(_) => { + if announce_if_rising(pool.connected_pool_peers().len()) { + restate_holdings(&pool, inventory.as_ref(), broadcaster.as_ref()).await; + } + } + // Lagged: the count is re-read anyway, so a missed event costs nothing but a re-check. + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, + } + } +} + +/// Flood the node's whole current inventory to the pool, logging what it re-stated. +async fn restate_holdings( + pool: &dig_gossip::GossipHandle, + inventory: &dyn HoldingsInventory, + broadcaster: &HoldingsBroadcaster, +) { + let cached = inventory.current().await; + let held = cached.len(); + let frames = announce_all_holdings(broadcaster, pool, &cached, now_unix_secs()).await; + tracing::info!( + held, + frames, + "dig-node holdings: peers arrived — re-announced this node's current holdings" + ); +} + /// Consume inbound opcode-222 frames from the gossip pool forever, applying each through `ingress`. /// /// Spawned once at bring-up. Non-222 frames are ignored (other subscribers handle them), and a lagged diff --git a/crates/dig-node-core/tests/holdings_wire.rs b/crates/dig-node-core/tests/holdings_wire.rs index a603014..491528b 100644 --- a/crates/dig-node-core/tests/holdings_wire.rs +++ b/crates/dig-node-core/tests/holdings_wire.rs @@ -33,8 +33,8 @@ use dig_gossip::{ }; use dig_node_core::peer::{install_crypto_provider, load_or_generate_node_cert}; use dig_node_core::seams::dig_peer::holdings::{ - announcement_for, reconcile_and_announce, signer_from_node_cert, AnnounceTransport, - HoldingsBroadcaster, HoldingsIngress, + announcement_for, reconcile_and_announce, run_first_peer_announcer, signer_from_node_cert, + AnnounceTransport, HoldingsBroadcaster, HoldingsIngress, HoldingsInventory, PoolPresence, }; /// The clock this test announces against — deliberately **real wall-clock**, unlike the pure-policy @@ -649,3 +649,166 @@ async fn an_unchanged_reconcile_floods_nothing() { "a no-op reconcile must not put a frame on the wire" ); } + +// ============================================================================================= +// The 0 -> N peer transition (#1734) — the ordering that made a holder invisible +// ============================================================================================= + +/// A fixed inventory, standing in for the node's cache list without a `Node` or a disk. +struct StubInventory(Vec); + +#[async_trait::async_trait] +impl HoldingsInventory for StubInventory { + async fn current(&self) -> Vec { + self.0.clone() + } +} + +/// PROPERTY (#1734): content pinned while the pool is EMPTY reaches the first peer that connects — +/// without an unpin/repin dance and without a restart. +/// +/// This is the P0 that made the live network show zero providers. The inventory-change reaction floods +/// only a NON-EMPTY delta diffed against the node's own local DHT records, and the pin at zero peers +/// moves those records — so the "announced" bookkeeping is already satisfied while nothing left the +/// box, and every later reconcile is a no-op. The node holds the capsule, believes it announced, and +/// is invisible to every peer. +/// +/// WHERE THIS ASSERTS, and why it must: at the RECEIVING node's ingest, over a real two-node mTLS +/// wire. The sender's `flooded an opcode-222 announcement` log line **already prints on the broken +/// path** (with `peers=0`), so any assertion keyed on the sender passes against the defect. Only a +/// frame a peer actually decoded, verified and ingested distinguishes the two implementations. +/// +/// The fixture drives the REAL ordering — pin at zero peers through the real `reconcile_and_announce` +/// with the real pool as the transport, THEN connect — because the whole defect is an ordering across +/// two nodes; a symmetric or mocked harness cannot see it. It also excludes the nearest wrong fix: an +/// announcer that re-runs the inventory RECONCILE on the peer edge computes an empty delta (the pin +/// already moved the records) and puts nothing on the wire, so this test reds for that variant too. +#[tokio::test] +async fn holdings_pinned_before_the_first_peer_reach_that_peer_when_it_connects() { + install_crypto_provider(); + let network = [0x5bu8; 32]; + let holder = WireNode::start([0x66u8; 32], network).await; + let receiver = WireNode::start([0x77u8; 32], network).await; + let serve_addr = dig_gossip::CandidateAddr { + host: "::1".to_string(), + port: 9_257, + }; + + // -- The trap's ordering, step one: PIN while the pool is empty ------------------------------ + let inventory = vec![cached_capsule(0x9a, 0x9b)]; + let capsule = ContentId::capsule([0x9au8; 32], [0x9bu8; 32]); + let dht = local_dht_handle(9_305); + let signer = signer_from_node_cert(&holder.cert).expect("the NodeCert leaf is ECDSA-P256"); + let broadcaster = Arc::new(HoldingsBroadcaster::new( + signer, + vec![serve_addr.clone()], + now_secs(), + )); + assert!( + holder.handle.connected_pool_peers().is_empty(), + "the defect's precondition: the pin happens with ZERO peers connected" + ); + let pinned = reconcile_and_announce( + &dht, + &inventory, + Some(( + broadcaster.as_ref(), + &holder.handle as &dyn AnnounceTransport, + )), + now_secs(), + ) + .await; + assert!( + pinned.gained.contains(&capsule), + "the pin must move the node's own durable provider records — that is what poisons the diff \ + every later reconcile is computed against; got {pinned:?}" + ); + + // -- Subscribe the receiver BEFORE the link exists, so no frame can be missed ---------------- + let mut inbound = receiver + .handle + .inbound_receiver() + .expect("a started service exposes its inbound receiver"); + + // -- Step two: the node runs its peer-presence announcer, and a peer arrives ----------------- + let announcer = tokio::spawn(run_first_peer_announcer( + holder.handle.clone(), + Arc::new(StubInventory(inventory)) as Arc, + Arc::clone(&broadcaster), + )); + holder + .handle + .connect_to(receiver.dial_addr()) + .await + .expect("the holder dials the receiver over loopback mTLS"); + + // -- The assertion the defect cannot satisfy: the RECEIVER got the announcement -------------- + let holder_id = holder.peer_id_hex(); + let (sender, decoded) = tokio::time::timeout(Duration::from_secs(20), async { + loop { + let (sender, msg) = inbound.recv().await.expect("inbound channel stays open"); + if let Some(a) = holdings_announce_payload(&msg) { + if a.provider_peer_id == holder_id { + break (sender, a); + } + } + } + }) + .await + .expect( + "a capsule pinned before the first peer MUST be announced once a peer connects — no frame \ + arrived, so this holder is invisible to the peer it is connected to", + ); + + let receiver_id = PeerId::from_hex(&receiver.peer_id_hex()).expect("64-hex peer id"); + let receiver_dht = Arc::new(DhtService::new( + receiver_id, + vec![CandidateAddr::direct("::1".to_string(), 9_306)], + DhtConfig::default(), + Arc::new(UnreachableTransport), + )); + let ingress = HoldingsIngress::new(receiver.peer_id_hex()); + let applied = ingress + .accept(&receiver_dht, &hex_of(&sender), &decoded, now_secs()) + .await + .expect("the announcement off the real wire is genuinely signed and accepted"); + assert!( + applied.ingested >= 1, + "the receiver must INGEST the holder's pinned inventory; got {applied:?}" + ); + let providers = receiver_dht + .find_providers(&capsule) + .await + .expect("a local provider-store hit needs no network"); + assert!( + providers.iter().any(|p| p.provider_peer_id == holder_id), + "the peer must be able to DISCOVER the holder of the capsule it pinned before connecting; \ + got {providers:?}" + ); + + announcer.abort(); +} + +/// PROPERTY: peers dropping to zero and returning re-announces, because that transition is the same +/// invisibility as the first one — a node whose only peer restarts must not go silently undiscovered. +/// +/// Asserted on the presence state machine rather than a wire, since the wire property above already +/// pins the announce itself; what is at stake here is only the EDGE definition. Both directions are +/// pinned: a rise fires, a further rise while peers are already present does NOT (that would re-flood +/// the whole inventory on every pool addition), and a fall to zero re-arms it. +#[test] +fn every_zero_to_nonzero_peer_transition_re_arms_the_announce() { + let mut presence = PoolPresence::default(); + + assert!(!presence.observe(0), "an empty pool announces nothing"); + assert!(presence.observe(1), "the first peer triggers the announce"); + assert!( + !presence.observe(3), + "growing an already-peered pool must not re-flood the whole inventory" + ); + assert!(!presence.observe(0), "losing every peer announces nothing"); + assert!( + presence.observe(1), + "peers returning after a total loss must re-announce — the node was invisible again" + ); +}