chore(kad): liveness-gated bucket eviction

Author: gmelodie

libp2p/protocols/kademlia/find.nim

@@ -416,6 +416,8 @@ proc lookOnce*(
   for (peerId, res) in completedRPCBatch:
     let reply = res.valueOr:
       continue
+    # A successful reply proves the peer useful; retain it through eviction.
+    rtable.markUseful(peerId)
     let newPeerInfos = state.updateShortlist(reply)
     kad.switch.updatePeers(
       kad.config.addressPolicy, rtable, newPeerInfos, kad.config.limits.maxPeersPerIp,

libp2p/protocols/kademlia/routing_table.nim

@@ -19,6 +19,7 @@ proc new*(
     hasher: Opt[XorDHasher] = NoneHasher,
     maxBuckets: int = DefaultMaxBuckets,
     selfIdPreHashed = false,
+    usefulnessGracePeriod: Duration = DefaultUsefulnessGracePeriod,
 ): T =
   doAssert maxBuckets > 0 and maxBuckets <= MaxBucketsLimit,
     "maxBuckets must be in 1 .. " & $MaxBucketsLimit
@@ -27,6 +28,7 @@ proc new*(
     hasher: hasher,
     maxBuckets: maxBuckets,
     selfIdPreHashed: selfIdPreHashed,
+    usefulnessGracePeriod: usefulnessGracePeriod,
   )
 
 proc `$`*(rt: RoutingTable): string =
@@ -72,18 +74,42 @@ proc oldestPeer*(bucket: Bucket): (NodeEntry, int) =
       oldestIdx = i
   (oldest, oldestIdx)
 
-proc replaceOldest(bucket: var Bucket, newNodeId: Key, replication: int): bool =
-  if bucket.peers.len < replication:
+func isReplaceable*(entry: NodeEntry, gracePeriod: Duration, now: Moment): bool =
+  ## A peer is replaceable once it has spent longer than `gracePeriod` in the
+  ## table without proving useful. A peer that answered a query within the grace
+  ## period — or that was only just added — is retained.
+  now - entry.lastUsefulAt.get(entry.addedAt) > gracePeriod
+
+proc replaceableCandidate(bucket: Bucket, gracePeriod: Duration): Opt[int] =
+  ## Least-recently-seen peer that is past the usefulness grace period, i.e. the
+  ## best candidate to evict. `Opt.none` when every peer is still useful/fresh.
+  let now = Moment.now()
+  var candidateIdx = -1
+  var oldestSeen: Moment
+  for i, p in bucket.peers:
+    if not p.isReplaceable(gracePeriod, now):
+      continue
+    if candidateIdx == -1 or p.lastSeen < oldestSeen:
+      candidateIdx = i
+      oldestSeen = p.lastSeen
+  if candidateIdx == -1:
+    return Opt.none(int)
+  Opt.some(candidateIdx)
+
+proc replaceReplaceable(
+    bucket: var Bucket, newNodeId: Key, config: RoutingTableConfig
+): bool =
+  if bucket.peers.len < config.replication:
     trace "Skipping replace: bucket is not full", newNodeId = newNodeId
     return false
 
-  let (oldest, oldestIdx) = bucket.oldestPeer()
-
-  if oldest.nodeId == newNodeId:
-    trace "Failed to replace: same nodeId", newNodeId = newNodeId
+  let idx = bucket.replaceableCandidate(config.usefulnessGracePeriod).valueOr:
+    trace "Skipping replace: no peer past usefulness grace period",
+      newNodeId = newNodeId
     return false
 
-  bucket.peers[oldestIdx] = NodeEntry(nodeId: newNodeId, lastSeen: Moment.now())
+  let now = Moment.now()
+  bucket.peers[idx] = NodeEntry(nodeId: newNodeId, lastSeen: now, addedAt: now)
   true
 
 proc updateRoutingTableMetrics*(rtable: RoutingTable) =
@@ -113,12 +139,14 @@ proc insert*(rtable: RoutingTable, nodeId: Key): bool =
   if keyx.isSome:
     bucket.peers[keyx.unsafeValue].lastSeen = Moment.now()
   elif bucket.peers.len < rtable.config.replication:
-    bucket.peers.add(NodeEntry(nodeId: nodeId, lastSeen: Moment.now()))
+    let now = Moment.now()
+    bucket.peers.add(NodeEntry(nodeId: nodeId, lastSeen: now, addedAt: now))
     kad_routing_table_insertions.inc()
   else:
-    # eviction policy: replace oldest key
-    if not bucket.replaceOldest(nodeId, rtable.config.replication):
-      debug "Cannot insert, failed to replace oldest key in bucket",
+    # eviction policy: replace a peer past the usefulness grace period, keeping
+    # peers that recently answered a query.
+    if not bucket.replaceReplaceable(nodeId, rtable.config):
+      debug "Cannot insert, no replaceable peer in bucket",
         bucket = idx, nodeId = nodeId
       return false
     kad_routing_table_replacements.inc()
@@ -130,6 +158,24 @@ proc insert*(rtable: RoutingTable, nodeId: Key): bool =
 proc insert*(rtable: RoutingTable, peerId: PeerId): bool =
   insert(rtable, peerId.toKey())
 
+proc markUseful*(rtable: RoutingTable, nodeId: Key) =
+  ## Records that `nodeId` answered a query: refreshes its usefulness and
+  ## last-seen timestamps so it is retained through bucket eviction. No-op when
+  ## the peer is not in the table.
+  let idx = rtable.bucketIndex(nodeId)
+  if idx >= rtable.buckets.len:
+    return
+  var bucket = rtable.buckets[idx]
+  let pos = peerIndexInBucket(bucket, nodeId).valueOr:
+    return
+  let now = Moment.now()
+  bucket.peers[pos].lastUsefulAt = Opt.some(now)
+  bucket.peers[pos].lastSeen = now
+  rtable.buckets[idx] = bucket
+
+proc markUseful*(rtable: RoutingTable, peerId: PeerId) =
+  rtable.markUseful(peerId.toKey())
+
 proc findClosest*(rtable: RoutingTable, targetId: Key, count: int): seq[Key] =
   ## Returns up to `count` nodes in the table with the smallest XOR distance to `targetId`.
   var allNodes: seq[Key] = @[]

libp2p/protocols/kademlia/types.nim

@@ -18,6 +18,10 @@ const
   DefaultBucketRefreshTime* = 10.minutes
   DefaultBucketStaleTime* = 30.minutes
     # peer not seen for this duration marks bucket stale
+  DefaultUsefulnessGracePeriod* = 1.hours
+    ## A newly inserted peer is protected from eviction until it has been in the
+    ## table this long without proving useful (answering a query). Mirrors
+    ## go-libp2p's usefulnessGracePeriod.
   DefaultRetries* = 5
   DefaultReplication* = 20 ## aka `k` in the spec
   DefaultAlpha* = 10 # concurrency parameter
@@ -194,6 +198,11 @@ type
   NodeEntry* = object
     nodeId*: Key
     lastSeen*: Moment
+    addedAt*: Moment
+    lastUsefulAt*: Opt[Moment]
+      ## Set when the peer answers a query. `Opt.none` means it has not yet
+      ## proven useful; such a peer becomes replaceable once past the grace
+      ## period since it was added.
 
   Bucket* = object
     peers*: seq[NodeEntry]
@@ -203,6 +212,7 @@ type
     hasher*: Opt[XorDHasher]
     maxBuckets*: int
     selfIdPreHashed*: bool
+    usefulnessGracePeriod*: Duration
 
   RoutingTable* = ref object
     selfId*: Key

tests/libp2p/kademlia/test_routing_table.nim

@@ -45,7 +45,14 @@ suite "KadDHT Routing Table":
     let bucket = rt.buckets[TargetBucket]
     check bucket.peers.len <= config.replication
 
-  test "evicts oldest key at max capacity":
+  proc agePastGrace(rt: RoutingTable, bucketIdx: int) =
+    ## Push every peer past the usefulness grace period so eviction can act.
+    var bucket = rt.buckets[bucketIdx]
+    for i in 0 ..< bucket.peers.len:
+      bucket.peers[i].addedAt = Moment.now() - 2.hours
+    rt.buckets[bucketIdx] = bucket
+
+  test "evicts oldest replaceable key at max capacity":
     let selfId = testKey(0)
     let config = RoutingTableConfig.new(hasher = Opt.some(noOpHasher))
     var rt = RoutingTable.new(selfId, config)
@@ -55,15 +62,52 @@ suite "KadDHT Routing Table":
 
     check rt.buckets[TargetBucket].peers.len == config.replication
 
+    # Fresh peers are protected by the grace period; age them so they are
+    # eligible for eviction.
+    rt.agePastGrace(TargetBucket)
+
     # new entry should evict oldest entry
     let (oldest, _) = rt.buckets[TargetBucket].oldestPeer()
 
     check rt.insert(randomKeyInBucket(selfId, TargetBucket, rng()))
 
-    let (oldestAfterInsert, _) = rt.buckets[TargetBucket].oldestPeer()
+    check:
+      rt.buckets[TargetBucket].peers.len == config.replication
+      not rt.buckets[TargetBucket].allKeys().contains(oldest.nodeId)
+
+  test "retains peers still within the usefulness grace period":
+    let selfId = testKey(0)
+    let config = RoutingTableConfig.new(hasher = Opt.some(noOpHasher))
+    var rt = RoutingTable.new(selfId, config)
+    for _ in 0 ..< config.replication:
+      discard rt.insert(randomKeyInBucket(selfId, TargetBucket, rng()))
+
+    let before = rt.buckets[TargetBucket].allKeys()
+    let newcomer = randomKeyInBucket(selfId, TargetBucket, rng())
+
+    # Bucket is full of fresh, unproven peers: the newcomer is rejected rather
+    # than blindly evicting one of them.
+    check:
+      not rt.insert(newcomer)
+      rt.buckets[TargetBucket].allKeys() == before
+      not rt.buckets[TargetBucket].allKeys().contains(newcomer)
 
-    # oldest was evicted
-    check oldest.nodeId != oldestAfterInsert.nodeId
+  test "markUseful protects a peer from eviction":
+    let selfId = testKey(0)
+    let config = RoutingTableConfig.new(hasher = Opt.some(noOpHasher))
+    var rt = RoutingTable.new(selfId, config)
+    for _ in 0 ..< config.replication:
+      discard rt.insert(randomKeyInBucket(selfId, TargetBucket, rng()))
+
+    rt.agePastGrace(TargetBucket)
+
+    # The oldest-seen peer would be evicted first, but marking it useful keeps
+    # it and forces a different peer out.
+    let (oldest, _) = rt.buckets[TargetBucket].oldestPeer()
+    rt.markUseful(oldest.nodeId)
+
+    check rt.insert(randomKeyInBucket(selfId, TargetBucket, rng()))
+    check rt.buckets[TargetBucket].allKeys().contains(oldest.nodeId)
 
   test "re-insert existing key updates lastSeen":
     let selfId = testKey(0)