feat: add block knowledge request mechanism, implement tests

Author: gmega

codex/blockexchange/engine/engine.nim

@@ -121,7 +121,6 @@ proc blockexcTaskRunner(self: BlockExcEngine) {.async: (raises: []).}
 proc start*(self: BlockExcEngine) {.async: (raises: []).} =
   ## Start the blockexc task
   ##
-
   await self.discovery.start()
   await self.advertiser.start()
 
@@ -200,11 +199,14 @@ proc refreshBlockKnowledge(self: BlockExcEngine) {.async: (raises: [CancelledErr
 
     # In dynamic swarms, staleness will dominate latency.
     if peer.lastRefresh < self.pendingBlocks.lastInclusion or peer.isKnowledgeStale:
+      trace "Refreshing block knowledge for peer", peer = peer.id
       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
 
 proc searchForNewPeers(self: BlockExcEngine, cid: Cid) =
   if self.lastDiscRequest + DiscoveryRateLimit < Moment.now():
@@ -336,10 +338,11 @@ proc requestBlocks*(
   for address in addresses:
     self.trackedFutures.track(self.downloadInternal(address))
 
-  var completed: int = 0
+  let totalHandles = handles.len
+  var completed = 0
 
   proc isFinished(): bool =
-    completed == handles.len
+    completed == totalHandles
 
   proc genNext(): Future[?!Block] {.async: (raises: [CancelledError]).} =
     # Be it success or failure, we're completing this future.

codex/blockexchange/engine/pendingblocks.nim

@@ -34,7 +34,7 @@ declareGauge(
 
 const
   DefaultBlockRetries* = 3000
-  DefaultRetryInterval* = 5.seconds
+  DefaultRetryInterval* = 1.seconds
 
 type
   RetriesExhaustedError* = object of CatchableError

codex/blockexchange/network/network.nim

@@ -341,7 +341,7 @@ proc handlePeerDeparted*(
   if not self.handlers.onPeerDeparted.isNil:
     await self.handlers.onPeerDeparted(peer)
 
-method init*(self: BlockExcNetwork) =
+method init*(self: BlockExcNetwork) {.raises: [].} =
   ## Perform protocol initialization
   ##
 

codex/blockexchange/peers/peercontext.nim

@@ -26,7 +26,7 @@ import ../../logutils
 export payments, nitro
 
 const
-  MinRefreshInterval = 5.seconds
+  MinRefreshInterval = 1.seconds
   MaxRefreshBackoff = 36 # 3 minutes
 
 type BlockExcPeerCtx* = ref object of RootObj

codex/stores/cachestore.nim

@@ -66,6 +66,21 @@ method getBlock*(
     trace "Error requesting block from cache", cid, error = exc.msg
     return failure exc
 
+method getBlocks*(
+    self: CacheStore, addresses: seq[BlockAddress]
+): Future[SafeAsyncIter[Block]] {.async: (raises: [CancelledError]).} =
+  var i = 0
+
+  proc isFinished(): bool =
+    i == addresses.len
+
+  proc genNext(): Future[?!Block] {.async: (raises: [CancelledError]).} =
+    let value = await self.getBlock(addresses[i])
+    inc(i)
+    return value
+
+  return SafeAsyncIter[Block].new(genNext, isFinished)
+
 method getCidAndProof*(
     self: CacheStore, treeCid: Cid, index: Natural
 ): Future[?!(Cid, CodexProof)] {.async: (raises: [CancelledError]).} =

tests/codex/blockexchange/discovery/testdiscovery.nim

@@ -54,7 +54,7 @@ asyncchecksuite "Block Advertising and Discovery":
     peerStore = PeerCtxStore.new()
     pendingBlocks = PendingBlocksManager.new()
 
-    (manifest, tree) = makeManifestAndTree(blocks).tryGet()
+    (_, tree, manifest) = makeDataset(blocks).tryGet()
     manifestBlock =
       bt.Block.new(manifest.encode().tryGet(), codec = ManifestCodec).tryGet()
 
@@ -172,7 +172,7 @@ asyncchecksuite "E2E - Multiple Nodes Discovery":
           break
 
         blocks.add(bt.Block.new(chunk).tryGet())
-      let (manifest, tree) = makeManifestAndTree(blocks).tryGet()
+      let (_, tree, manifest) = makeDataset(blocks).tryGet()
       manifests.add(manifest)
       mBlocks.add(manifest.asBlock())
       trees.add(tree)

tests/codex/blockexchange/discovery/testdiscoveryengine.nim

@@ -43,7 +43,7 @@ asyncchecksuite "Test Discovery Engine":
 
       blocks.add(bt.Block.new(chunk).tryGet())
 
-    (manifest, tree) = makeManifestAndTree(blocks).tryGet()
+    (_, tree, manifest) = makeDataset(blocks).tryGet()
     manifestBlock = manifest.asBlock()
     blocks.add(manifestBlock)
 

tests/codex/blockexchange/engine/testblockexc.nim

@@ -24,19 +24,12 @@ asyncchecksuite "NetworkStore engine - 2 nodes":
     pendingBlocks1, pendingBlocks2: seq[BlockHandle]
 
   setup:
-    blocks1 = await makeRandomBlocks(datasetSize = 2048, blockSize = 256'nb)
-    blocks2 = await makeRandomBlocks(datasetSize = 2048, blockSize = 256'nb)
+    blocks1 = makeRandomBlocks(nBlocks = 8, blockSize = 256'nb)
+    blocks2 = makeRandomBlocks(nBlocks = 8, blockSize = 256'nb)
     nodeCmps1 = generateNodes(1, blocks1)[0]
     nodeCmps2 = generateNodes(1, blocks2)[0]
 
-    await allFuturesThrowing(
-      nodeCmps1.switch.start(),
-      nodeCmps1.blockDiscovery.start(),
-      nodeCmps1.engine.start(),
-      nodeCmps2.switch.start(),
-      nodeCmps2.blockDiscovery.start(),
-      nodeCmps2.engine.start(),
-    )
+    await allFuturesThrowing(nodeCmps1.start(), nodeCmps2.start())
 
     # initialize our want lists
     pendingBlocks1 =
@@ -65,14 +58,7 @@ asyncchecksuite "NetworkStore engine - 2 nodes":
     check isNil(peerCtx2).not
 
   teardown:
-    await allFuturesThrowing(
-      nodeCmps1.blockDiscovery.stop(),
-      nodeCmps1.engine.stop(),
-      nodeCmps1.switch.stop(),
-      nodeCmps2.blockDiscovery.stop(),
-      nodeCmps2.engine.stop(),
-      nodeCmps2.switch.stop(),
-    )
+    await allFuturesThrowing(nodeCmps1.stop(), nodeCmps2.stop())
 
   test "Should exchange blocks on connect":
     await allFuturesThrowing(allFinished(pendingBlocks1)).wait(10.seconds)
@@ -145,7 +131,7 @@ asyncchecksuite "NetworkStore - multiple nodes":
     blocks: seq[bt.Block]
 
   setup:
-    blocks = await makeRandomBlocks(datasetSize = 4096, blockSize = 256'nb)
+    blocks = makeRandomBlocks(nBlocks = 16, blockSize = 256'nb)
     nodes = generateNodes(5)
     for e in nodes:
       await e.engine.start()
@@ -203,3 +189,28 @@ asyncchecksuite "NetworkStore - multiple nodes":
 
     check pendingBlocks1.mapIt(it.read) == blocks[0 .. 3]
     check pendingBlocks2.mapIt(it.read) == blocks[12 .. 15]
+
+asyncchecksuite "NetworkStore - dissemination":
+  var nodes: seq[NodesComponents]
+
+  teardown:
+    if nodes.len > 0:
+      await nodes.stop()
+
+  test "Should disseminate blocks across large diameter swarm":
+    let dataset = makeRandomDataset(nBlocks = 60, blockSize = 256'nb).tryGet()
+
+    nodes = generateNodes(6, enableDiscovery = false)
+
+    await assignBlocks(nodes[0], dataset, 0 .. 9)
+    await assignBlocks(nodes[1], dataset, 10 .. 19)
+    await assignBlocks(nodes[2], dataset, 20 .. 29)
+    await assignBlocks(nodes[3], dataset, 30 .. 39)
+    await assignBlocks(nodes[4], dataset, 40 .. 49)
+    await assignBlocks(nodes[5], dataset, 50 .. 59)
+
+    await nodes.start()
+    await nodes.linearTopology()
+
+    let downloads = nodes.mapIt(downloadDataset(it, dataset))
+    await allFuturesThrowing(downloads).wait(20.seconds)

tests/codex/helpers.nim

@@ -12,13 +12,16 @@ import pkg/codex/rng
 import pkg/codex/utils
 
 import ./helpers/nodeutils
+import ./helpers/datasetutils
 import ./helpers/randomchunker
 import ./helpers/mockchunker
 import ./helpers/mockdiscovery
 import ./helpers/always
 import ../checktest
 
-export randomchunker, nodeutils, mockdiscovery, mockchunker, always, checktest, manifest
+export
+  randomchunker, nodeutils, datasetutils, mockdiscovery, mockchunker, always, checktest,
+  manifest
 
 export libp2p except setup, eventually
 
@@ -46,23 +49,6 @@ proc lenPrefix*(msg: openArray[byte]): seq[byte] =
 
   return buf
 
-proc makeManifestAndTree*(blocks: seq[Block]): ?!(Manifest, CodexTree) =
-  if blocks.len == 0:
-    return failure("Blocks list was empty")
-
-  let
-    datasetSize = blocks.mapIt(it.data.len).foldl(a + b)
-    blockSize = blocks.mapIt(it.data.len).foldl(max(a, b))
-    tree = ?CodexTree.init(blocks.mapIt(it.cid))
-    treeCid = ?tree.rootCid
-    manifest = Manifest.new(
-      treeCid = treeCid,
-      blockSize = NBytes(blockSize),
-      datasetSize = NBytes(datasetSize),
-    )
-
-  return success((manifest, tree))
-
 proc makeWantList*(
     cids: seq[Cid],
     priority: int = 0,
@@ -91,7 +77,7 @@ proc storeDataGetManifest*(
     (await store.putBlock(blk)).tryGet()
 
   let
-    (manifest, tree) = makeManifestAndTree(blocks).tryGet()
+    (_, tree, manifest) = makeDataset(blocks).tryGet()
     treeCid = tree.rootCid.tryGet()
 
   for i in 0 ..< tree.leavesCount:
@@ -110,19 +96,6 @@ proc storeDataGetManifest*(
 
   return await storeDataGetManifest(store, blocks)
 
-proc makeRandomBlocks*(
-    datasetSize: int, blockSize: NBytes
-): Future[seq[Block]] {.async.} =
-  var chunker =
-    RandomChunker.new(Rng.instance(), size = datasetSize, chunkSize = blockSize)
-
-  while true:
-    let chunk = await chunker.getBytes()
-    if chunk.len <= 0:
-      break
-
-    result.add(Block.new(chunk).tryGet())
-
 proc corruptBlocks*(
     store: BlockStore, manifest: Manifest, blks, bytes: int
 ): Future[seq[int]] {.async.} =
@@ -147,4 +120,5 @@ proc corruptBlocks*(
 
       bytePos.add(ii)
       blk.data[ii] = byte 0
+
   return pos

tests/codex/helpers/datasetutils.nim

@@ -0,0 +1,35 @@
+import std/random
+
+import pkg/codex/blocktype as bt
+import pkg/codex/merkletree
+import pkg/codex/manifest
+
+type TestDataset* = tuple[blocks: seq[Block], tree: CodexTree, manifest: Manifest]
+
+proc makeRandomBlock*(size: NBytes): Block =
+  let bytes = newSeqWith(size.int, rand(uint8))
+  Block.new(bytes).tryGet()
+
+proc makeRandomBlocks*(nBlocks: int, blockSize: NBytes): seq[Block] =
+  for i in 0 ..< nBlocks:
+    result.add(makeRandomBlock(blockSize))
+
+proc makeDataset*(blocks: seq[Block]): ?!TestDataset =
+  if blocks.len == 0:
+    return failure("Blocks list was empty")
+
+  let
+    datasetSize = blocks.mapIt(it.data.len).foldl(a + b)
+    blockSize = blocks.mapIt(it.data.len).foldl(max(a, b))
+    tree = ?CodexTree.init(blocks.mapIt(it.cid))
+    treeCid = ?tree.rootCid
+    manifest = Manifest.new(
+      treeCid = treeCid,
+      blockSize = NBytes(blockSize),
+      datasetSize = NBytes(datasetSize),
+    )
+
+  return success((blocks, tree, manifest))
+
+proc makeRandomDataset*(nBlocks: int, blockSize: NBytes): ?!TestDataset =
+  makeDataset(makeRandomBlocks(nBlocks, blockSize))