fix: implement standard Kademlia DHT maintenance#60
Conversation
The DHT routing table was severely underpopulated because several standard Kademlia mechanisms were missing. ant-node testing showed nodes discovering only 13-17% of the network via DHT despite being transport-connected to ~90%, causing close group instability that breaks merkle payment verification. Changes: 1. K=8 to K=20: Bucket size was far below the standard Kademlia and libp2p value of 20. More peers per distance range means iterative lookups have better starting candidates and converge faster. 2. Stale-node eviction: When a k-bucket is full, the least-recently- seen node is checked for liveness instead of silently rejecting the new node. Dead peers get evicted and replaced. Per Kademlia Section 2.2: "contacts that have been around longer are more likely to remain." 3. Replacement cache: Nodes that cannot be inserted into full buckets (because the LRS node is alive) are held in a per-bucket cache (max 5). When a node is removed for any reason, the most recently seen replacement is promoted. 4. Periodic bucket refresh: Background task checks every 5 minutes and refreshes any bucket not looked up in the last hour by performing a find_closest_nodes with a random key in the bucket's XOR distance range. Per Kademlia Section 2.4. 5. Add unknown senders on message receipt: Per Kademlia Section 2.1, "when a node receives any message from another node, it updates the appropriate k-bucket." Previously only touched existing entries; now unknown senders are added (subject to diversity checks and bucket capacity).
There was a problem hiding this comment.
Pull request overview
Implements missing Kademlia maintenance mechanisms in the DHT to improve routing-table completeness and stabilize “closest K nodes” selection across the network (needed for merkle payment close-group stability in ant-node).
Changes:
- Adds periodic k-bucket refresh scheduling and executes random-key
FIND_NODElookups for stale buckets. - Implements full-bucket insert handling with least-recently-seen (LRS) eviction flow plus a per-bucket replacement cache.
- Updates routing-table maintenance to also add previously-unknown message senders on receipt, and increases core K-bucket size from 8 to 20.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 10 comments.
| File | Description |
|---|---|
| src/dht_network_manager.rs | Starts a background bucket-refresh task and updates routing-table population/eviction behavior on message receipt and peer connect. |
| src/dht/core_engine.rs | Updates routing table internals for K=20, introduces replacement cache + bucket refresh metadata, and exposes APIs used by the manager refresh task. |
| Err(_) => { | ||
| // Diversity check failed — expected, no action needed |
There was a problem hiding this comment.
Swallowing all add_node errors here makes it hard to diagnose unexpected failures (e.g., non-diversity errors). At least log the error at debug/trace level, and consider only silencing the specific expected diversity-rejection cases.
| Err(_) => { | |
| // Diversity check failed — expected, no action needed | |
| Err(err) => { | |
| // Typically indicates a diversity check failure; log for diagnostics. | |
| debug!( | |
| "Failed to add message sender {} to routing table: {:?}", | |
| app_peer_id.to_hex(), | |
| err | |
| ); |
| /// Return indices of buckets that haven't been looked up within `max_age` | ||
| /// and have at least one node (empty buckets don't need refresh). | ||
| fn stale_bucket_indices(&self, max_age: std::time::Duration) -> Vec<usize> { | ||
| self.buckets | ||
| .iter() | ||
| .enumerate() | ||
| .filter(|(_, b)| !b.nodes.is_empty() && b.last_lookup.elapsed() > max_age) |
There was a problem hiding this comment.
stale_bucket_indices excludes empty buckets from refresh. In standard Kademlia bucket refresh, empty buckets can still need refresh (a random-ID lookup is exactly how you discover peers in that range). Skipping empty buckets may leave the table permanently underpopulated in those ranges.
| /// Return indices of buckets that haven't been looked up within `max_age` | |
| /// and have at least one node (empty buckets don't need refresh). | |
| fn stale_bucket_indices(&self, max_age: std::time::Duration) -> Vec<usize> { | |
| self.buckets | |
| .iter() | |
| .enumerate() | |
| .filter(|(_, b)| !b.nodes.is_empty() && b.last_lookup.elapsed() > max_age) | |
| /// Return indices of buckets that haven't been looked up within `max_age`. | |
| /// All buckets (including empty ones) may need refresh to discover peers. | |
| fn stale_bucket_indices(&self, max_age: std::time::Duration) -> Vec<usize> { | |
| self.buckets | |
| .iter() | |
| .enumerate() | |
| .filter(|(_, b)| b.last_lookup.elapsed() > max_age) |
| /// Find nodes closest to a key | ||
| pub async fn find_nodes(&self, key: &DhtKey, count: usize) -> Result<Vec<NodeInfo>> { | ||
| let routing = self.routing_table.read().await; | ||
| let mut routing = self.routing_table.write().await; |
There was a problem hiding this comment.
find_nodes now takes a write lock on the routing table just to update last_lookup, which reduces concurrency for read-only lookups and may become a contention point under load. Consider storing per-bucket last_lookup in a separate, cheaper-to-update structure (e.g., atomics or a small mutex per bucket) so lookups can keep using a read lock for routing-table access.
| let mut routing = self.routing_table.write().await; | |
| let routing = self.routing_table.read().await; |
| /// Add a node to the DHT with security checks. | ||
| /// | ||
| /// Diversity limits (IP subnet and geographic region) are derived from the | ||
| /// live routing table contents on every call, so counts are always accurate | ||
| /// regardless of evictions, reconnections, or address changes. | ||
| pub async fn add_node(&mut self, node: NodeInfo) -> Result<()> { | ||
| /// Add a node to the DHT routing table with diversity checks. | ||
| /// |
There was a problem hiding this comment.
The doc comment for add_node looks duplicated (it has both “Add a node to the DHT with security checks.” and “Add a node to the DHT routing table with diversity checks.”). Consider consolidating to a single description to avoid confusion.
| // Wait for initial bootstrap to complete before starting refresh | ||
| tokio::time::sleep(std::time::Duration::from_secs(60)).await; |
There was a problem hiding this comment.
The refresh task won’t observe shutdown cancellation during the initial 60s sleep, so stop() can return while this task is still pending and may run a refresh cycle later. Consider wrapping the initial delay in a tokio::select! on shutdown.cancelled() (or using an interval that is immediately cancellable).
| // Wait for initial bootstrap to complete before starting refresh | |
| tokio::time::sleep(std::time::Duration::from_secs(60)).await; | |
| // Wait for initial bootstrap to complete before starting refresh, | |
| // but allow early exit if shutdown is requested. | |
| tokio::select! { | |
| () = shutdown.cancelled() => { | |
| return; | |
| } | |
| _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => { | |
| // Initial delay elapsed; proceed to periodic refresh. | |
| } | |
| } |
| // Request 20 closest nodes (K value) | ||
| if let Err(e) = self_arc | ||
| .find_closest_nodes_network( | ||
| key.as_bytes(), | ||
| 20, | ||
| ) |
There was a problem hiding this comment.
This hard-codes 20 as the K value, but other parts of the DHT manager still use a separate closest-nodes constant. To avoid inconsistencies, use a single source of truth (e.g., the DHT core’s K constant/config) instead of embedding 20 here.
| for bucket_idx in stale { | ||
| let key = { | ||
| let dht = self_arc.dht.read().await; | ||
| dht.random_key_for_bucket_refresh(bucket_idx).await | ||
| }; | ||
|
|
There was a problem hiding this comment.
Refreshing every stale bucket in one loop can trigger a large burst of network lookups (potentially up to 256) in a single tick, which can add load and interfere with normal operations. Consider capping the number of buckets refreshed per interval and/or spreading work over time with bounded concurrency.
| tokio::spawn(async move { | ||
| let lrs_alive = transport.is_peer_connected(&lrs_id).await; | ||
| let dht_guard = dht.write().await; | ||
| if lrs_alive { | ||
| dht_guard.add_to_replacement_cache(pending_node).await; |
There was a problem hiding this comment.
The bucket-full handling claims to “ping” the least-recently-seen node, but it only checks transport.is_peer_connected. Connection status is not a liveness probe in Kademlia (the node could be alive but currently disconnected), and this can cause incorrect evictions/caching decisions. Use an actual ping RPC (and optionally dial first) with a timeout to decide whether to evict or keep the LRS node.
| let lrs_alive = transport | ||
| .is_peer_connected(&lrs_id) | ||
| .await; | ||
| let dht_guard = dht.write().await; | ||
| if lrs_alive { | ||
| // LRS node is still connected — keep it, cache the new node | ||
| dht_guard.add_to_replacement_cache(pending_node).await; |
There was a problem hiding this comment.
Same issue as above: this code path says it is “pinging LRS”, but it only checks whether the peer is currently connected. That’s not equivalent to a Kademlia ping/liveness check and can lead to evicting reachable-but-disconnected nodes. Consider performing a real DHT ping (and dialing if needed) before deciding whether to evict or cache.
| key_bytes[byte_idx] ^= 1 << bit_idx; | ||
| // Randomize remaining bits (after the prefix) for diversity |
There was a problem hiding this comment.
random_key_in_bucket only randomizes bytes after byte_idx + 1, leaving the remaining bits in the same byte as the flipped bucket bit unchanged. That reduces randomness within the bucket range and can bias refresh lookups. Consider randomizing all bits after bucket_index (including the lower bits of key_bytes[byte_idx]) while preserving the prefix bits before bucket_index.
| key_bytes[byte_idx] ^= 1 << bit_idx; | |
| // Randomize remaining bits (after the prefix) for diversity | |
| // First flip the bucket bit | |
| let flipped = key_bytes[byte_idx] ^ (1 << bit_idx); | |
| // Preserve prefix bits (more significant than `bucket_index`), fix the bucket bit, | |
| // and randomize all less-significant bits (including those in the same byte). | |
| let high_mask: u8 = (0xFFu8 << (bit_idx + 1)) & 0xFF; | |
| let low_mask: u8 = !high_mask; | |
| let bucket_mask: u8 = 1 << bit_idx; | |
| let rand_low: u8 = rand::random(); | |
| let preserved_high = flipped & high_mask; | |
| let bucket_bit = flipped & bucket_mask; | |
| let randomized_low = rand_low & (low_mask & !bucket_mask); | |
| key_bytes[byte_idx] = preserved_high | bucket_bit | randomized_low; | |
| // Randomize all subsequent bytes for diversity within the bucket range |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| pub async fn find_nodes(&self, key: &DhtKey, count: usize) -> Result<Vec<NodeInfo>> { | ||
| let routing = self.routing_table.read().await; | ||
| let mut routing = self.routing_table.write().await; | ||
| Ok(routing.find_closest_nodes(key, count)) |
There was a problem hiding this comment.
find_nodes downgraded to write-lock on a hot read path
find_closest_nodes now mutates last_lookup on every call, forcing find_nodes to acquire a routing_table.write() lock. Previously this was a read() lock. Because tokio::sync::RwLock gives writers priority over readers, this change serialises all concurrent DHT lookups — any code that calls find_nodes in parallel (e.g. iterative network walks, bucket-refresh loops, and the find_closest_nodes_network fan-out) will queue behind a single writer instead of running concurrently.
last_lookup is only needed for the once-per-hour refresh heuristic. Consider storing it as an AtomicU64 (epoch milliseconds) inside KBucket, which allows find_closest_nodes to stay a &self function and find_nodes to keep its read() lock:
use std::sync::atomic::{AtomicU64, Ordering};
struct KBucket {
nodes: Vec<NodeInfo>,
max_size: usize,
replacement_cache: Vec<NodeInfo>,
last_lookup_ms: AtomicU64, // replaces `last_lookup: Instant`
}Then the staleness check becomes a simple Ordering::Relaxed load, and find_closest_nodes can be &self again, restoring the read lock in find_nodes.
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/dht/core_engine.rs
Line: 547-549
Comment:
**`find_nodes` downgraded to write-lock on a hot read path**
`find_closest_nodes` now mutates `last_lookup` on every call, forcing `find_nodes` to acquire a `routing_table.write()` lock. Previously this was a `read()` lock. Because `tokio::sync::RwLock` gives writers priority over readers, this change serialises all concurrent DHT lookups — any code that calls `find_nodes` in parallel (e.g. iterative network walks, bucket-refresh loops, and the `find_closest_nodes_network` fan-out) will queue behind a single writer instead of running concurrently.
`last_lookup` is only needed for the once-per-hour refresh heuristic. Consider storing it as an `AtomicU64` (epoch milliseconds) inside `KBucket`, which allows `find_closest_nodes` to stay a `&self` function and `find_nodes` to keep its `read()` lock:
```rust
use std::sync::atomic::{AtomicU64, Ordering};
struct KBucket {
nodes: Vec<NodeInfo>,
max_size: usize,
replacement_cache: Vec<NodeInfo>,
last_lookup_ms: AtomicU64, // replaces `last_lookup: Instant`
}
```
Then the staleness check becomes a simple `Ordering::Relaxed` load, and `find_closest_nodes` can be `&self` again, restoring the read lock in `find_nodes`.
How can I resolve this? If you propose a fix, please make it concise.|
|
||
| for bucket_idx in stale { | ||
| let key = { | ||
| let dht = self_arc.dht.read().await; | ||
| dht.random_key_for_bucket_refresh(bucket_idx).await | ||
| }; | ||
|
|
||
| // Perform a find_node lookup for the random key. | ||
| // The lookup itself populates the routing table | ||
| // as a side effect of the iterative walk. | ||
| // Request 20 closest nodes (K value) | ||
| if let Err(e) = self_arc | ||
| .find_closest_nodes_network( | ||
| key.as_bytes(), | ||
| 20, | ||
| ) | ||
| .await | ||
| { | ||
| tracing::debug!( | ||
| "Bucket refresh lookup for bucket {bucket_idx} failed: {e}" | ||
| ); | ||
| } |
There was a problem hiding this comment.
random_key_for_bucket_refresh calls routing.mark_bucket_lookup(bucket_index) before the find_closest_nodes_network call. If the network lookup fails (timeout, no peers, etc.), last_lookup has already been set to Instant::now(), so the bucket is treated as freshly refreshed and won't be retried for another hour — defeating the purpose of the refresh.
Move mark_bucket_lookup to after a successful lookup, or split the helper so the key is generated without updating the timestamp:
// generate key (does NOT update last_lookup)
let key = {
let routing = self_arc.dht.read().await.routing_table.read().await;
routing.random_key_in_bucket(bucket_idx)
};
// only mark refreshed if lookup succeeds
if self_arc.find_closest_nodes_network(key.as_bytes(), 20).await.is_ok() {
let dht = self_arc.dht.read().await;
dht.mark_bucket_lookup(bucket_idx).await;
}Prompt To Fix With AI
This is a comment left during a code review.
Path: src/dht_network_manager.rs
Line: 440-461
Comment:
**Bucket timestamp updated before the refresh lookup succeeds**
`random_key_for_bucket_refresh` calls `routing.mark_bucket_lookup(bucket_index)` _before_ the `find_closest_nodes_network` call. If the network lookup fails (timeout, no peers, etc.), `last_lookup` has already been set to `Instant::now()`, so the bucket is treated as freshly refreshed and won't be retried for another hour — defeating the purpose of the refresh.
Move `mark_bucket_lookup` to after a successful lookup, or split the helper so the key is generated without updating the timestamp:
```rust
// generate key (does NOT update last_lookup)
let key = {
let routing = self_arc.dht.read().await.routing_table.read().await;
routing.random_key_in_bucket(bucket_idx)
};
// only mark refreshed if lookup succeeds
if self_arc.find_closest_nodes_network(key.as_bytes(), 20).await.is_ok() {
let dht = self_arc.dht.read().await;
dht.mark_bucket_lookup(bucket_idx).await;
}
```
How can I resolve this? If you propose a fix, please make it concise.| .find_closest_nodes_network( | ||
| key.as_bytes(), | ||
| 20, | ||
| ) |
There was a problem hiding this comment.
20 should reference the K constant
The count passed to find_closest_nodes_network is hardcoded as 20. If K is ever changed again, this won't track it automatically. Reference the constant instead (requires making K pub(crate) or re-exporting it).
Prompt To Fix With AI
This is a comment left during a code review.
Path: src/dht_network_manager.rs
Line: 452-455
Comment:
**Hardcoded `20` should reference the `K` constant**
The count passed to `find_closest_nodes_network` is hardcoded as `20`. If `K` is ever changed again, this won't track it automatically. Reference the constant instead (requires making `K` `pub(crate)` or re-exporting it).
How can I resolve this? If you propose a fix, please make it concise.| fn random_key_in_bucket(&self, bucket_index: usize) -> DhtKey { | ||
| let mut key_bytes = *self.node_id.to_bytes(); | ||
| // Flip the bit at `bucket_index` to ensure the key is in the bucket's range | ||
| let byte_idx = bucket_index / 8; | ||
| let bit_idx = 7 - (bucket_index % 8); | ||
| key_bytes[byte_idx] ^= 1 << bit_idx; | ||
| // Randomize remaining bits (after the prefix) for diversity | ||
| for byte in key_bytes.iter_mut().skip(byte_idx + 1) { | ||
| *byte = rand::random(); | ||
| } | ||
| DhtKey::from_bytes(key_bytes) |
There was a problem hiding this comment.
random_key_in_bucket skips lower bits of the pivot byte
After XOR-flipping the bit at bit_idx, the code randomizes all bytes after byte_idx but leaves the bits below bit_idx in byte_idx unchanged (they retain the local node ID's value). For bucket 1 (second MSB), bits 0–6 of byte 0 are always fixed, artificially constraining key diversity for that bucket.
The bits below the pivot bit in the same byte should also be randomized:
let lower_mask: u8 = (1u8 << bit_idx).wrapping_sub(1);
let random_low: u8 = rand::random();
key_bytes[byte_idx] = (key_bytes[byte_idx] & !lower_mask) | (random_low & lower_mask);Prompt To Fix With AI
This is a comment left during a code review.
Path: src/dht/core_engine.rs
Line: 383-393
Comment:
**`random_key_in_bucket` skips lower bits of the pivot byte**
After XOR-flipping the bit at `bit_idx`, the code randomizes all bytes after `byte_idx` but leaves the bits below `bit_idx` in `byte_idx` unchanged (they retain the local node ID's value). For bucket 1 (second MSB), bits 0–6 of byte 0 are always fixed, artificially constraining key diversity for that bucket.
The bits below the pivot bit in the same byte should also be randomized:
```rust
let lower_mask: u8 = (1u8 << bit_idx).wrapping_sub(1);
let random_low: u8 = rand::random();
key_bytes[byte_idx] = (key_bytes[byte_idx] & !lower_mask) | (random_low & lower_mask);
```
How can I resolve this? If you propose a fix, please make it concise.
|
Closing for now, @mickvandijke feel free to re-open |
Summary
Problem
ant-node e2e testing (PR WithAutonomi/ant-node#45) measured:
This causes different nodes to return different "closest K nodes" for the same address, breaking merkle payment verification where client (quoting) and node (verification) must agree on the close group.
Changes
Test plan
cargo buildcleanGreptile Summary
This PR implements five standard Kademlia DHT maintenance mechanisms that were missing from the routing table, directly addressing severe network under-discovery (13–17% DHT connectivity vs. 90%+ transport connectivity). The changes are well-motivated, clearly documented with references to the Kademlia paper, and the core data structures (
BucketInsertResult, replacement cache,last_lookup) are implemented correctly.\n\nKey changes:\n-Kbumped from 8 → 20 (libp2p/Kademlia standard), matching CLAUDE.md config docs\n- Stale-node eviction: full buckets now ping the LRS node instead of silently rejecting new nodes\n- Per-bucket replacement cache (5 slots, FIFO eviction) correctly promoted onremove_node\n- Periodic bucket refresh task (5 min interval, 1 hr stale threshold) with 60 s initial delay\n- Unknown message senders are now added to the routing table (Kademlia Section 2.1)\n\nIssues found:\n-find_nodeswas changed fromrouting_table.read()torouting_table.write()becausefind_closest_nodesnow mutateslast_lookup. This serialises all DHT lookups behind a single write lock — a meaningful performance regression on a very hot path\n- The bucket refresh task stampslast_lookup = Instant::now()before the network lookup runs; a failed lookup silently suppresses retry for ~1 hour\n-find_closest_nodes_networkis called with a hardcoded20instead of theKconstant\n- A stale doc-comment fragment was left at the top ofupdate_peer_info's doc blockConfidence Score: 3/5
The functional Kademlia logic is sound, but converting
find_nodesto a write lock introduces a serialisation bottleneck on the hottest path in the DHT that is likely to regress throughput under load.Two P1 issues exist: the write-lock regression on
find_nodes(impacts every concurrent DHT lookup) and the prematurelast_lookupupdate that suppresses bucket refresh retries for up to an hour after a failed lookup. Neither causes data loss or a security issue, but the lock contention issue is a production reliability concern at scale. The correctness of the core Kademlia data structures (eviction, replacement cache, sender-add) is clean. Resolving the two P1s would move this to 4–5.src/dht/core_engine.rs(write-lock regression infind_nodes,random_key_in_bucketpivot-byte coverage) andsrc/dht_network_manager.rs(prematurelast_lookupstamp in bucket refresh)Important Files Changed
find_nodesnow requiring a write lock on every DHT lookup due tolast_lookupmutation infind_closest_nodes.Sequence Diagram
sequenceDiagram participant NM as DhtNetworkManager participant DHT as DhtCoreEngine participant RT as KademliaRoutingTable participant Net as Network (other peers) Note over NM: On message receipt (Kademlia §2.1) NM->>DHT: touch_node(sender_id) [read lock on dht] DHT->>RT: touch_node → move to tail [write lock on routing_table] alt node already in table RT-->>NM: true → return early else node unknown NM->>DHT: add_node(sender_info) [write lock on dht] DHT->>RT: add_node [write lock on routing_table] alt bucket has space RT-->>NM: BucketInsertResult::Inserted else bucket full RT-->>NM: BucketFull { lrs_node_id, pending_node } NM->>Net: is_peer_connected(lrs_id) [spawned task] alt LRS alive NM->>DHT: add_to_replacement_cache(pending_node) else LRS gone NM->>DHT: evict_and_insert(lrs_id, pending_node) DHT->>RT: promote_replacement from cache end end end Note over NM: Periodic bucket refresh (every 5 min) loop every 5 minutes NM->>DHT: stale_bucket_indices(1h) [read lock] DHT->>RT: check last_lookup per bucket [read lock] RT-->>NM: stale bucket indices loop for each stale bucket NM->>DHT: random_key_for_bucket_refresh(idx) [write lock] DHT->>RT: mark_bucket_lookup + random_key_in_bucket RT-->>NM: DhtKey NM->>Net: find_closest_nodes_network(key, 20) Net-->>NM: closest nodes (routing table populated as side-effect) end endComments Outside Diff (1)
src/dht_network_manager.rs, line 1638-1651 (link)update_peer_infoThe function's doc block contains a fragment of the old comment ("routing table entry to move it to the tail of its k-bucket…") immediately followed by the new Kademlia Section 2.1 comment. The old fragment was not removed, leaving a run-on sentence at the top of the doc block. The stale lines should be deleted so only the new description remains.
Prompt To Fix With AI
Prompt To Fix All With AI
Reviews (1): Last reviewed commit: "fix: implement standard Kademlia DHT mai..." | Re-trigger Greptile