Cleanup engine and rework discovery (#87)

* rework discovery with async queues

* misc style changes

* increase max message size for large manifests

* use upraises and avoid exceptions on key access

* increase sleep time to 100 millis

* pass config

* make list blocks trigger a callback on each block

* check for nil on start/stop

* fix tests and split out discovery tests

* don't auto mount network

* add list block tests

* add discovery tests

* rework moc discovery

* move discovery moc to disc dir

* don't force logging syncs

* don't force moc discovery on all tests

* rework discovery with methods

* add top level utils file

* don't use asyncCheck

* don't pass entire blocks to list blocks calback

* spelling

config.nims

@@ -68,10 +68,6 @@ switch("warning", "ObservableStores:off")
 switch("warning", "LockLevel:off")
 
 switch("define", "libp2p_pki_schemes=secp256k1")
-#TODO this infects everything in this folder, ideally it would only
-# apply to dagger.nim, but since dagger.nims is used for other purpose
-# we can't use it. And dagger.cfg doesn't work
-switch("define", "chronicles_sinks=textlines[dynamic],json[dynamic]")
 
 # begin Nimble config (version 1)
 when system.fileExists("nimble.paths"):

dagger/blockexchange/engine.nim

@@ -7,7 +7,8 @@
 ## This file may not be copied, modified, or distributed except according to
 ## those terms.
 
-import std/[sequtils, sets, tables, sugar]
+import std/sequtils
+import std/sets
 
 import pkg/chronos
 import pkg/chronicles
@@ -15,7 +16,7 @@ import pkg/libp2p
 
 import ../stores/blockstore
 import ../blocktype as bt
-import ../utils/asyncheapqueue
+import ../utils
 import ../discovery
 
 import ./protobuf/blockexc
@@ -32,31 +33,19 @@ logScope:
   topics = "dagger blockexc engine"
 
 const
-  DefaultBlockTimeout* = 5.minutes
   DefaultMaxPeersPerRequest* = 10
   DefaultTaskQueueSize = 100
   DefaultConcurrentTasks = 10
   DefaultMaxRetries = 3
-
+  DefaultConcurrentDiscRequests = 10
-  # Current advertisement is meant to be more efficient than
+  DefaultConcurrentAdvertRequests = 10
-  # correct, so blocks could be advertised more slowly than that
+  DefaultDiscoveryTimeout = 1.minutes
-  # Put some margin
+  DefaultMaxQueriedBlocksCache = 1000
-  BlockAdvertisementFrequency = 30.minutes
 
 type
   TaskHandler* = proc(task: BlockExcPeerCtx): Future[void] {.gcsafe.}
   TaskScheduler* = proc(task: BlockExcPeerCtx): bool {.gcsafe.}
 
-  BlockDiscovery* = ref object
-    discoveredProvider: AsyncEvent
-    discoveryLoop: Future[void]
-    toDiscover: Cid
-    treatedPeer: HashSet[PeerId]
-    inflightIWant: HashSet[PeerId]
-    gotIWantResponse: AsyncEvent
-    provides: seq[PeerId]
-    lastDhtQuery: Moment
-
   BlockExcEngine* = ref object of RootObj
     localStore*: BlockStore                       # where we localStore blocks for this instance
     network*: BlockExcNetwork                     # network interface
@@ -70,12 +59,15 @@ type
     peersPerRequest: int                          # max number of peers to request from
     wallet*: WalletRef                            # nitro wallet for micropayments
     pricing*: ?Pricing                            # optional bandwidth pricing
-    advertisedBlocks: seq[Cid]
+    discovery*: Discovery                         # Discovery interface
-    advertisedIndex: int
+    concurrentAdvReqs: int                        # Concurrent advertise requests
-    advertisementFrequency: Duration
+    advertiseLoop*: Future[void]                  # Advertise loop task handle
-    runningDiscoveries*: Table[Cid, BlockDiscovery]
+    advertiseQueue*: AsyncQueue[Cid]              # Advertise queue
-    blockAdded: AsyncEvent
+    advertiseTasks*: seq[Future[void]]            # Advertise tasks
-    discovery*: Discovery
+    concurrentDiscReqs: int                       # Concurrent discovery requests
+    discoveryLoop*: Future[void]                  # Discovery loop task handle
+    discoveryTasks*: seq[Future[void]]            # Discovery tasks
+    discoveryQueue*: AsyncQueue[Cid]              # Discovery queue
 
   Pricing* = object
     address*: EthAddress
@@ -100,7 +92,76 @@ proc scheduleTask(b: BlockExcEngine, task: BlockExcPeerCtx): bool {.gcsafe} =
   b.taskQueue.pushOrUpdateNoWait(task).isOk()
 
 proc blockexcTaskRunner(b: BlockExcEngine): Future[void] {.gcsafe.}
-proc advertiseLoop(b: BlockExcEngine): Future[void] {.gcsafe.}
+
+proc discoveryLoopRunner(b: BlockExcEngine) {.async.} =
+  while b.blockexcRunning:
+    for cid in toSeq(b.pendingBlocks.wantList):
+      try:
+        await b.discoveryQueue.put(cid)
+      except CatchableError as exc:
+        trace "Exception in discovery loop", exc = exc.msg
+
+    trace "About to sleep, number of wanted blocks", wanted = b.pendingBlocks.len
+    await sleepAsync(30.seconds)
+
+proc advertiseLoopRunner*(b: BlockExcEngine) {.async.} =
+  proc onBlock(cid: Cid) {.async.} =
+    try:
+      await b.advertiseQueue.put(cid)
+    except CatchableError as exc:
+      trace "Exception listing blocks", exc = exc.msg
+
+  while b.blockexcRunning:
+    await b.localStore.listBlocks(onBlock)
+    await sleepAsync(30.seconds)
+
+  trace "Exiting advertise task loop"
+
+proc advertiseTaskRunner(b: BlockExcEngine) {.async.} =
+  ## Run advertise tasks
+  ##
+
+  while b.blockexcRunning:
+    try:
+      let cid = await b.advertiseQueue.get()
+      await b.discovery.provideBlock(cid)
+    except CatchableError as exc:
+      trace "Exception in advertise task runner", exc = exc.msg
+
+  trace "Exiting advertise task runner"
+
+proc discoveryTaskRunner(b: BlockExcEngine) {.async.} =
+  ## Run discovery tasks
+  ##
+
+  while b.blockexcRunning:
+    try:
+      let
+        cid = await b.discoveryQueue.get()
+        providers = await b.discovery
+          .findBlockProviders(cid)
+          .wait(DefaultDiscoveryTimeout)
+
+      await allFuturesThrowing(
+        allFinished(providers.mapIt( b.network.dialPeer(it.data) )))
+    except CatchableError as exc:
+      trace "Exception in discovery task runner", exc = exc.msg
+
+  trace "Exiting discovery task runner"
+
+proc queueFindBlocksReq(b: BlockExcEngine, cids: seq[Cid]) {.async.} =
+  try:
+    for cid in cids:
+      await b.discoveryQueue.put(cid)
+  except CatchableError as exc:
+    trace "Exception queueing discovery request", exc = exc.msg
+
+proc queueProvideBlocksReq(b: BlockExcEngine, cids: seq[Cid]) {.async.} =
+  try:
+    for cid in cids:
+      await b.advertiseQueue.put(cid)
+  except CatchableError as exc:
+    trace "Exception queueing discovery request", exc = exc.msg
 
 proc start*(b: BlockExcEngine) {.async.} =
   ## Start the blockexc task
@@ -116,13 +177,14 @@ proc start*(b: BlockExcEngine) {.async.} =
   for i in 0..<b.concurrentTasks:
     b.blockexcTasks.add(blockexcTaskRunner(b))
 
-  info "Getting existing block list"
+  for i in 0..<b.concurrentAdvReqs:
-  let blocks = await b.localStore.blockList()
+    b.advertiseTasks.add(advertiseTaskRunner(b))
-  b.advertisedBlocks = blocks
-  # We start faster to publish everything ASAP
-  b.advertisementFrequency = 5.seconds
 
-  b.blockexcTasks.add(b.advertiseLoop())
+  for i in 0..<b.concurrentDiscReqs:
+    b.discoveryTasks.add(discoveryTaskRunner(b))
+
+  b.advertiseLoop = advertiseLoopRunner(b)
+  b.discoveryLoop = discoveryLoopRunner(b)
 
 proc stop*(b: BlockExcEngine) {.async.} =
   ## Stop the blockexc blockexc
@@ -140,156 +202,87 @@ proc stop*(b: BlockExcEngine) {.async.} =
       await t.cancelAndWait()
       trace "Task stopped"
 
-  for _, bd in b.runningDiscoveries:
+  for t in b.advertiseTasks:
-    await bd.discoveryLoop.cancelAndWait()
+    if not t.finished:
+      trace "Awaiting task to stop"
+      await t.cancelAndWait()
+      trace "Task stopped"
 
-  b.runningDiscoveries.clear()
+  for t in b.discoveryTasks:
+    if not t.finished:
+      trace "Awaiting task to stop"
+      await t.cancelAndWait()
+      trace "Task stopped"
+
+  if not b.advertiseLoop.isNil and not b.advertiseLoop.finished:
+    trace "Awaiting advertise loop to stop"
+    await b.advertiseLoop.cancelAndWait()
+    trace "Advertise loop stopped"
+
+  if not b.discoveryLoop.isNil and not b.discoveryLoop.finished:
+    trace "Awaiting discovery loop to stop"
+    await b.discoveryLoop.cancelAndWait()
+    trace "Discovery loop stopped"
 
   trace "NetworkStore stopped"
 
-proc discoverOnDht(b: BlockExcEngine, bd: BlockDiscovery) {.async.} =
-  bd.lastDhtQuery = Moment.fromNow(10.hours)
-  defer: bd.lastDhtQuery = Moment.now()
-
-  let discoveredProviders = await b.discovery.findBlockProviders(bd.toDiscover)
-
-  for peer in discoveredProviders:
-    asyncSpawn b.network.dialPeer(peer.data)
-
-proc discoverLoop(b: BlockExcEngine, bd: BlockDiscovery) {.async.} =
-  # First, try connected peers
-  # After a percent of peers declined, or a timeout passed, query DHT
-  # rinse & repeat
-  #
-  # TODO add a global timeout
-
-  debug "starting block discovery", cid=bd.toDiscover
-
-  bd.gotIWantResponse.fire()
-  while true:
-    # wait for iwant replies
-    await bd.gotIWantResponse.wait()
-    bd.gotIWantResponse.clear()
-
-    var foundPeerNew = false
-    for p in b.peers:
-      if bd.toDiscover in p.peerHave and p.id notin bd.treatedPeer:
-        bd.provides.add(p.id)
-        bd.treatedPeer.incl(p.id)
-        bd.inflightIWant.excl(p.id)
-        foundPeerNew = true
-
-    if foundPeerNew:
-      bd.discoveredProvider.fire()
-      continue
-
-    trace "asking peers", cid=bd.toDiscover, peers=b.peers.len, treated=bd.treatedPeer.len, inflight=bd.inflightIWant.len
-    for p in b.peers:
-      if p.id notin bd.treatedPeer and p.id notin bd.inflightIWant:
-        # just send wants
-        bd.inflightIWant.incl(p.id)
-        b.network.request.sendWantList(
-          p.id,
-          @[bd.toDiscover],
-          wantType = WantType.wantHave,
-          sendDontHave = true)
-
-    if bd.inflightIWant.len < 3 and #TODO or a timeout
-      bd.lastDhtQuery < Moment.now() - 5.seconds:
-        #start query
-        asyncSpawn b.discoverOnDht(bd)
-
-proc discoverBlock*(b: BlockExcEngine, cid: Cid): BlockDiscovery =
-  if cid in b.runningDiscoveries:
-    return b.runningDiscoveries[cid]
-  else:
-    result = BlockDiscovery(
-      toDiscover: cid,
-      discoveredProvider: newAsyncEvent(),
-      gotIWantResponse: newAsyncEvent(),
-    )
-    result.discoveryLoop = b.discoverLoop(result)
-    b.runningDiscoveries[cid] = result
-    return result
-
-proc stopDiscovery(b: BlockExcEngine, cid: Cid) =
-  if cid in b.runningDiscoveries:
-    b.runningDiscoveries[cid].discoveryLoop.cancel()
-    b.runningDiscoveries.del(cid)
-
 proc requestBlock*(
   b: BlockExcEngine,
   cid: Cid,
-  timeout = DefaultBlockTimeout): Future[bt.Block] {.async.} =
+  timeout = DefaultBlockTimeout): Future[bt.Block] =
   ## Request a block from remotes
   ##
 
