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" + ); +}