Kademlia DHT Spec Compliance & Bug Fixes - #1426
Open
sumanjeet0012 wants to merge 44 commits into
Open
Conversation
…, and ADD_PROVIDER fire-and-forget
- Fix self-exclusion in DHT walk: exclude local peer from peers_to_query
to prevent hanging when walk discovers own ID from remote peer
- Fix CID key encoding in provide/find_providers: use cid_to_bytes(parse_cid())
instead of key.encode('utf-8') to produce correct 36-byte raw CID multihash
- Fix ADD_PROVIDER to be fire-and-forget per IPFS DHT spec: send message
and close stream without waiting for response (Kubo doesn't respond)
…ling - kad_dht.py:708: break → continue in ADD_PROVIDER; skip invalid providers instead of aborting batch - kad_dht.py:953: PUT_VALUE key validation compares message.key != record.key (was comparing record.key to itself) - provider_store.py:331: Re-raise trio.Cancelled in find_providers to allow proper cancellation propagation - routing_table.py:238: Remove misleading 'Successfully refreshed peer' log that fired on ping failure - utils.py:62: maybe_consume_signed_record now skips peer_id check when peer_id is None instead of always returning False
- get_value: use find_closest_peers_network instead of find_local_closest_peers so values at peers outside local routing table are found (kad_dht.py:1163) - ADD_PROVIDER: remove overly restrictive len(key) > 80 check, only reject empty keys per go-libp2p spec (kad_dht.py:654) - peer_routing: add QUERY_TIMEOUT to _query_peer_for_closest to prevent hanging on unresponsive peers (peer_routing.py:302)
…VIDER >80 check - find_peer: re-check peerstore after network lookup (matches go-libp2p) - Add max varint byte limit (10 bytes) to _get_from_peer, _store_at_peer, _query_peer_for_closest, _get_providers_from_peer to prevent DoS - Restore len(key) > 80 check in ADD_PROVIDER handler (matches go-libp2p)
Implements critical and high-priority fixes to align the Kademlia DHT implementation with the libp2p Kademlia specification (r2, 2022-12-09). Critical fixes: - timeReceived now uses RFC3339 format per spec (was Unix epoch float) - Added format_time_rfc3339() and parse_time_received() helpers - Backward compatible: parses both RFC3339 and legacy Unix epoch - Fixes cross-implementation incompatibility with go-libp2p High-priority fixes: - GET_VALUE record age/expiry validation now parses RFC3339 timestamps - clean_record() prevents timestamp forgery by stripping timeReceived - PUT_VALUE uses validator.Select comparison per spec - PUT_VALUE record size validation (1MB max) - Provider address validation rejects reserved/private IPs (RFC 6890) - ADD_PROVIDER rate limiting (10 msgs per peer per 10s window) - ADD_PROVIDER per-message provider record cap (20) - Value store LRU eviction at 50K entries prevents memory exhaustion Medium-priority fixes: - Stream errors now use reset() instead of close() per spec - Beta resiliency (β=3) in peer lookup termination - Total query timeout (30s) prevents infinite loops - IP diversity filtering (max 2 peers per /16 subnet) - Connection type dynamic reporting (CONNECTED/CAN_CONNECT/NOT_CONNECTED) - Peer removal on connection failure and invalid signed records - Empty key validation for GET_VALUE, PUT_VALUE, GET_PROVIDERS Low-priority fixes: - Bootstrap interval changed from 1min to 10min (spec default) - PUT_VALUE only sends response on success (not on failure) - Added is_cid_like_key() for CID structure validation - Added is_reserved_or_private_addr() for comprehensive IP filtering Tests: 108 passed, 0 failed
Per Kademlia spec for value retrieval algorithm: GAP 6 - Pb/Po tracking: - Track Pb (peers that returned the best value) and Po (peers with outdated values) sets separately during value retrieval - When a new value is better than current best, old best peers move to Po - Entry correction now only propagates to Po peers (not all peers with different values), matching spec behavior GAP 7 - Cancel on quorum: - When quorum is reached, cancel all outstanding queries using trio nursery cancel_scope.cancel() - Previously let in-flight queries complete for robustness, but spec says to cancel when quorum is met - This ensures faster return once enough answers are collected Tests: 108 passed
- Improve CID validation with multihash structure checks (GAP 5+9) - Add sliding window PUT_VALUE propagation via _propagate_to_closest_peers (GAP 11) - Make provider discovery iterative with closer peers (GAP 16) - Add record republishing (GAP 14) - Fix test for async _get_providers_from_peer_with_closers
- ADD_PROVIDER: echo response per spec (was fire-and-forget) - ADD_PROVIDER: key length limit 80 bytes (was 128) - Provider address TTL: 24h (was 30min) per spec Section 7.3.3 - GET_VALUE: use closer peers from responses for iterative lookup - Stream reset on errors (was graceful close) per spec Section 9 - PUT_VALUE without record: reset stream (was sending empty response)
- FIND_NODE: validate key is a valid multihash (PeerId format) - GET_VALUE: add 128-byte max key size validation - ProviderStore: add JSON persistence via persist_dir parameter - ValueStore: add JSON persistence via persist_dir parameter - Use pathlib.Path for cross-platform path handling
- get_value: re-sort candidates by distance after discovering closer peers - PUT_VALUE: reject record when validator.select() fails (was storing anyway) - Routing table refresh: generate random key within bucket's XOR range - Provider addresses: return addresses for valid records (was empty after TTL) - Provider store: limit providers per key to k=20 - FIND_NODE: accept both raw multihash and reasonable-length PeerIds - GET_VALUE: validate record signature before serving - Fix contradictory ADD_PROVIDER comments
…closer peer address storage - peer_routing: continue lookup when unqueried closest peers remain even if no new peers discovered in a round - provider_store: break provider lookup only when no candidates remain instead of when a batch returns empty - kad_dht: use BUCKET_SIZE constant instead of hardcoded 20 in FIND_NODE, GET_PROVIDERS, GET_VALUE responses - kad_dht: set should_reset on PUT_VALUE storage failure and validation failure to comply with spec stream reset requirement - value_store: store closer peer addresses in peerbook per spec when processing GET_VALUE responses
…e transfers - Store None for block_bytes in leaf_triples after putting block in blockstore - balanced_layout() only uses CID and file_data_size from leaves, ignores block_bytes - Fixes O(n) memory growth where peak memory was ~8x file size for large files - Handles single-leaf edge case: retrieve root_data from blockstore when None - Fixes balanced_layout to handle None block_bytes in single-leaf and multi-leaf cases - Tested with 2GB file: peak memory reduced from ~16GB to 103MB (0.05x file size)
…r, add public ensure_peer_stream API - Peer registration is no longer one-shot: a failed new_stream is retried with capped exponential backoff while the peer stays connected (_handle_new_peer_with_retry). This fixes the race where the 'connected' notifee fires before the muxer handshake completes, leaving the peer silently unregistered and messaging dead. - _handle_dead_peer now re-establishes the pubsub stream when the stream dies but the peer still has active connections (e.g. mDNS auto-connect racing an explicit dial produces multiple simultaneous connections; the stream on a broken one dies while a healthy one remains). The peer is no longer silently dropped from pubsub. - Add public Pubsub.ensure_peer_stream(peer_id, timeout) so applications that connect out-of-band (connect_peer reusing an existing connection fires no new notifee) can (re)register a peer on demand. - Add PeerDiscovery.unregister_peer_discovered_handler() so consumers can clean up handlers on the module-level singleton. - Re-export the pubsub public API (Pubsub, GossipSub, PROTOCOL_ID*) from libp2p.pubsub via lazy PEP-562 module __getattr__ (eager imports cause a circular import during package init). - Tests: add regression tests for retry-on-failed-stream, ensure_peer_stream, and stream-failure-does-not-unregister-connected-peer.
The global max_connections limit was only checked on the inbound path, so a node could exceed its configured cap purely through outbound dials (concurrent app dials, DHT queries, auto-connector). Add the same admission check at the top of _dial_addr_single_attempt so outbound dials fail fast before opening a socket. Regression tests added.
…on (Bug 3) When the same IMuxedConn was added twice, add_conn closed the duplicate wrapper, which closed the underlying muxed_conn shared with the existing connection - tearing down the live connection it was about to return. Fix: dedup early (before scope/task setup) and mark late-race duplicates as _shared_muxed_conn so SwarmConn.close() skips muxed_conn.close(). The stream-monitor loop now also exits once the wrapper is closed.
…ion (Bug 8) The inbound max_connections check was check-then-act: concurrent handshakes each read the connection count before any registered, so a burst could overshoot the cap. Enforce the global cap atomically in add_conn at registration time (append + check are contiguous, no awaits between), closing and rejecting the connection that pushes the count over the limit. The pre-upgrade inbound check remains as a fast-fail heuristic.
…(Bug 5) maybe_prune_connections ran synchronously inside add_conn (the dial/accept hot path). Above the high watermark every new connection triggered an O(n log n) sort of all connections plus per-connection closes, each of which sleeps 100ms in SwarmConn._cleanup - stalling connection establishment for seconds. Pruning now runs fire-and-forget via manager.run_task with a 1s debounce and a concurrent-run guard.
The auto-connector was only driven by a 30s periodic task, so after disconnects dropped the connection count below the low watermark the node could sit under its floor for up to 30s. notify_disconnected now schedules AutoConnector.maybe_connect() as a background task (5s cooldown, cheap no-op above the watermark), matching go-libp2p's prompt refill behavior.
…ctions (Bug 7) A raising notifee propagated out of notify_connected/notify_disconnected/notify_opened_stream and into add_conn's except BaseException, closing a successfully established connection (and aborting teardown in _cleanup). All notify_* methods now fan out each notifee in its own task with exceptions caught and logged, so a broken or slow notifee can no longer break connection setup/teardown.
…ly (Bug 9) When a manager was present, close() only called manager.stop(), cancelling connection-monitor tasks without closing the connections - leaking rcmgr resource scopes and leaving sockets for GC. close() now closes all active connections (best-effort), then listeners and transports, then stops the manager. A _closing flag prevents shutdown disconnects from triggering the auto-connector to dial new peers.
…ug 6 follow-up) The disconnect-triggered auto-connect (Bug 6) re-dialed the peer that just disconnected, causing an immediate reconnect loop whenever a connection dropped below the low watermark. AutoConnector now records recent disconnects and skips those peers for 60s, matching go-libp2p's backoff behavior; a successful connection clears the backoff.
The happy-eyeballs cancel in dial_peer is asynchronous, so several concurrent dials could all succeed before cancellation landed, exceeding max_connections_per_peer. dial_peer now caps the returned connections at the per-peer limit and closes the excess.
_trim_connections closed the oldest connections with no safeguards (no grace period, no protection check, no stream awareness) and via untracked trio.lowlevel.spawn_system_task calls that could be dropped on shutdown. It now skips connections within the grace period, never trims protected peers, trims fewest-active-streams/oldest first, and schedules closes through the swarm manager (tracked, cancellable).
…ug 10) Peers whose addresses all failed were blacklisted for 300s with no way to lift the block, so transient failures (closed port, NAT, gate reconfiguration, peerstore updates) blocked dials for minutes. TTL reduced to 60s, Swarm.unblock_peer() added, the cache is cleared when a peer identity-mismatch clears the peerstore addresses, and clear() resets the whole cache.
get_connections(peer_id) returned the internal list and get_connections_map returned a shallow copy whose inner lists were shared, so callers could mutate the swarm's connection tracking. Both now return copies with fresh lists.
…P (Bug 14) is_connection_in_allow_list consulted the peer's peerstore addresses, so a connection made over a non-allow-listed IP could be exempted from pruning just because the peer had an allow-listed address on record. It now uses the connection's actual remote address (via the muxed connection), building the multiaddr with the correct IPv4/IPv6 protocol; also adds the previously-missing ipaddress import.
min_connections was dead config (only used in a log line). It now drives behavior: when the connection count drops below the critical floor, the auto-connector's periodic task polls at a 5s critical interval instead of the 30s auto_connect_interval, so the node recovers from a critically-low state promptly.
new_resource_manager() and ProductionConfig defaulted enable_connection_pooling=True, so every ResourceManager allocated a ConnectionPool whose acquire()/release() are never called anywhere. Defaults flipped to False (opt-in only) and the ENABLE_CONNECTION_POOLING env default changed to false; tests cover the new defaults.
…ction limits are enforced (Bug 1) The Rust-style connection limits (max_pending_*, max_established_inbound/outbound, max_established_per_peer, max_established_total) were never enforced: ConnectionLifecycleManager was never instantiated and its handlers never called. ResourceManager now builds the lifecycle manager (tracker seeded with the configured limits), Swarm.add_conn admits each connection through the per-direction/per-peer/total handlers (rejecting with a clear error when a limit is exceeded) and Swarm.remove_conn decrements the tracker. The QUIC listener now passes direction='inbound' to add_conn so the directional accounting is correct.
Apply ruff check --fix (import sorting, unused import removal) and ruff format to the files touched by the connection-manager fixes.
- swarm.py: guard get_remote_address with hasattr (IMuxedConn has no such declared method) and hoist background_nursery into a local so the None-check narrows for pyrefly. - connection_pruner.py: use getattr+callable instead of hasattr so both type checkers accept the dynamic attribute access, and validate the returned remote address shape. - regression tests: replace nonlocal closure counters with lists (pyrefly [unknown-name]) and use setattr for the dynamic _shared_muxed_conn flag. Remaining pyrefly errors (6 in bitswap/dag.py) and mypy errors (pubsub/bitswap) are pre-existing on this branch and unrelated to these fixes.
…ate notifee docs, cache cap hardening Addresses code-review feedback on the connection-manager fixes:\n- remove_conn no longer decrements the lifecycle tracker for duplicate wrappers from the add_conn race (Bug 3 + Bug 1 interaction): the duplicate shares the same muxed_conn and therefore the same tracker connection_id, so its close would remove the slot still held by the surviving connection. Regression test locks in admit -> dedup-race -> close balancing to zero.\n- _notify docstring now states accurately that exceptions are isolated but a blocking notifee still stalls the caller (matching go-libp2p's inline callback semantics).\n- _NegativePeerCache.mark_failed evicts the oldest entry when full and nothing is expired, so the cache cannot exceed max_size (Bug 10 hardening).
read() previously blocked until the stream closed or the buffer emptied, even when data was already available — an EOF-gated semantic that diverged from read_stream() and could deadlock: the reader waits for close while the writer waits for the reader to consume data and respond. read() now returns any available data immediately, checks reset/EOF only when no data is pending, and the interleaving-EOF regression test accumulates across multiple reads until EOF.
…tics read() now returns available data without waiting for stream close, so the test asserts data arrives promptly and accumulates across reads until StreamEOF.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Kademlia DHT Spec Compliance & Bug Fixes
Closes #1425
Summary
This PR implements comprehensive spec compliance fixes and bug fixes for the Kademlia DHT module, bringing the implementation closer to the IPFS DHT specification and libp2p Kademlia spec.
Changes
Critical Fixes
timeReceivedfield now uses RFC3339Nano format per spec (was Unix epoch float), with backward-compatible parsingtimeReceivedfrom incoming records to prevent timestamp forgerySpec Compliance
Provider System
persist_dirparameterValue Store
persist_dirparameter_propagate_to_closest_peers()Peer Routing
Security
Bug Fixes
20withBUCKET_SIZEconstantvalidator.select()failsFiles Changed
Testing