-  debug "requesting block", cid
-
-  # TODO
-  # we could optimize "groups of related chunks"
-  # be requesting multiple chunks, and running discovery
-  # less often
-
-  if cid in b.localStore:
-    return (await b.localStore.getBlock(cid)).get()
-
-  # be careful, don't give back control to main loop here
-  # otherwise, the block might slip in
-
-  if cid in b.pendingBlocks:
-    return await b.pendingBlocks.blocks[cid].wait(timeout)
-
-  # We are the first one to request this block, so we handle it
   let
-    timeoutFut = sleepAsync(timeout)
+    blk = b.pendingBlocks.getWantHandle(cid, timeout)
-    blk = b.pendingBlocks.addOrAwait(cid)
-    discovery = b.discoverBlock(cid)
 
-  # Just take the first discovered peer
+  if b.peers.len <= 0:
-  try:
+    trace "No peers to request blocks from", cid = $cid
-    await timeoutFut or blk or discovery.discoveredProvider.wait()
+    asyncSpawn b.queueFindBlocksReq(@[cid])
-    discovery.discoveredProvider.clear()
+    return blk
-  except CancelledError as exc:
-    #TODO also wrong, same issue as below
-    blk.cancel()
-    b.stopDiscovery(cid)
-    raise exc
 
-  if timeoutFut.finished:
+  var peers = b.peers
-    # TODO this is wrong, because other user may rely on us
-    # to handle this block. This proc should be asyncSpawned
-    #
-    # Other people may be using the discovery or blk
-    # so don't kill them
-    blk.cancel()
-    b.stopDiscovery(cid)
-    raise newException(AsyncTimeoutError, "")
 
-  if blk.finished:
+  # get the first peer with at least one (any)
-    # a peer sent us the block out of the blue, why not
+  # matching cid
-    b.stopDiscovery(cid)
+  var blockPeer: BlockExcPeerCtx
-    return await blk
+  for p in peers:
+    if cid in p.peerHave:
+      blockPeer = p
+      break
 
-  # We got a provider
+  # didn't find any peer with matching cids
-  # Currently, we just ask him for the block, and hope he gives it to us
+  # use the first one in the sorted array
-  #
+  if isNil(blockPeer):
-  # In reality, we could keep discovering until we find a suitable price, etc
+    blockPeer = peers[0]
-  b.stopDiscovery(cid)
-  timeoutFut.cancel()
 
-  assert discovery.provides.len > 0
+  peers.keepItIf(
+    it != blockPeer
+  )
 
-  debug "Requesting block from peer", providerCount = discovery.provides.len,
-    peer = discovery.provides[0], cid
   # request block
   b.network.request.sendWantList(
-    discovery.provides[0],
+    blockPeer.id,
     @[cid],
     wantType = WantType.wantBlock) # we want this remote to send us a block
 
