feat(blockexchange): implement delta WantList updates with batching

Implements delta-based WantList updates to reduce network traffic during
block exchange. Only sends newly added blocks instead of resending the
entire WantList on every refresh.

Also some network related fixes:

- Add TCP_NODELAY flag to prevent Nagle's algorithm delays
- Clear sendConn on stream reset to allow garbage collection
- Improve error handling in NetworkPeer.send()

Part of #974

Signed-off-by: Chrysostomos Nanakos <[email protected]>

Author: cnanakos

codex/blockexchange/engine/engine.nim

@@ -187,15 +187,65 @@ proc sendWantBlock(
 proc refreshBlockKnowledge(
     self: BlockExcEngine, peer: BlockExcPeerCtx
 ) {.async: (raises: [CancelledError]).} =
-  if self.pendingBlocks.wantListLen > 0:
-    # We send only blocks that the peer hasn't already told us that they already have.
-    let
-      peerHave = peer.peerHave
-      toAsk = self.pendingBlocks.wantList.toSeq.filterIt(it notin peerHave)
+  if self.pendingBlocks.wantListLen == 0:
+    if peer.lastSentWants.len > 0:
+      trace "Clearing want list tracking, no pending blocks", peer = peer.id
+      peer.lastSentWants.clear()
+    return
+
+  # We send only blocks that the peer hasn't already told us that they already have.
+  let
+    peerHave = peer.peerHave
+    toAsk = toHashSet(self.pendingBlocks.wantList.toSeq.filterIt(it notin peerHave))
+
+  if toAsk.len == 0:
+    if peer.lastSentWants.len > 0:
+      trace "Clearing want list tracking, peer has all blocks", peer = peer.id
+      peer.lastSentWants.clear()
+    return
+
+  let newWants = toAsk - peer.lastSentWants
+
+  if peer.lastSentWants.len > 0:
+    if newWants.len > 0:
+      trace "Sending delta want list update",
+        peer = peer.id, newWants = newWants.len, totalWants = toAsk.len
+
+      let newWantsSeq = newWants.toSeq
+      var offset = 0
+      while offset < newWantsSeq.len:
+        let batchEnd = min(offset + MaxWantListBatchSize, newWantsSeq.len)
+        let batch = newWantsSeq[offset ..< batchEnd]
+
+        trace "Sending want list batch",
+          peer = peer.id,
+          batchSize = batch.len,
+          offset = offset,
+          total = newWantsSeq.len
+
+        await self.network.request.sendWantList(peer.id, batch, full = false)
+        offset = batchEnd
+    else:
+      trace "No changes in want list, skipping send", peer = peer.id
+
+    peer.lastSentWants = toAsk
+  else:
+    trace "Sending full want list",
+      peer = peer.id, length = toAsk.len, reason = "first_send"
+
+    let toAskSeq = toAsk.toSeq
+    var offset = 0
+    while offset < toAskSeq.len:
+      let batchEnd = min(offset + MaxWantListBatchSize, toAskSeq.len)
+      let batch = toAskSeq[offset ..< batchEnd]
+
+      trace "Sending full want list batch",
+        peer = peer.id, batchSize = batch.len, offset = offset, total = toAskSeq.len
+
+      await self.network.request.sendWantList(peer.id, batch, full = (offset == 0))
+      offset = batchEnd
 
-    if toAsk.len > 0:
-      trace "Sending want list to a peer", peer = peer.id, length = toAsk.len
-      await self.network.request.sendWantList(peer.id, toAsk, full = true)
+    peer.lastSentWants = toAsk
 
 proc refreshBlockKnowledge(self: BlockExcEngine) {.async: (raises: [CancelledError]).} =
   let runtimeQuota = 10.milliseconds
@@ -215,9 +265,6 @@ proc refreshBlockKnowledge(self: BlockExcEngine) {.async: (raises: [CancelledErr
     if peer.isKnowledgeStale or peer.lastRefresh < self.pendingBlocks.lastInclusion:
       if not peer.refreshInProgress:
         peer.refreshRequested()
-        # TODO: optimize this by keeping track of what was sent and sending deltas.
-        #   This should allow us to run much more frequent refreshes, and be way more
-        #   efficient about it.
         await self.refreshBlockKnowledge(peer)
     else:
       trace "Not refreshing: peer is up to date", peer = peer.id

codex/blockexchange/network/networkpeer.nim

@@ -65,7 +65,9 @@ proc readLoop*(self: NetworkPeer, conn: Connection) {.async: (raises: []).} =
   except CatchableError as err:
     warn "Exception in blockexc read loop", msg = err.msg
   finally:
-    trace "Detaching read loop", peer = self.id, connId = conn.oid
+    warn "Detaching read loop", peer = self.id, connId = conn.oid
+    if self.sendConn == conn:
+      self.sendConn = nil
     await conn.close()
 
 proc connect*(
@@ -89,7 +91,12 @@ proc send*(
     return
 
   trace "Sending message", peer = self.id, connId = conn.oid
-  await conn.writeLp(protobufEncode(msg))
+  try:
+    await conn.writeLp(protobufEncode(msg))
+  except CatchableError as err:
+    if self.sendConn == conn:
+      self.sendConn = nil
+    raise newException(LPStreamError, "Failed to send message: " & err.msg)
 
 func new*(
     T: type NetworkPeer,

codex/blockexchange/peers/peercontext.nim

@@ -29,6 +29,7 @@ export payments, nitro
 const
   MinRefreshInterval = 1.seconds
   MaxRefreshBackoff = 36 # 36 seconds
+  MaxWantListBatchSize* = 1024 # Maximum blocks to send per WantList message
 
 type BlockExcPeerCtx* = ref object of RootObj
   id*: PeerId
@@ -44,6 +45,8 @@ type BlockExcPeerCtx* = ref object of RootObj
   blocksRequested*: HashSet[BlockAddress] # pending block requests to this peer
   lastExchange*: Moment # last time peer has sent us a block
   activityTimeout*: Duration
+  lastSentWants*: HashSet[BlockAddress]
+    # track what wantList we last sent for delta updates
 
 proc isKnowledgeStale*(self: BlockExcPeerCtx): bool =
   let staleness =

codex/codex.nim

@@ -211,7 +211,7 @@ proc new*(
     .withMaxConnections(config.maxPeers)
     .withAgentVersion(config.agentString)
     .withSignedPeerRecord(true)
-    .withTcpTransport({ServerFlags.ReuseAddr})
+    .withTcpTransport({ServerFlags.ReuseAddr, ServerFlags.TcpNoDelay})
     .build()
 
   var