Skip to content

fix(node): make peer bring-up honest and add a peer-ping ladder diagnostic #149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
MichaelTaylor3d wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

fix(node): make peer bring-up honest and add a peer-ping ladder diagnostic #149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
68c857d
chore: open #1974 lane
MichaelTaylor3d Aug 2, 2026
9c4c4fb
fix(peer): take the inventory announce off the bring-up critical path
MichaelTaylor3d Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 18 additions & 4 deletions crates/dig-node-core/src/peer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2400,7 +2400,13 @@ async fn run_peer_network(node: Arc<crate::Node>) -> Result<(), String> {
// and find_providers finds nobody. This forwards every PoolEvent (add → insert, remove →
// evict) into routing so cross-node discovery works as the pool fills.
if let Some(dht) = dht.clone() {
spawn_dht_routing_feed(dht, handle.clone());
spawn_dht_routing_feed(dht.clone(), handle.clone());
// 4a-ii. Publish this node's local inventory into the DHT — in the BACKGROUND, and only now
// that routing is being fed from the live pool (#1974). Two reasons for this position:
// bring-up must not wait on it (the listener bind is still ahead of us), and every
// announce run before the feed exists queries the empty table bootstrap left behind,
// so it times out instead of reaching the peers that would store the record.
crate::dht::spawn_initial_inventory_announce(dht);
}

// 4b. Bring up the P2P CONTENT engine (#164/#165) over the live DHT + this node's mTLS identity: the
Expand Down Expand Up @@ -2666,12 +2672,20 @@ async fn bring_up_dht(
tracing::debug!(error = %e, "DHT bootstrap found no peers yet; records republish once the pool fills");
}

// Announce the node's CURRENT inventory so peers can immediately find the content it holds.
// Derive the node's CURRENT inventory and the content ids it provides, and record them on the
// handle so every later reconcile diffs against the truth from the first moment.
//
// The ANNOUNCE itself — one Kademlia lookup + PUT per id, each RPC bounded by the DHT timeout —
// is deliberately NOT run here. Awaiting it on the bring-up path is what left the mTLS peer-RPC
// listener (bound near the end of `run_peer_network`) unbound for 12m40s on a node holding 44
// capsules: undialable and undiscoverable, yet holding a relay reservation that advertised it as
// up, and logging nothing for the whole window (dig_ecosystem#1974). The caller starts it in the
// background via `spawn_initial_inventory_announce` once the pool→routing feed is live.
let cached = node.cache_list_cached().await;
let announced = crate::dht::announce_inventory(&service, &cached).await;
let initial_ids = crate::dht::inventory_content_ids(&cached);
println!(
"dig-node peer network: DHT up — announced {announced} content id(s) for local inventory"
"dig-node peer network: DHT up — {} content id(s) to announce for local inventory",
initial_ids.len()
);

let dht = crate::dht::DhtHandle::new(service, initial_ids);
Expand Down
104 changes: 60 additions & 44 deletions crates/dig-node-core/src/seams/capsule/capsule_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,59 @@ use digstore_core::Bytes32;

use crate::{module_exists, CachedCapsule, Node, PeerNetwork};

/// Walk `<modules_root>/<store_id_hex>/<root_hex>.dig` and describe every capsule this node holds.
///
/// BLOCKING (`read_dir` plus one `stat` per entry) — drive it from a blocking thread, never from an
/// async worker: on a network-mounted cache each of those is a round trip. It reads only directory
/// metadata (name, size, mtime) and never opens a capsule's bytes, so its cost is exactly one `stat`
/// per held generation.
fn list_cached_capsules(modules_root: &std::path::Path) -> Vec<CachedCapsule> {
let mut out = Vec::new();
// Outer level: one directory per store id (hex). Inner: `<root>.dig` (or a legacy `<root>.module`).
let Ok(stores) = std::fs::read_dir(modules_root) else {
return out; // no modules cached yet
};
for store_entry in stores.flatten() {
if !store_entry.path().is_dir() {
continue;
}
let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else {
continue;
};
let Ok(modules) = std::fs::read_dir(store_entry.path()) else {
continue;
};
for m in modules.flatten() {
let path = m.path();
// A capsule module is `<root_hex>.dig` (or a legacy `<root_hex>.module` a prior binary
// wrote — #1896); either names a held capsule, so stripping BOTH suffixes from one
// authority is what keeps a legacy holder discoverable through the upgrade.
let Some(root_hex) = path
.file_name()
.and_then(|f| f.to_str())
.and_then(crate::capsule_key::cached_root_stem)
.map(str::to_string)
else {
continue;
};
let Ok(md) = m.metadata() else { continue };
let last_used_unix_ms = md
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
out.push(CachedCapsule {
store_id: store_hex.clone(),
root: root_hex,
size_bytes: md.len(),
last_used_unix_ms,
});
}
}
out
}

/// Seam 6 (capsule management) — the node's on-disk `.dig` capsule cache: list/remove/fetch a held
/// capsule, gap-fill a missing chain-confirmed generation, and the self-reference plumbing that lets
/// `&self` read handlers spawn an owned background backfill.
Expand Down Expand Up @@ -134,51 +187,14 @@ pub trait CapsuleStore: Send + Sync {
#[async_trait::async_trait]
impl CapsuleStore for Node {
async fn cache_list_cached(&self) -> Vec<CachedCapsule> {
// Handed to a blocking thread rather than run inline: this is `std::fs` `read_dir` + `stat`
// per held capsule, and on a node whose cache is a NETWORK mount (the S3-backed store, #1943)
// every one of those is a round trip. Running them on an async worker parks a runtime thread
// for the duration, stalling whatever else was scheduled onto it (dig_ecosystem#1974).
let modules_root = self.cache_dir.join("modules");
let mut out = Vec::new();
// Outer level: one directory per store id (hex). Inner: `<root>.dig` (or a legacy `<root>.module`).
let Ok(stores) = std::fs::read_dir(&modules_root) else {
return out; // no modules cached yet
};
for store_entry in stores.flatten() {
if !store_entry.path().is_dir() {
continue;
}
let Some(store_hex) = store_entry.file_name().to_str().map(str::to_string) else {
continue;
};
let Ok(modules) = std::fs::read_dir(store_entry.path()) else {
continue;
};
for m in modules.flatten() {
let path = m.path();
// A capsule module is `<root_hex>.dig` (or a legacy `<root_hex>.module` a prior binary
// wrote — #1896); either names a held capsule, so stripping BOTH suffixes from one
// authority is what keeps a legacy holder discoverable through the upgrade.
let Some(root_hex) = path
.file_name()
.and_then(|f| f.to_str())
.and_then(crate::capsule_key::cached_root_stem)
.map(str::to_string)
else {
continue;
};
let Ok(md) = m.metadata() else { continue };
let last_used_unix_ms = md
.modified()
.ok()
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
out.push(CachedCapsule {
store_id: store_hex.clone(),
root: root_hex,
size_bytes: md.len(),
last_used_unix_ms,
});
}
}
out
tokio::task::spawn_blocking(move || list_cached_capsules(&modules_root))
.await
.unwrap_or_default()
}

async fn cache_remove_cached(
Expand Down
Loading
Loading