-  #TODO substract the discovery time
+  if peers.len == 0:
-  return await blk.wait(timeout)
+    trace "Not enough peers to send want list to", cid = $cid
+    asyncSpawn b.queueFindBlocksReq(@[cid])
+    return blk # no peers to send wants to
+
+  # filter out the peer we've already requested from
+  let stop = min(peers.high, b.peersPerRequest)
+  trace "Sending want list requests to remaining peers", count = stop + 1
+  for p in peers[0..stop]:
+    if cid notin p.peerHave:
+      # just send wants
+      b.network.request.sendWantList(
+        p.id,
+        @[cid],
+        wantType = WantType.wantHave) # we only want to know if the peer has the block
+
+  return blk
 
 proc blockPresenceHandler*(
   b: BlockExcEngine,
@@ -298,18 +291,25 @@ proc blockPresenceHandler*(
   ## Handle block presence
   ##
 
+  trace "Received presence update for peer", peer
   let peerCtx = b.getPeerCtx(peer)
+  if isNil(peerCtx):
+    return
 
   for blk in blocks:
     if presence =? Presence.init(blk):
-      if not isNil(peerCtx):
+      peerCtx.updatePresence(presence)
-        peerCtx.updatePresence(presence)
+
-      if presence.cid in b.runningDiscoveries:
+  let
-        let bd = b.runningDiscoveries[presence.cid]
+    cids = toSeq(b.pendingBlocks.wantList).filterIt(
-        if not presence.have:
+      it in peerCtx.peerHave
-          bd.inflightIWant.excl(peer)
+    )
-          bd.treatedPeer.incl(peer)
+
-        bd.gotIWantResponse.fire()
+  if cids.len > 0:
+    b.network.request.sendWantList(
+      peerCtx.id,
+      cids,
+      wantType = WantType.wantBlock) # we want this remote to send us a block
 
 proc scheduleTasks(b: BlockExcEngine, blocks: seq[bt.Block]) =
   trace "Schedule a task for new blocks"
@@ -330,21 +330,11 @@ proc resolveBlocks*(b: BlockExcEngine, blocks: seq[bt.Block]) =
   ## and schedule any new task to be ran
   ##
 
-  trace "Resolving blocks"
+  trace "Resolving blocks", blocks = blocks.len
 
-  var gotNewBlocks = false
+  b.pendingBlocks.resolve(blocks)
-  for bl in blocks:
+  b.scheduleTasks(blocks)
-    if bl.cid notin b.advertisedBlocks: #TODO that's very slow, maybe a ordered hashset instead
+  asyncSpawn b.queueProvideBlocksReq(blocks.mapIt( it.cid ))
-      #TODO could do some smarter ordering here (insert it just before b.advertisedIndex, or similar)
-      b.advertisedBlocks.add(bl.cid)
-      asyncSpawn b.discovery.publishProvide(bl.cid)
-      gotNewBlocks = true
-
-  if gotNewBlocks:
-    b.pendingBlocks.resolve(blocks)
-    b.scheduleTasks(blocks)
-
-    b.blockAdded.fire()
 
 proc payForBlocks(engine: BlockExcEngine,
                   peer: BlockExcPeerCtx,
@@ -420,14 +410,20 @@ proc wantListHandler*(
   if not b.scheduleTask(peerCtx):
     trace "Unable to schedule task for peer", peer
 
-proc accountHandler*(engine: BlockExcEngine, peer: PeerID, account: Account) {.async.} =
+proc accountHandler*(
+  engine: BlockExcEngine,
+  peer: PeerID,
+  account: Account) {.async.} =
   let context = engine.getPeerCtx(peer)
   if context.isNil:
     return
 
   context.account = account.some
 
-proc paymentHandler*(engine: BlockExcEngine, peer: PeerId, payment: SignedState) {.async.} =
+proc paymentHandler*(
+  engine: BlockExcEngine,
+  peer: PeerId,
+  payment: SignedState) {.async.} =
   without context =? engine.getPeerCtx(peer).option and
           account =? context.account:
     return
@@ -450,13 +446,8 @@ proc setupPeer*(b: BlockExcEngine, peer: PeerID) =
     ))
 
   # broadcast our want list, the other peer will do the same
-  let wantList = collect(newSeqOfCap(b.runningDiscoveries.len)):
+  if b.pendingBlocks.len > 0:
-    for cid, bd in b.runningDiscoveries:
+    b.network.request.sendWantList(peer, toSeq(b.pendingBlocks.wantList), full = true)
-      bd.inflightIWant.incl(peer)
-      cid
-
-  if wantList.len > 0:
-    b.network.request.sendWantList(peer, wantList, full = true, sendDontHave = true)
 
   if address =? b.pricing.?address:
     b.network.request.sendAccount(peer, Account(address: address))
@@ -470,31 +461,6 @@ proc dropPeer*(b: BlockExcEngine, peer: PeerID) =
   # drop the peer from the peers table
   b.peers.keepItIf( it.id != peer )
 
-proc advertiseLoop(b: BlockExcEngine) {.async, gcsafe.} =
-  while true:
-    if b.advertisedIndex >= b.advertisedBlocks.len:
-      b.advertisedIndex = 0
-      b.advertisementFrequency = BlockAdvertisementFrequency
-
-    # check that we still have this block.
-    while
-      b.advertisedIndex < b.advertisedBlocks.len and
-      not(b.localStore.contains(b.advertisedBlocks[b.advertisedIndex])):
-        b.advertisedBlocks.delete(b.advertisedIndex)
-
-    #publish it
-    if b.advertisedIndex < b.advertisedBlocks.len:
-      asyncSpawn b.discovery.publishProvide(b.advertisedBlocks[b.advertisedIndex])
-
-    inc b.advertisedIndex
-    let toSleep =
-      if b.advertisedBlocks.len > 0:
-        b.advertisementFrequency div b.advertisedBlocks.len
-      else:
-        30.minutes
-    await sleepAsync(toSleep) or b.blockAdded.wait()
-    b.blockAdded.clear()
-
 proc taskHandler*(b: BlockExcEngine, task: BlockExcPeerCtx) {.gcsafe, async.} =
   trace "Handling task for peer", peer = task.id
 
@@ -516,6 +482,7 @@ proc taskHandler*(b: BlockExcEngine, task: BlockExcPeerCtx) {.gcsafe, async.} =
       .mapIt(!it.read)
 
     if blocks.len > 0:
+      trace "Sending blocks to peer", peer = task.id, blocks = blocks.len
       b.network.request.sendBlocks(
         task.id,
         blocks)
@@ -558,19 +525,25 @@ proc new*(
   discovery: Discovery,
   concurrentTasks = DefaultConcurrentTasks,
   maxRetries = DefaultMaxRetries,
-  peersPerRequest = DefaultMaxPeersPerRequest): T =
+  peersPerRequest = DefaultMaxPeersPerRequest,
+  concurrentAdvReqs = DefaultConcurrentAdvertRequests,
+  concurrentDiscReqs = DefaultConcurrentDiscRequests): T =
 
-  let engine = BlockExcEngine(
+  let
-    localStore: localStore,
+    engine = BlockExcEngine(
-    pendingBlocks: PendingBlocksManager.new(),
+      localStore: localStore,
-    blockAdded: newAsyncEvent(),
+      pendingBlocks: PendingBlocksManager.new(),
-    peersPerRequest: peersPerRequest,
+      peersPerRequest: peersPerRequest,
-    network: network,
+      network: network,
-    wallet: wallet,
+      wallet: wallet,
-    concurrentTasks: concurrentTasks,
+      concurrentTasks: concurrentTasks,
-    maxRetries: maxRetries,
+      concurrentAdvReqs: concurrentAdvReqs,
-    discovery: discovery,
+      concurrentDiscReqs: concurrentDiscReqs,
-    taskQueue: newAsyncHeapQueue[BlockExcPeerCtx](DefaultTaskQueueSize))
+      maxRetries: maxRetries,
+      taskQueue: newAsyncHeapQueue[BlockExcPeerCtx](DefaultTaskQueueSize),
+      discovery: discovery,
+      advertiseQueue: newAsyncQueue[Cid](DefaultTaskQueueSize),
+      discoveryQUeue: newAsyncQueue[Cid](DefaultTaskQueueSize))
 
   proc peerEventHandler(peerId: PeerID, event: PeerEvent) {.async.} =
     if event.kind == PeerEventKind.Joined:
@@ -608,7 +581,6 @@ proc new*(
     onBlocks: blocksHandler,
     onPresence: blockPresenceHandler,
     onAccount: accountHandler,
-    onPayment: paymentHandler
+    onPayment: paymentHandler)
-  )
 
   return engine

