fix(holdings): re-announce holdings when the peer pool rises from zero (#1734)

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

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:

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."

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<crate::Node>);
+
+#[async_trait::async_trait]
+impl crate::seams::dig_peer::holdings::HoldingsInventory for NodeHoldingsInventory {
+    async fn current(&self) -> Vec<crate::CachedCapsule> {
+        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<crate::Node>) -> 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<dyn crate::seams::dig_peer::holdings::HoldingsInventory>,
+            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 →

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<crate::CachedCapsule>;
+}
+
+/// 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<dyn HoldingsInventory>,
+    broadcaster: Arc<HoldingsBroadcaster>,
+) {
+    // 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