dagger/blockexchange/network.nim

@@ -120,14 +120,16 @@ proc broadcastWantList*(
 
   trace "Sending want list to peer", peer = id, `type` = $wantType, len = cids.len
 
-  let wantList = makeWantList(
+  let
-    cids,
+    wantList = makeWantList(
-    priority,
+      cids,
-    cancel,
+      priority,
-    wantType,
+      cancel,
-    full,
+      wantType,
-    sendDontHave)
+      full,
-  b.peers[id].broadcast(Message(wantlist: wantList))
+      sendDontHave)
+  b.peers.withValue(id, peer):
+    peer[].broadcast(Message(wantlist: wantList))
 
 proc handleBlocks(
   b: BlockExcNetwork,
@@ -153,9 +155,7 @@ proc handleBlocks(
 
   b.handlers.onBlocks(peer.id, blks)
 
-template makeBlocks*(
+template makeBlocks*(blocks: seq[bt.Block]): seq[pb.Block] =
-  blocks: seq[bt.Block]):
-  seq[pb.Block] =
   var blks: seq[pb.Block]
   for blk in blocks:
     blks.add(pb.Block(
@@ -176,7 +176,8 @@ proc broadcastBlocks*(
     return
 
   trace "Sending blocks to peer", peer = id, len = blocks.len
-  b.peers[id].broadcast(pb.Message(payload: makeBlocks(blocks)))
+  b.peers.withValue(id, peer):
+    peer[].broadcast(pb.Message(payload: makeBlocks(blocks)))
 
 proc handleBlockPresence(
   b: BlockExcNetwork,
@@ -202,7 +203,8 @@ proc broadcastBlockPresence*(
     return
 
   trace "Sending presence to peer", peer = id
-  b.peers[id].broadcast(Message(blockPresences: presence))
+  b.peers.withValue(id, peer):
+    peer[].broadcast(Message(blockPresences: @presence))
 
 proc handleAccount(network: BlockExcNetwork,
                    peer: NetworkPeer,
@@ -218,7 +220,8 @@ proc broadcastAccount*(network: BlockExcNetwork,
     return
 
   let message = Message(account: AccountMessage.init(account))
-  network.peers[id].broadcast(message)
+  network.peers.withValue(id, peer):
+    peer[].broadcast(message)
 
 proc broadcastPayment*(network: BlockExcNetwork,
                        id: PeerId,
@@ -227,7 +230,8 @@ proc broadcastPayment*(network: BlockExcNetwork,
     return
 
   let message = Message(payment: StateChannelUpdate.init(payment))
-  network.peers[id].broadcast(message)
+  network.peers.withValue(id, peer):
+    peer[].broadcast(message)
 
 proc handlePayment(network: BlockExcNetwork,
                    peer: NetworkPeer,
@@ -261,7 +265,7 @@ proc getOrCreatePeer(b: BlockExcNetwork, peer: PeerID): NetworkPeer =
   ##
 
   if peer in b.peers:
-    return b.peers[peer]
+    return b.peers.getOrDefault(peer, nil)
 
   var getConn = proc(): Future[Connection] {.async.} =
     try:
@@ -363,8 +367,7 @@ proc new*(
     sendBlocks: sendBlocks,
     sendPresence: sendPresence,
     sendAccount: sendAccount,
-    sendPayment: sendPayment
+    sendPayment: sendPayment)
-  )
 
   b.init()
   return b

dagger/blockexchange/networkpeer.nim

@@ -17,7 +17,7 @@ import ./protobuf/blockexc
 logScope:
   topics = "dagger blockexc networkpeer"
 
-const MaxMessageSize = 8 * 1024 * 1024
+const MaxMessageSize = 100 * 1024 * 1024 # manifest files can be big
 
 type
   RPCHandler* = proc(peer: NetworkPeer, msg: Message): Future[void] {.gcsafe.}

dagger/blockexchange/peercontext.nim

@@ -13,12 +13,12 @@ export payments, nitro
 type
   BlockExcPeerCtx* = ref object of RootObj
     id*: PeerID
-    peerPrices*: Table[Cid, UInt256] # remote peer have list including price
+    peerPrices*: Table[Cid, UInt256]  # remote peer have list including price
-    peerWants*: seq[Entry]      # remote peers want lists
+    peerWants*: seq[Entry]            # remote peers want lists
-    exchanged*: int             # times peer has exchanged with us
+    exchanged*: int                   # times peer has exchanged with us
-    lastExchange*: Moment       # last time peer has exchanged with us
+    lastExchange*: Moment             # last time peer has exchanged with us
-    account*: ?Account          # ethereum account of this peer
+    account*: ?Account                # ethereum account of this peer
-    paymentChannel*: ?ChannelId # payment channel id
+    paymentChannel*: ?ChannelId       # payment channel id
 
 proc peerHave*(context: BlockExcPeerCtx): seq[Cid] =
   toSeq(context.peerPrices.keys)

dagger/blockexchange/pendingblocks.nim

@@ -8,6 +8,11 @@
 ## those terms.
 
 import std/tables
+import std/sequtils
+
+import pkg/upraises
+
+push: {.upraises: [].}
 
 import pkg/questionable
 import pkg/chronicles
@@ -19,18 +24,22 @@ import ../blocktype
 logScope:
   topics = "dagger blockexc pendingblocks"
 
+const
+  DefaultBlockTimeout* = 10.minutes
+
 type
   PendingBlocksManager* = ref object of RootObj
     blocks*: Table[Cid, Future[Block]] # pending Block requests
 
-proc addOrAwait*(
+proc getWantHandle*(
   p: PendingBlocksManager,
-  cid: Cid): Future[Block] {.async.} =
+  cid: Cid,
+  timeout = DefaultBlockTimeout): Future[Block] {.async.} =
   ## Add an event for a block
   ##
 
   if cid notin p.blocks:
-     p.blocks[cid] = newFuture[Block]()
+     p.blocks[cid] = newFuture[Block]().wait(timeout)
      trace "Adding pending future for block", cid
 
   try:
@@ -52,11 +61,11 @@ proc resolve*(
   for blk in blocks:
     # resolve any pending blocks
     if blk.cid in p.blocks:
-      let pending = p.blocks[blk.cid]
+      p.blocks.withValue(blk.cid, pending):
-      if not pending.finished:
+        if not pending[].finished:
-        trace "Resolving block", cid = $blk.cid
+          trace "Resolving block", cid = $blk.cid
-        pending.complete(blk)
+          pending[].complete(blk)
-        p.blocks.del(blk.cid)
+          p.blocks.del(blk.cid)
 
 proc pending*(
   p: PendingBlocksManager,
@@ -66,6 +75,17 @@ proc contains*(
   p: PendingBlocksManager,
   cid: Cid): bool = p.pending(cid)
 
+iterator wantList*(p: PendingBlocksManager): Cid =
+  for k in p.blocks.keys:
+    yield k
+
+iterator wantHandles*(p: PendingBlocksManager): Future[Block] =
+  for v in p.blocks.values:
+    yield v
+
+func len*(p: PendingBlocksManager): int =
+  p.blocks.len
+
 func new*(T: type PendingBlocksManager): T =
   T(
     blocks: initTable[Cid, Future[Block]]()

dagger/dagger.nim

@@ -123,7 +123,7 @@ proc new*(T: type DaggerServer, config: DaggerConf): T =
     erasure = Erasure.new(store, leoEncoderProvider, leoDecoderProvider)
     daggerNode = DaggerNodeRef.new(switch, store, engine, erasure, discovery)
     restServer = RestServerRef.new(
-      daggerNode.initRestApi(),
+      daggerNode.initRestApi(config),
       initTAddress("127.0.0.1" , config.apiPort),
       bufferSize = (1024 * 64),
       maxRequestBodySize = int.high)

dagger/discovery.nim

@@ -8,18 +8,20 @@
 ## those terms.
 
 import pkg/chronos
+import pkg/chronicles
 import pkg/libp2p
 import pkg/questionable
 import pkg/questionable/results
 import pkg/stew/shims/net
 import pkg/libp2pdht/discv5/protocol as discv5
 
-import rng
+import ./rng
+import ./errors
 
 export discv5
 
 type
-  Discovery* = ref object
+  Discovery* = ref object of RootObj
     protocol: discv5.Protocol
     localInfo: PeerInfo
 
@@ -55,15 +57,33 @@ proc toDiscoveryId*(cid: Cid): NodeId =
   ## To discovery id
   readUintBE[256](keccak256.digest(cid.data.buffer).data)
 
-proc findBlockProviders*(
+method findBlockProviders*(
   d: Discovery,
-  cid: Cid): Future[seq[SignedPeerRecord]] {.async.} =
+  cid: Cid): Future[seq[SignedPeerRecord]] {.async, base.} =
-  return (await d.protocol.getProviders(cid.toDiscoveryId())).get()
+  ## Find block providers
+  ##
 
-proc publishProvide*(d: Discovery, cid: Cid) {.async.} =
+  trace "Finding providers for block", cid = $cid
-  let bid = cid.toDiscoveryId()
+  without providers =?
-  discard await d.protocol.addProvider(bid, d.localInfo.signedPeerRecord)
+    (await d.protocol.getProviders(cid.toDiscoveryId())).mapFailure, error:
+    trace "Error finding providers for block", cid = $cid, error = error.msg
 
+  return providers
+
+method provideBlock*(d: Discovery, cid: Cid) {.async, base.} =
+  ## Provide a bock Cid
+  ##
+
+  trace "Providing block", cid = $cid
+  let
+    nodes = await d.protocol.addProvider(
+      cid.toDiscoveryId(),
+      d.localInfo.signedPeerRecord)
+
+  if nodes.len <= 0:
+    trace "Couldn't provide to any nodes!"
+
+  trace "Provided to nodes", nodes = nodes.len
 
 proc start*(d: Discovery) {.async.} =
   d.protocol.updateRecord(d.localInfo.signedPeerRecord).expect("updating SPR")

dagger/erasure/erasure.nim

@@ -109,7 +109,7 @@ proc encode*(
       # TODO: this is a tight blocking loop so we sleep here to allow
       # other events to be processed, this should be addressed
       # by threading
-      await sleepAsync(10.millis)
+      await sleepAsync(100.millis)
 
       for j in 0..<blocks:
         let idx = blockIdx[j]

dagger/node.nim

@@ -44,10 +44,17 @@ type
     discovery*: Discovery
 
 proc start*(node: DaggerNodeRef) {.async.} =
-  await node.switch.start()
+  if not node.switch.isNil:
-  await node.engine.start()
+    await node.switch.start()
-  await node.erasure.start()
+
-  await node.discovery.start()
+  if not node.engine.isNil:
+    await node.engine.start()
+
+  if not node.erasure.isNil:
+    await node.erasure.start()
+
+  if not node.discovery.isNil:
+    await node.discovery.start()
 
   node.networkId = node.switch.peerInfo.peerId
   notice "Started dagger node", id = $node.networkId, addrs = node.switch.peerInfo.addrs
@@ -55,10 +62,17 @@ proc start*(node: DaggerNodeRef) {.async.} =
 proc stop*(node: DaggerNodeRef) {.async.} =
   trace "Stopping node"
 
-  await node.engine.stop()
+  if not node.engine.isNil:
-  await node.switch.stop()
+    await node.engine.stop()
-  await node.erasure.stop()
+
-  await node.discovery.stop()
+  if not node.switch.isNil:
+    await node.switch.stop()
+
+  if not node.erasure.isNil:
+    await node.erasure.stop()
+
+  if not node.discovery.isNil:
+    await node.discovery.stop()
 
 proc findPeer*(
   node: DaggerNodeRef,

dagger/rest/api.nim

@@ -21,11 +21,13 @@ import pkg/chronos
 import pkg/presto
 import pkg/libp2p
 import pkg/stew/base10
+import pkg/confutils
 
 import pkg/libp2p/routing_record
 
 import ../node
 import ../blocktype
+import ../conf
 
 proc validate(
   pattern: string,
@@ -83,7 +85,7 @@ proc decodeString(T: type bool, value: string): Result[T, cstring] =
 proc encodeString(value: bool): Result[string, cstring] =
   ok($value)
 
-proc initRestApi*(node: DaggerNodeRef): RestRouter =
+proc initRestApi*(node: DaggerNodeRef, conf: DaggerConf): RestRouter =
   var router = RestRouter.init(validate)
   router.api(
     MethodGet,
@@ -318,6 +320,7 @@ proc initRestApi*(node: DaggerNodeRef): RestRouter =
 
       return RestApiResponse.response(
         "Id: " & $node.switch.peerInfo.peerId &
-        "\nAddrs: \n" & addrs & "\n")
+        "\nAddrs: \n" & addrs &
+        "\nRoot Dir: " & $conf.dataDir)
 
   return router

dagger/stores/blockstore.nim

@@ -20,6 +20,7 @@ import ../blocktype
 export blocktype, libp2p
 
 type
+  OnBlock* = proc(cid: Cid): Future[void] {.upraises: [], gcsafe.}
   BlockStore* = ref object of RootObj
 
 method getBlock*(
@@ -52,7 +53,7 @@ method hasBlock*(s: BlockStore, cid: Cid): bool {.base.} =
 
   return false
 
-method blockList*(s: BlockStore): Future[seq[Cid]] {.base.} =
+method listBlocks*(s: BlockStore, onBlock: OnBlock): Future[void] {.base.} =
   ## Get the list of blocks in the BlockStore. This is an intensive operation
   ##
 

dagger/stores/cachestore.nim

@@ -68,8 +68,9 @@ method hasBlock*(self: CacheStore, cid: Cid): bool =
 
   cid in self.cache
 
-method blockList*(s: CacheStore): Future[seq[Cid]] {.async.} =
+method listBlocks*(s: CacheStore, onBlock: OnBlock) {.async.} =
-  return toSeq(s.cache.keys)
+  for cid in toSeq(s.cache.keys):
+    await onBlock(cid)
 
 func putBlockSync(self: CacheStore, blk: Block): bool =
 

dagger/stores/fsstore.nim

@@ -90,7 +90,7 @@ method putBlock*(
     trace "Unable to store block", path, cid = blk.cid, error
     return false
 
-  if await self.cache.putBlock(blk):
+  if not (await self.cache.putBlock(blk)):
     trace "Unable to store block in cache", cid = blk.cid
 
   return true
@@ -113,8 +113,8 @@ method delBlock*(
     trace "Unable to delete block", path, cid, error
     return false
 
-  if await self.cache.delBlock(cid):
+  if not (await self.cache.delBlock(cid)):
-    trace "Unable to store block in cache", cid
+    trace "Unable to delete block from cache", cid
 
   return true
 
@@ -129,21 +129,25 @@ method hasBlock*(self: FSStore, cid: Cid): bool =
 
   self.blockPath(cid).isFile()
 
-method blockList*(s: FSStore): Future[seq[Cid]] {.async.} =
+method listBlocks*(self: FSStore, onBlock: OnBlock) {.async.} =
-  ## Very expensive AND blocking!
+  debug "Finding all blocks in store"
-
+  for (pkind, folderPath) in self.repoDir.walkDir():
-  debug "finding all blocks in store"
-  for (pkind, folderPath) in s.repoDir.walkDir():
     if pkind != pcDir: continue
     let baseName = basename(folderPath)
-    if baseName.len != s.postfixLen: continue
+    if baseName.len != self.postfixLen: continue
 
     for (fkind, filePath) in folderPath.walkDir(false):
       if fkind != pcFile: continue
       let cid = Cid.init(basename(filePath))
       if cid.isOk:
-        result.add(cid.get())
+        # getting a weird `Error: unhandled exception: index 1 not in 0 .. 0 [IndexError]`
-  return result
+        # compilation error if using different syntax/construct bellow
+        try:
+          await onBlock(cid.get())
+        except CatchableError as exc:
+          trace "Couldn't get block", cid = $(cid.get())
+
+      await sleepAsync(100.millis) # avoid blocking
 
 proc new*(
   T: type FSStore,

dagger/stores/networkstore.nim

@@ -42,10 +42,10 @@ method getBlock*(
   trace "Getting block", cid
   without var blk =? (await self.localStore.getBlock(cid)):
     trace "Couldn't get from local store", cid
-    blk = try:
+    try:
-      await self.engine.requestBlock(cid)
+      blk = await self.engine.requestBlock(cid)
     except CatchableError as exc:
-      trace "Exception requestig block", cid, exc = exc.msg
+      trace "Exception requesting block", cid, exc = exc.msg
       return failure(exc.msg)
 
   trace "Retrieved block from local store", cid

dagger/utils.nim

@@ -0,0 +1,4 @@
+import ./utils/asyncheapqueue
+import ./utils/fileutils
+
+export asyncheapqueue, fileutils

tests/config.nims

@@ -1 +0,0 @@
-patchFile("dagger", "discovery", "dagger/mockdiscovery")

tests/dagger/blockexc/discovery/mockdiscovery.nim

@@ -12,17 +12,15 @@ import pkg/libp2p
 import pkg/questionable
 import pkg/questionable/results
 import pkg/stew/shims/net
-import pkg/libp2pdht/discv5/protocol as discv5
+import pkg/dagger/discovery
-
-export discv5
 
 type
-  Discovery* = ref object
+  MockDiscovery* = ref object of Discovery
-    findBlockProviders_var*: proc(d: Discovery, cid: Cid): seq[SignedPeerRecord] {.gcsafe.}
+    findBlockProvidersHandler*: proc(d: MockDiscovery, cid: Cid): seq[SignedPeerRecord] {.gcsafe.}
-    publishProvide_var*: proc(d: Discovery, cid: Cid) {.gcsafe.}
+    publishProvideHandler*: proc(d: MockDiscovery, cid: Cid) {.gcsafe.}
 
 proc new*(
-  T: type Discovery,
+  T: type MockDiscovery,
   localInfo: PeerInfo,
   discoveryPort: Port,
   bootstrapNodes = newSeq[SignedPeerRecord](),
@@ -35,17 +33,16 @@ proc findPeer*(
   peerId: PeerID): Future[?PeerRecord] {.async.} =
   return none(PeerRecord)
 
-proc findBlockProviders*(
+method findBlockProviders*(
-  d: Discovery,
+  d: MockDiscovery,
   cid: Cid): Future[seq[SignedPeerRecord]] {.async.} =
-  if isNil(d.findBlockProviders_var): return
+  if isNil(d.findBlockProvidersHandler): return
 
-  return d.findBlockProviders_var(d, cid)
+  return d.findBlockProvidersHandler(d, cid)
-
-proc publishProvide*(d: Discovery, cid: Cid) {.async.} =
-  if isNil(d.publishProvide_var): return
-  d.publishProvide_var(d, cid)
 
+method provideBlock*(d: MockDiscovery, cid: Cid) {.async.} =
+  if isNil(d.publishProvideHandler): return
+  d.publishProvideHandler(d, cid)
 
 proc start*(d: Discovery) {.async.} =
   discard

tests/dagger/blockexc/discovery/testdiscovery.nim

@@ -0,0 +1,192 @@
+import std/sequtils
+import std/sugar
+import std/algorithm
+import std/tables
+
+import pkg/asynctest
+import pkg/chronos
+import pkg/stew/byteutils
+
+import pkg/libp2p
+import pkg/libp2p/errors
+
+import pkg/dagger/rng
+import pkg/dagger/stores
+import pkg/dagger/blockexchange
+import pkg/dagger/chunker
+import pkg/dagger/blocktype as bt
+
+import ./mockdiscovery
+
+import ../../helpers
+import ../../examples
+
+suite "Block Advertising and Discovery":
+  let chunker = RandomChunker.new(Rng.instance(), size = 4096, chunkSize = 256)
+
+  var
+    switch: seq[Switch]
+    blockexc: seq[NetworkStore]
+    blocks: seq[bt.Block]
+
+  setup:
+    while true:
+      let chunk = await chunker.getBytes()
+      if chunk.len <= 0:
+        break
+
+      blocks.add(bt.Block.new(chunk).tryGet())
+
+  teardown:
+    switch = @[]
+    blockexc = @[]
+
+  test "Should discover want list":
+    let
+      s = newStandardSwitch(transportFlags = {ServerFlags.ReuseAddr})
+      discovery = MockDiscovery.new(s.peerInfo, 0.Port)
+      wallet = WalletRef.example
+      network = BlockExcNetwork.new(s)
+      localStore = CacheStore.new(blocks.mapIt( it ))
+      engine = BlockExcEngine.new(localStore, wallet, network, discovery)
+
+    s.mount(network)
+    switch.add(s)
+
+    await allFuturesThrowing(
+      switch.mapIt( it.start() )
+    )
+
+    let
+      pendingBlocks = blocks.mapIt(
+        engine.pendingBlocks.getWantHandle(it.cid)
+      )
+
+    await engine.start() # fire up discovery loop
+    discovery.findBlockProvidersHandler =
+      proc(d: MockDiscovery, cid: Cid): seq[SignedPeerRecord] =
+        engine.resolveBlocks(blocks.filterIt( it.cid == cid ))
+
+  test "Should advertise have blocks":
+    let
+      s = newStandardSwitch(transportFlags = {ServerFlags.ReuseAddr})
+      discovery = MockDiscovery.new(s.peerInfo, 0.Port)
+      wallet = WalletRef.example
+      network = BlockExcNetwork.new(s)
+      localStore = CacheStore.new(blocks.mapIt( it ))
+      engine = BlockExcEngine.new(localStore, wallet, network, discovery)
+
+    s.mount(network)
+    switch.add(s)
+
+    await allFuturesThrowing(
+      switch.mapIt( it.start() )
+    )
+
+    let
+      advertised = initTable.collect:
+        for b in blocks: {b.cid: newFuture[void]()}
+
+    discovery.publishProvideHandler = proc(d: MockDiscovery, cid: Cid) =
+      if cid in advertised and not advertised[cid].finished():
+        advertised[cid].complete()
+
+    await engine.start() # fire up advertise loop
+    await allFuturesThrowing(
+      allFinished(toSeq(advertised.values)))
+
+suite "E2E - Multiple Nodes Discovery":
+  let chunker = RandomChunker.new(Rng.instance(), size = 4096, chunkSize = 256)
+
+  var
+    switch: seq[Switch]
+    blockexc: seq[NetworkStore]
+    blocks: seq[bt.Block]
+
+  setup:
+    while true:
+      let chunk = await chunker.getBytes()
+      if chunk.len <= 0:
+        break
+
+      blocks.add(bt.Block.new(chunk).tryGet())
+
+    for _ in 0..<4:
+      let
+        s = newStandardSwitch(transportFlags = {ServerFlags.ReuseAddr})
+        discovery = MockDiscovery.new(s.peerInfo, 0.Port)
+        wallet = WalletRef.example
+        network = BlockExcNetwork.new(s)
+        localStore = CacheStore.new(blocks.mapIt( it ))
+        engine = BlockExcEngine.new(localStore, wallet, network, discovery)
+        networkStore = NetworkStore.new(engine, localStore)
+
+      s.mount(network)
+      switch.add(s)
+      blockexc.add(networkStore)
+
+  teardown:
+    switch = @[]
+    blockexc = @[]
+
+  test "Should not launch discovery request if we are already connected":
+    await allFuturesThrowing(
+      blockexc.mapIt( it.engine.start() ) &
+      switch.mapIt( it.start() )
+    )
+
+    await blockexc[0].engine.blocksHandler(switch[1].peerInfo.peerId, blocks)
+    MockDiscovery(blockexc[0].engine.discovery)
+      .findBlockProvidersHandler = proc(d: MockDiscovery, cid: Cid): seq[SignedPeerRecord] =
+        check false
+
+    await connectNodes(switch)
+    let blk = await blockexc[1].engine.requestBlock(blocks[0].cid)
+
+    await allFuturesThrowing(
+      blockexc.mapIt( it.engine.stop() ) &
+      switch.mapIt( it.stop() )
+    )
+
+  test "E2E - Should advertise and discover blocks":
+    # Distribute the blocks amongst 1..3
+    # Ask 0 to download everything without connecting him beforehand
+
+    var advertised: Table[Cid, SignedPeerRecord]
+
+    MockDiscovery(blockexc[1].engine.discovery)
+      .publishProvideHandler = proc(d: MockDiscovery, cid: Cid) =
+        advertised.add(cid, switch[1].peerInfo.signedPeerRecord)
+
+    MockDiscovery(blockexc[2].engine.discovery)
+      .publishProvideHandler = proc(d: MockDiscovery, cid: Cid) =
+        advertised.add(cid, switch[2].peerInfo.signedPeerRecord)
+
+    MockDiscovery(blockexc[3].engine.discovery)
+      .publishProvideHandler = proc(d: MockDiscovery, cid: Cid) =
+        advertised.add(cid, switch[3].peerInfo.signedPeerRecord)
+
+    await blockexc[1].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[0..5])
+    await blockexc[2].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[4..10])
+    await blockexc[3].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[10..15])
+
+    MockDiscovery(blockexc[0].engine.discovery)
+      .findBlockProvidersHandler = proc(d: MockDiscovery, cid: Cid): seq[SignedPeerRecord] =
+        if cid in advertised:
+          result.add(advertised[cid])
+
+    let futs = collect(newSeq):
+      for b in blocks:
+        blockexc[0].engine.requestBlock(b.cid)
+
+    await allFuturesThrowing(
+      switch.mapIt( it.start() ) &
+      blockexc.mapIt( it.engine.start() )
+    )
+
+    await allFutures(futs)
+
+    await allFuturesThrowing(
+      blockexc.mapIt( it.engine.stop() ) &
+      switch.mapIt( it.stop() )
+    )

tests/dagger/blockexc/testblockexc.nim

@@ -1,5 +1,4 @@
 import std/sequtils
-import std/sugar
 import std/algorithm
 
 import pkg/asynctest
@@ -8,7 +7,6 @@ import pkg/stew/byteutils
 
 import pkg/libp2p
 import pkg/libp2p/errors
-import pkg/libp2pdht/discv5/protocol as discv5
 
 import pkg/dagger/rng
 import pkg/dagger/stores
@@ -38,6 +36,7 @@ suite "NetworkStore engine - 2 nodes":
     engine1, engine2: BlockExcEngine
     localStore1, localStore2: BlockStore
     discovery1, discovery2: Discovery
+    pendingBlocks1, pendingBlocks2: seq[Future[bt.Block]]
 
   setup:
     while true:
@@ -86,8 +85,8 @@ suite "NetworkStore engine - 2 nodes":
     )
 
     # initialize our want lists
-    for b in blocks2: discard blockexc1.engine.discoverBlock(b.cid)
+    pendingBlocks1 = blocks2.mapIt( blockexc1.engine.pendingBlocks.getWantHandle( it.cid ) )
-    for b in blocks1: discard blockexc2.engine.discoverBlock(b.cid)
+    pendingBlocks2 = blocks1.mapIt( blockexc2.engine.pendingBlocks.getWantHandle( it.cid ) )
 
     pricing1.address = wallet1.address
     pricing2.address = wallet2.address
@@ -98,7 +97,7 @@ suite "NetworkStore engine - 2 nodes":
       switch2.peerInfo.peerId,
       switch2.peerInfo.addrs)
 
-    await sleepAsync(100.milliseconds) # give some time to exchange lists
+    await sleepAsync(1.seconds) # give some time to exchange lists
     peerCtx2 = blockexc1.engine.getPeerCtx(peerId2)
     peerCtx1 = blockexc2.engine.getPeerCtx(peerId1)
 
@@ -113,12 +112,18 @@ suite "NetworkStore engine - 2 nodes":
     check not isNil(peerCtx1)
     check not isNil(peerCtx2)
 
+    await allFuturesThrowing(
+      allFinished(pendingBlocks1))
+
+    await allFuturesThrowing(
+      allFinished(pendingBlocks2))
+
     check:
       peerCtx1.peerHave.mapIt( $it ).sorted(cmp[string]) ==
-        toSeq(blockexc2.engine.runningDiscoveries.keys()).mapIt( $it ).sorted(cmp[string])
+        pendingBlocks2.mapIt( $it.read.cid ).sorted(cmp[string])
 
       peerCtx2.peerHave.mapIt( $it ).sorted(cmp[string]) ==
-        toSeq(blockexc1.engine.runningDiscoveries.keys()).mapIt( $it ).sorted(cmp[string])
+        pendingBlocks1.mapIt( $it.read.cid ).sorted(cmp[string])
 
   test "exchanges accounts on connect":
     check peerCtx1.account.?address == pricing1.address.some
@@ -175,7 +180,8 @@ suite "NetworkStore engine - 2 nodes":
     check wallet2.balance(channel, Asset) > 0
 
 suite "NetworkStore - multiple nodes":
-  let chunker = RandomChunker.new(Rng.instance(), size = 4096, chunkSize = 256)
+  let
+    chunker = RandomChunker.new(Rng.instance(), size = 4096, chunkSize = 256)
 
   var
     switch: seq[Switch]
@@ -213,10 +219,9 @@ suite "NetworkStore - multiple nodes":
       engine = downloader.engine
 
     # Add blocks from 1st peer to want list
-    for b in blocks[0..3]:
+    let
-      discard engine.discoverBlock(b.cid)
+      pendingBlocks1 = blocks[0..3].mapIt( engine.pendingBlocks.getWantHandle( it.cid ) )
-    for b in blocks[12..15]:
+      pendingBlocks2 = blocks[12..15].mapIt( engine.pendingBlocks.getWantHandle( it.cid ))
-      discard engine.discoverBlock(b.cid)
 
     await allFutures(
       blocks[0..3].mapIt( blockexc[0].engine.localStore.putBlock(it) ))
@@ -230,12 +235,16 @@ suite "NetworkStore - multiple nodes":
     await connectNodes(switch)
     await sleepAsync(1.seconds)
 
+    await allFuturesThrowing(
+      allFinished(pendingBlocks1),
+      allFinished(pendingBlocks2))
+
     check:
       engine.peers[0].peerHave.mapIt($it).sorted(cmp[string]) ==
-        blocks[0..3].mapIt( it.cid ).mapIt($it).sorted(cmp[string])
+        blocks[0..3].mapIt( $(it.cid) ).sorted(cmp[string])
 
       engine.peers[3].peerHave.mapIt($it).sorted(cmp[string]) ==
-        blocks[12..15].mapIt( it.cid ).mapIt($it).sorted(cmp[string])
+        blocks[12..15].mapIt( $(it.cid) ).sorted(cmp[string])
 
   test "should exchange blocks with multiple nodes":
     let
@@ -243,10 +252,9 @@ suite "NetworkStore - multiple nodes":
       engine = downloader.engine
 
     # Add blocks from 1st peer to want list
-    for b in blocks[0..3]:
+    let
-      discard engine.discoverBlock(b.cid)
+      pendingBlocks1 = blocks[0..3].mapIt( engine.pendingBlocks.getWantHandle( it.cid ) )
-    for b in blocks[12..15]:
+      pendingBlocks2 = blocks[12..15].mapIt( engine.pendingBlocks.getWantHandle( it.cid ))
-      discard engine.discoverBlock(b.cid)
 
     await allFutures(
       blocks[0..3].mapIt( blockexc[0].engine.localStore.putBlock(it) ))
@@ -260,74 +268,9 @@ suite "NetworkStore - multiple nodes":
     await connectNodes(switch)
     await sleepAsync(1.seconds)
 
-    let wantListBlocks = await allFinished(
-      blocks[0..3].mapIt( downloader.getBlock(it.cid) ))
-    check wantListBlocks.mapIt( !it.read ) == blocks[0..3]
-
-suite "NetworkStore - discovery":
-  let chunker = RandomChunker.new(Rng.instance(), size = 4096, chunkSize = 256)
-
-  var
-    switch: seq[Switch]
-    blockexc: seq[NetworkStore]
-    blocks: seq[bt.Block]
-
-  setup:
-    while true:
-      let chunk = await chunker.getBytes()
-      if chunk.len <= 0:
-        break
-
-      blocks.add(bt.Block.new(chunk).tryGet())
-
-    for e in generateNodes(4):
-      switch.add(e.switch)
-      blockexc.add(e.blockexc)
-      await e.blockexc.engine.start()
-
     await allFuturesThrowing(
-      switch.mapIt( it.start() )
+      allFinished(pendingBlocks1),
-    )
+      allFinished(pendingBlocks2))
 
-  teardown:
+    check pendingBlocks1.mapIt( it.read ) == blocks[0..3]
-    await allFuturesThrowing(
+    check pendingBlocks2.mapIt( it.read ) == blocks[12..15]
-      switch.mapIt( it.stop() )
-    )
-
-    switch = @[]
-    blockexc = @[]
-
-  test "Shouldn't launch discovery request if we are already connected":
-    await blockexc[0].engine.blocksHandler(switch[1].peerInfo.peerId, blocks)
-    blockexc[0].engine.discovery.findBlockProviders_var = proc(d: Discovery, cid: Cid): seq[SignedPeerRecord] =
-      check false
-    await connectNodes(switch)
-    let blk = await blockexc[1].engine.requestBlock(blocks[0].cid)
-
-  test "E2E discovery":
-    # Distribute the blocks amongst 1..3
-    # Ask 0 to download everything without connecting him beforehand
-
-    var advertised: Table[Cid, SignedPeerRecord]
-
-    blockexc[1].engine.discovery.publishProvide_var = proc(d: Discovery, cid: Cid) =
-      advertised[cid] = switch[1].peerInfo.signedPeerRecord
-
-    blockexc[2].engine.discovery.publishProvide_var = proc(d: Discovery, cid: Cid) =
-      advertised[cid] = switch[2].peerInfo.signedPeerRecord
-
-    blockexc[3].engine.discovery.publishProvide_var = proc(d: Discovery, cid: Cid) =
-      advertised[cid] = switch[3].peerInfo.signedPeerRecord
-
-    await blockexc[1].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[0..5])
-    await blockexc[2].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[4..10])
-    await blockexc[3].engine.blocksHandler(switch[0].peerInfo.peerId, blocks[10..15])
-
-    blockexc[0].engine.discovery.findBlockProviders_var = proc(d: Discovery, cid: Cid): seq[SignedPeerRecord] =
-      if cid in advertised:
-        result.add(advertised[cid])
-
-    let futs = collect(newSeq):
-      for b in blocks:
-        blockexc[0].engine.requestBlock(b.cid)
-    await allFutures(futs)

tests/dagger/blockexc/testengine.nim

@@ -66,8 +66,9 @@ suite "NetworkStore engine basic":
         wallet,
         network,
         discovery)
+
     for b in blocks:
-      discard engine.discoverBlock(b.cid)
+      discard engine.pendingBlocks.getWantHandle(b.cid)
     engine.setupPeer(peerId)
 
     await done
@@ -171,7 +172,7 @@ suite "NetworkStore engine handlers":
 
   test "stores blocks in local store":
     let pending = blocks.mapIt(
-      engine.pendingBlocks.addOrAwait( it.cid )
+      engine.pendingBlocks.getWantHandle( it.cid )
     )
 
     await engine.blocksHandler(peerId, blocks)

tests/dagger/helpers/nodeutils.nim

@@ -11,10 +11,8 @@ import ../examples
 
 proc generateNodes*(
   num: Natural,
-  blocks: openArray[bt.Block] = [],
+  blocks: openArray[bt.Block] = []):
-  secureManagers: openarray[SecureProtocol] = [
+  seq[tuple[switch: Switch, blockexc: NetworkStore]] =
-    SecureProtocol.Noise,
-  ]): seq[tuple[switch: Switch, blockexc: NetworkStore]] =
   for i in 0..<num:
     let
       switch = newStandardSwitch(transportFlags = {ServerFlags.ReuseAddr})
@@ -25,8 +23,6 @@ proc generateNodes*(
       engine = BlockExcEngine.new(localStore, wallet, network, discovery)
       networkStore = NetworkStore.new(engine, localStore)
 
-    switch.mount(network)
-
     switch.mount(network)
     result.add((switch, networkStore))
 

tests/dagger/stores/testcachestore.nim

@@ -106,3 +106,15 @@ suite "Cache Store tests":
       await store.delBlock(newBlock2.cid)
       store.currentSize == 200
       newBlock2.cid notin store
+
+  test "listBlocks":
+    discard await store.putBlock(newBlock1)
+
+    var listed = false
+    await store.listBlocks(
+      proc(cid: Cid) {.gcsafe, async.} =
+        check cid in store
+        listed = true
+    )
+
+    check listed

tests/dagger/stores/testfsstore.nim

@@ -52,11 +52,13 @@ suite "FS Store":
 
     check store.hasBlock(newBlock.cid)
 
-  test "blockList":
+  test "listBlocks":
     createDir(store.blockPath(newBlock.cid).parentDir)
     writeFile(store.blockPath(newBlock.cid), newBlock.data)
 
-    check (await store.blockList()) == @[newBlock.cid]
+    await store.listBlocks(
+      proc(cid: Cid) {.gcsafe, async.} =
+        check cid == newBlock.cid)
 
   test "fail hasBlock":
     check not store.hasBlock(newBlock.cid)

tests/dagger/testblockexc.nim

@@ -4,5 +4,6 @@ import ./blockexc/protobuf/testpayments as testprotobufpayments
 import ./blockexc/protobuf/testpresence
 import ./blockexc/engine/testpayments as testenginepayments
 import ./blockexc/testblockexc
+import ./blockexc/discovery/testdiscovery
 
 {.warning[UnusedImport]: off.}