Rework AsyncIter (#811)

* Rework AsyncIter

* Add tests for finishing iter on error

* Improved error handling for  and additional tests

* Use new style of constructors

* Handle future cancellation

* Docs for constructors

codex/erasure/erasure.nim

@@ -120,7 +120,7 @@ proc getPendingBlocks(
         CatchableError,
         "Future for block id not found, tree cid: " & $manifest.treeCid & ", index: " & $index)
 
-  Iter.new(genNext, isFinished)
+  AsyncIter[(?!bt.Block, int)].new(genNext, isFinished)
 
 proc prepareEncodingData(
   self: Erasure,
@@ -440,8 +440,7 @@ proc decode*(
   if treeCid != encoded.originalTreeCid:
     return failure("Original tree root differs from the tree root computed out of recovered data")
 
-  let idxIter = Iter
+  let idxIter = Iter[Natural].new(recoveredIndices)
-    .fromItems(recoveredIndices)
     .filter((i: Natural) => i < tree.leavesCount)
 
   if err =? (await self.store.putSomeProofs(tree, idxIter)).errorOption:

codex/indexingstrategy.nim

@@ -39,22 +39,9 @@ func checkIteration(self: IndexingStrategy, iteration: int): void {.raises: [Ind
       IndexingError,
       "Indexing iteration can't be greater than or equal to iterations.")
 
-proc getIter(first, last, step: int): Iter[int] =
+func getIter(first, last, step: int): Iter[int] =
-  var
+  {.cast(noSideEffect).}:
-    finish = false
+    Iter[int].new(first, last, step)
-    cur = first
-
-  func get(): int =
-    result = cur
-    cur += step
-
-    if cur > last:
-      finish = true
-
-  func isFinished(): bool =
-    finish
-
-  Iter.new(get, isFinished)
 
 func getLinearIndicies(
   self: IndexingStrategy,

codex/node.nim

@@ -157,10 +157,8 @@ proc updateExpiry*(
 
   try:
     let
-      ensuringFutures = Iter
+      ensuringFutures = Iter[int].new(0..<manifest.blocksCount)
-        .fromSlice(0..<manifest.blocksCount)
+        .mapIt(self.networkStore.localStore.ensureExpiry( manifest.treeCid, it, expiry ))
-        .mapIt(
-          self.networkStore.localStore.ensureExpiry( manifest.treeCid, it, expiry ))
     await allFuturesThrowing(ensuringFutures)
   except CancelledError as exc:
     raise exc
@@ -209,7 +207,7 @@ proc fetchBatched*(
 
   trace "Fetching blocks in batches of", size = batchSize
 
-  let iter = Iter.fromSlice(0..<manifest.blocksCount)
+  let iter = Iter[int].new(0..<manifest.blocksCount)
   self.fetchBatched(manifest.treeCid, iter, batchSize, onBatch)
 
 proc streamSingleBlock(

codex/stores/cachestore.nim

@@ -132,51 +132,35 @@ method listBlocks*(
   ## Get the list of blocks in the BlockStore. This is an intensive operation
   ##
 
-  var
-    iter = AsyncIter[?Cid]()
-
   let
     cids = self.cids()
 
-  proc next(): Future[?Cid] {.async.} =
+  proc isFinished(): bool =
-    await idleAsync()
+    return finished(cids)
 
-    var cid: Cid
+  proc genNext(): Future[Cid] {.async.} =
-    while true:
+    cids()
-      if iter.finished:
-        return Cid.none
 
-      cid = cids()
+  let iter = await (AsyncIter[Cid].new(genNext, isFinished)
+    .filter(
+      proc (cid: Cid): Future[bool] {.async.} =
+        without isManifest =? cid.isManifest, err:
+          trace "Error checking if cid is a manifest", err = err.msg
+          return false
 
-      if finished(cids):
+        case blockType:
-        iter.finish
+        of BlockType.Both:
-        return Cid.none
+          return true
+        of BlockType.Manifest:
+          return isManifest
+        of BlockType.Block:
+          return not isManifest
+    ))
 
-      without isManifest =? cid.isManifest, err:
+  return success(map[Cid, ?Cid](iter,
-        trace "Error checking if cid is a manifest", err = err.msg
+      proc (cid: Cid): Future[?Cid] {.async.} =
-        return Cid.none
+        some(cid)
-
+  ))
-      case blockType:
-      of BlockType.Manifest:
-        if not isManifest:
-          trace "Cid is not manifest, skipping", cid
-          continue
-
-        break
-      of BlockType.Block:
-        if isManifest:
-          trace "Cid is a manifest, skipping", cid
-          continue
-
-        break
-      of BlockType.Both:
-        break
-
-    return cid.some
-
-  iter.next = next
-
-  return success iter
 
 func putBlockSync(self: CacheStore, blk: Block): bool =
 

codex/stores/queryiterhelper.nim

@@ -0,0 +1,66 @@
+import pkg/questionable
+import pkg/questionable/results
+import pkg/chronos
+import pkg/chronicles
+import pkg/datastore/typedds
+
+import ../utils/asynciter
+
+type KeyVal[T] = tuple[key: Key, value: T]
+
+proc toAsyncIter*[T](
+  queryIter: QueryIter[T],
+  finishOnErr: bool = true
+  ): Future[?!AsyncIter[?!QueryResponse[T]]] {.async.} =
+  ## Converts `QueryIter[T]` to `AsyncIter[?!QueryResponse[T]]` and automatically
+  ## runs dispose whenever `QueryIter` finishes or whenever an error occurs (only
+  ## if the flag finishOnErr is set to true)
+  ##
+
+  if queryIter.finished:
+    trace "Disposing iterator"
+    if error =? (await queryIter.dispose()).errorOption:
+      return failure(error)
+    return success(AsyncIter[?!QueryResponse[T]].empty())
+
+  var errOccurred = false
+
+  proc genNext: Future[?!QueryResponse[T]] {.async.} =
+    let queryResOrErr = await queryIter.next()
+
+    if queryResOrErr.isErr:
+      errOccurred = true
+
+    if queryIter.finished or (errOccurred and finishOnErr):
+      trace "Disposing iterator"
+      if error =? (await queryIter.dispose()).errorOption:
+        return failure(error)
+
+    return queryResOrErr
+
+  proc isFinished(): bool =
+    queryIter.finished or (errOccurred and finishOnErr)
+
+  AsyncIter[?!QueryResponse[T]].new(genNext, isFinished).success
+
+proc filterSuccess*[T](
+  iter: AsyncIter[?!QueryResponse[T]]
+  ): Future[AsyncIter[tuple[key: Key, value: T]]] {.async.} =
+  ## Filters out any items that are not success
+
+  proc mapping(resOrErr: ?!QueryResponse[T]): Future[?KeyVal[T]] {.async.} =
+    without res =? resOrErr, error:
+      error "Error occurred when getting QueryResponse", msg = error.msg
+      return KeyVal[T].none
+
+    without key =? res.key:
+      warn "No key for a QueryResponse"
+      return KeyVal[T].none
+
+    without value =? res.value, error:
+      error "Error occurred when getting a value from QueryResponse", msg = error.msg
+      return KeyVal[T].none
+
+    (key: key, value: value).some
+
+  await mapFilter[?!QueryResponse[T], KeyVal[T]](iter, mapping)

codex/stores/treehelper.nim

@@ -47,4 +47,4 @@ proc putSomeProofs*(store: BlockStore, tree: CodexTree, iter: Iter[Natural]): Fu
   store.putSomeProofs(tree, iter.map((i: Natural) => i.ord))
 
 proc putAllProofs*(store: BlockStore, tree: CodexTree): Future[?!void] =
-  store.putSomeProofs(tree, Iter.fromSlice(0..<tree.leavesCount))
+  store.putSomeProofs(tree, Iter[int].new(0..<tree.leavesCount))

codex/utils/asynciter.nim

@@ -3,26 +3,29 @@ import std/sugar
 import pkg/questionable
 import pkg/chronos
 
-type
+import ./iter
-  Function*[T, U] = proc(fut: T): U {.raises: [CatchableError], gcsafe, noSideEffect.}
-  IsFinished* = proc(): bool {.raises: [], gcsafe, noSideEffect.}
-  GenNext*[T] = proc(): T {.raises: [CatchableError], gcsafe.}
-  Iter*[T] = ref object
-    finished: bool
-    next*: GenNext[T]
-  AsyncIter*[T] = Iter[Future[T]]
 
-proc finish*[T](self: Iter[T]): void =
+export iter
+
+## AsyncIter[T] is similar to `Iter[Future[T]]` with addition of methods specific to asynchronous processing
+##
+
+type
+  AsyncIter*[T] = ref object
+    finished: bool
+    next*: GenNext[Future[T]]
+
+proc finish*[T](self: AsyncIter[T]): void =
   self.finished = true
 
-proc finished*[T](self: Iter[T]): bool =
+proc finished*[T](self: AsyncIter[T]): bool =
   self.finished
 
-iterator items*[T](self: Iter[T]): T =
+iterator items*[T](self: AsyncIter[T]): Future[T] =
   while not self.finished:
     yield self.next()
 
-iterator pairs*[T](self: Iter[T]): tuple[key: int, val: T] {.inline.} =
+iterator pairs*[T](self: AsyncIter[T]): tuple[key: int, val: Future[T]] {.inline.} =
   var i = 0
   while not self.finished:
     yield (i, self.next())
@@ -32,14 +35,25 @@ proc map*[T, U](fut: Future[T], fn: Function[T, U]): Future[U] {.async.} =
   let t = await fut
   fn(t)
 
-proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOnErr: bool = true): Iter[T] =
+proc flatMap*[T, U](fut: Future[T], fn: Function[T, Future[U]]): Future[U] {.async.} =
-  var iter = Iter[T]()
+  let t = await fut
+  await fn(t)
 
-  proc next(): T {.raises: [CatchableError].} =
+proc new*[T](_: type AsyncIter[T], genNext: GenNext[Future[T]], isFinished: IsFinished, finishOnErr: bool = true): AsyncIter[T] =
+  ## Creates a new Iter using elements returned by supplier function `genNext`.
+  ## Iter is finished whenever `isFinished` returns true.
+  ##
+
+  var iter = AsyncIter[T]()
+
+  proc next(): Future[T] {.async.} =
     if not iter.finished:
       var item: T
       try:
-        item = genNext()
+        item = await genNext()
+      except CancelledError as err:
+        iter.finish
+        raise err
       except CatchableError as err:
         if finishOnErr or isFinished():
           iter.finish
@@ -49,7 +63,7 @@ proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOn
         iter.finish
       return item
     else:
-      raise newException(CatchableError, "Iterator is finished but next item was requested")
+      raise newException(CatchableError, "AsyncIter is finished but next item was requested")
 
   if isFinished():
     iter.finish
@@ -57,90 +71,95 @@ proc new*[T](_: type Iter, genNext: GenNext[T], isFinished: IsFinished, finishOn
   iter.next = next
   return iter
 
-proc fromItems*[T](_: type Iter, items: seq[T]): Iter[T] =
+proc mapAsync*[T, U](iter: Iter[T], fn: Function[T, Future[U]]): AsyncIter[U] =
-  ## Create new iterator from items
+  AsyncIter[U].new(
+    genNext = () => fn(iter.next()),
+    isFinished = () => iter.finished()
+  )
+
+proc new*[U, V: Ordinal](_: type AsyncIter[U], slice: HSlice[U, V]): AsyncIter[U] =
+  ## Creates new Iter from a slice
   ##
 
-  Iter.fromSlice(0..<items.len)
+  let iter = Iter[U].new(slice)
-    .map((i: int) => items[i])
+  mapAsync[U, U](iter,
+    proc (i: U): Future[U] {.async.} =
+      i
+  )
 
-proc fromSlice*[U, V: Ordinal](_: type Iter, slice: HSlice[U, V]): Iter[U] =
+proc new*[U, V, S: Ordinal](_: type AsyncIter[U], a: U, b: V, step: S = 1): AsyncIter[U] =
-  ## Creates new iterator from slice
+  ## Creates new Iter in range a..b with specified step (default 1)
   ##
 
-  Iter.fromRange(slice.a.int, slice.b.int, 1)
+  let iter = Iter[U].new(a, b, step)
+  mapAsync[U, U](iter,
+    proc (i: U): Future[U] {.async.} =
+      i
+  )
 
-proc fromRange*[U, V, S: Ordinal](_: type Iter, a: U, b: V, step: S = 1): Iter[U] =
+proc empty*[T](_: type AsyncIter[T]): AsyncIter[T] =
-  ## Creates new iterator in range a..b with specified step (default 1)
+  ## Creates an empty AsyncIter
   ##
 
-  var i = a
+  proc genNext(): Future[T] {.raises: [CatchableError].} =
+    raise newException(CatchableError, "Next item requested from an empty AsyncIter")
+  proc isFinished(): bool = true
 
-  proc genNext(): U =
+  AsyncIter[T].new(genNext, isFinished)
-    let u = i
-    inc(i, step)
-    u
 
-  proc isFinished(): bool =
+proc map*[T, U](iter: AsyncIter[T], fn: Function[T, Future[U]]): AsyncIter[U] =
-    (step > 0 and i > b) or
+  AsyncIter[U].new(
-      (step < 0 and i < b)
+    genNext    = () => iter.next().flatMap(fn),
-
-  Iter.new(genNext, isFinished)
-
-proc map*[T, U](iter: Iter[T], fn: Function[T, U]): Iter[U] =
-  Iter.new(
-    genNext    = () => fn(iter.next()),
     isFinished = () => iter.finished
   )
 
-proc filter*[T](iter: Iter[T], predicate: Function[T, bool]): Iter[T] =
+proc mapFilter*[T, U](iter: AsyncIter[T], mapPredicate: Function[T, Future[Option[U]]]): Future[AsyncIter[U]] {.async.} =
-  var nextT: Option[T]
+  var nextFutU: Option[Future[U]]
 
-  proc tryFetch(): void =
+  proc tryFetch(): Future[void] {.async.} =
-    nextT = T.none
+    nextFutU = Future[U].none
     while not iter.finished:
-      let t = iter.next()
+      let futT = iter.next()
-      if predicate(t):
+      try:
-        nextT = some(t)
+        if u =? await futT.flatMap(mapPredicate):
+          let futU = newFuture[U]("AsyncIter.mapFilterAsync")
+          futU.complete(u)
+          nextFutU = some(futU)
+          break
+      except CancelledError as err:
+        raise err
+      except CatchableError as err:
+        let errFut = newFuture[U]("AsyncIter.mapFilterAsync")
+        errFut.fail(err)
+        nextFutU = some(errFut)
         break
 
-  proc genNext(): T =
+  proc genNext(): Future[U] {.async.} =
-    let t = nextT.unsafeGet
+    let futU = nextFutU.unsafeGet
-    tryFetch()
+    await tryFetch()
-    return t
+    await futU
 
   proc isFinished(): bool =
-    nextT.isNone
+    nextFutU.isNone
 
-  tryFetch()
+  await tryFetch()
-  Iter.new(genNext, isFinished)
+  AsyncIter[U].new(genNext, isFinished)
 
-proc prefetch*[T](iter: Iter[T], n: Positive): Iter[T] =
+proc filter*[T](iter: AsyncIter[T], predicate: Function[T, Future[bool]]): Future[AsyncIter[T]] {.async.} =
-  var ringBuf = newSeq[T](n)
+  proc wrappedPredicate(t: T): Future[Option[T]] {.async.} =
-  var iterLen = int.high
+    if await predicate(t):
-  var i = 0
+      some(t)
-
-  proc tryFetch(j: int): void =
-    if not iter.finished:
-      let item = iter.next()
-      ringBuf[j mod n] = item
-      if iter.finished:
-        iterLen = min(j + 1, iterLen)
     else:
-      if j == 0:
+      T.none
-        iterLen = 0
 
-  proc genNext(): T =
+  await mapFilter[T, T](iter, wrappedPredicate)
-    let item = ringBuf[i mod n]
-    tryFetch(i + n)
-    inc i
-    return item
 
-  proc isFinished(): bool =
+proc delayBy*[T](iter: AsyncIter[T], d: Duration): AsyncIter[T] =
-    i >= iterLen
+  ## Delays emitting each item by given duration
+  ##
 
-  # initialize ringBuf with n prefetched values
+  map[T, T](iter,
-  for j in 0..<n:
+    proc (t: T): Future[T] {.async.} =
-    tryFetch(j)
+      await sleepAsync(d)
-
+      t
-  Iter.new(genNext, isFinished)
+  )

codex/utils/iter.nim

@@ -0,0 +1,140 @@
+import std/sugar
+
+import pkg/questionable
+import pkg/questionable/results
+
+type
+  Function*[T, U] = proc(fut: T): U {.raises: [CatchableError], gcsafe, closure.}
+  IsFinished* = proc(): bool {.raises: [], gcsafe, closure.}
+  GenNext*[T] = proc(): T {.raises: [CatchableError], gcsafe.}
+  Iter*[T] = ref object
+    finished: bool
+    next*: GenNext[T]
+
+proc finish*[T](self: Iter[T]): void =
+  self.finished = true
+
+proc finished*[T](self: Iter[T]): bool =
+  self.finished
+
+iterator items*[T](self: Iter[T]): T =
+  while not self.finished:
+    yield self.next()
+
+iterator pairs*[T](self: Iter[T]): tuple[key: int, val: T] {.inline.} =
+  var i = 0
+  while not self.finished:
+    yield (i, self.next())
+    inc(i)
+
+proc new*[T](_: type Iter[T], genNext: GenNext[T], isFinished: IsFinished, finishOnErr: bool = true): Iter[T] =
+  ## Creates a new Iter using elements returned by supplier function `genNext`.
+  ## Iter is finished whenever `isFinished` returns true.
+  ##
+
+  var iter = Iter[T]()
+
+  proc next(): T {.raises: [CatchableError].} =
+    if not iter.finished:
+      var item: T
+      try:
+        item = genNext()
+      except CatchableError as err:
+        if finishOnErr or isFinished():
+          iter.finish
+        raise err
+
+      if isFinished():
+        iter.finish
+      return item
+    else:
+      raise newException(CatchableError, "Iter is finished but next item was requested")
+
+  if isFinished():
+    iter.finish
+
+  iter.next = next
+  return iter
+
+proc new*[U, V, S: Ordinal](_: type Iter[U], a: U, b: V, step: S = 1): Iter[U] =
+  ## Creates a new Iter in range a..b with specified step (default 1)
+  ##
+
+  var i = a
+
+  proc genNext(): U =
+    let u = i
+    inc(i, step)
+    u
+
+  proc isFinished(): bool =
+    (step > 0 and i > b) or
+      (step < 0 and i < b)
+
+  Iter[U].new(genNext, isFinished)
+
+proc new*[U, V: Ordinal](_: type Iter[U], slice: HSlice[U, V]): Iter[U] =
+  ## Creates a new Iter from a slice
+  ##
+
+  Iter[U].new(slice.a.int, slice.b.int, 1)
+
+proc new*[T](_: type Iter[T], items: seq[T]): Iter[T] =
+  ## Creates a new Iter from a sequence
+  ##
+
+  Iter[int].new(0..<items.len)
+    .map((i: int) => items[i])
+
+proc empty*[T](_: type Iter[T]): Iter[T] =
+  ## Creates an empty Iter
+  ##
+
+  proc genNext(): T {.raises: [CatchableError].} =
+    raise newException(CatchableError, "Next item requested from an empty Iter")
+  proc isFinished(): bool = true
+
+  Iter[T].new(genNext, isFinished)
+
+proc map*[T, U](iter: Iter[T], fn: Function[T, U]): Iter[U] =
+  Iter[U].new(
+    genNext    = () => fn(iter.next()),
+    isFinished = () => iter.finished
+  )
+
+proc mapFilter*[T, U](iter: Iter[T], mapPredicate: Function[T, Option[U]]): Iter[U] =
+  var nextUOrErr: Option[Result[U, ref CatchableError]]
+
+  proc tryFetch(): void =
+    nextUOrErr = Result[U, ref CatchableError].none
+    while not iter.finished:
+      try:
+        let t = iter.next()
+        if u =? mapPredicate(t):
+          nextUOrErr = some(success(u))
+          break
+      except CatchableError as err:
+        nextUOrErr = some(U.failure(err))
+
+  proc genNext(): U {.raises: [CatchableError].} =
+    # at this point nextUOrErr should always be some(..)
+    without u =? nextUOrErr.unsafeGet, err:
+      raise err
+
+    tryFetch()
+    return u
+
+  proc isFinished(): bool =
+    nextUOrErr.isNone
+
+  tryFetch()
+  Iter[U].new(genNext, isFinished)
+
+proc filter*[T](iter: Iter[T], predicate: Function[T, bool]): Iter[T] =
+  proc wrappedPredicate(t: T): Option[T] =
+    if predicate(t):
+      some(t)
+    else:
+      T.none
+
+  mapFilter[T, T](iter, wrappedPredicate)

tests/codex/stores/testqueryiterhelper.nim

@@ -0,0 +1,65 @@
+import std/sugar
+
+import pkg/stew/results
+import pkg/questionable
+import pkg/chronos
+import pkg/datastore/typedds
+import pkg/datastore/sql/sqliteds
+import pkg/codex/stores/queryiterhelper
+import pkg/codex/utils/asynciter
+
+import ../../asynctest
+import ../helpers
+
+proc encode(s: string): seq[byte] =
+  s.toBytes()
+
+proc decode(T: type string, bytes: seq[byte]): ?!T =
+  success(string.fromBytes(bytes))
+
+asyncchecksuite "Test QueryIter helper":
+  var
+    tds: TypedDatastore
+
+  setupAll:
+    tds = TypedDatastore.init(SQLiteDatastore.new(Memory).tryGet())
+
+  teardownAll:
+    (await tds.close()).tryGet
+
+  test "Should auto-dispose when QueryIter finishes":
+    let
+      source = {
+        "a": "11",
+        "b": "22"
+      }.toTable
+      Root = Key.init("/queryitertest").tryGet()
+
+    for k, v in source:
+      let key = (Root / k).tryGet()
+      (await tds.put(key, v)).tryGet()
+
+    var
+      disposed = false
+      queryIter = (await query[string](tds, Query.init(Root))).tryGet()
+
+    let iterDispose: IterDispose = queryIter.dispose
+    queryIter.dispose = () => (disposed = true; iterDispose())
+
+    let
+      iter1 = (await toAsyncIter[string](queryIter)).tryGet()
+      iter2 = await filterSuccess[string](iter1)
+
+    var items = initTable[string, string]()
+
+    for fut in iter2:
+      let item = await fut
+
+      items[item.key.value] = item.value
+
+    check:
+      items == source
+      disposed == true
+      queryIter.finished == true
+      iter1.finished == true
+      iter2.finished == true

tests/codex/teststores.nim

@@ -1,5 +1,6 @@
 import ./stores/testcachestore
 import ./stores/testrepostore
 import ./stores/testmaintenance
+import ./stores/testqueryiterhelper
 
 {.warning[UnusedImport]: off.}

tests/codex/testutils.nim

@@ -1,6 +1,7 @@
 import ./utils/testoptions
 import ./utils/testkeyutils
 import ./utils/testasyncstatemachine
+import ./utils/testasynciter
 import ./utils/testtimer
 import ./utils/testthen
 import ./utils/testtrackedfutures

tests/codex/utils/testasynciter.nim

@@ -0,0 +1,160 @@
+import std/sugar
+
+import pkg/questionable
+import pkg/chronos
+import pkg/codex/utils/asynciter
+
+import ../../asynctest
+import ../helpers
+
+asyncchecksuite "Test AsyncIter":
+
+  test "Should be finished":
+    let iter = AsyncIter[int].empty()
+
+    check:
+      iter.finished == true
+
+  test "Should map each item using `map`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = map[int, string](iter1,
+        proc (i: int): Future[string] {.async.} =
+          $i
+      )
+
+    var collected: seq[string]
+
+    for fut in iter2:
+      collected.add(await fut)
+
+    check:
+      collected == @["0", "1", "2", "3", "4"]
+
+  test "Should leave only odd items using `filter`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = await filter[int](iter1,
+        proc (i: int): Future[bool] {.async.} =
+          (i mod 2) == 1
+      )
+
+    var collected: seq[int]
+
+    for fut in iter2:
+      collected.add(await fut)
+
+    check:
+      collected == @[1, 3]
+
+  test "Should leave only odd items using `mapFilter`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = await mapFilter[int, string](iter1,
+        proc (i: int): Future[?string] {.async.} =
+          if (i mod 2) == 1:
+            some($i)
+          else:
+            string.none
+      )
+
+    var collected: seq[string]
+
+    for fut in iter2:
+      collected.add(await fut)
+
+    check:
+      collected == @["1", "3"]
+
+  test "Should yield all items before err using `map`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = map[int, string](iter1,
+          proc (i: int): Future[string] {.async.} =
+            if i < 3:
+              return $i
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[string]
+
+    expect CatchableError:
+      for fut in iter2:
+        collected.add(await fut)
+
+    check:
+      collected == @["0", "1", "2"]
+      iter2.finished
+
+  test "Should yield all items before err using `filter`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = await filter[int](iter1,
+          proc (i: int): Future[bool] {.async.} =
+            if i < 3:
+              return true
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[int]
+
+    expect CatchableError:
+      for fut in iter2:
+        collected.add(await fut)
+
+    check:
+      collected == @[0, 1, 2]
+      iter2.finished
+
+  test "Should yield all items before err using `mapFilter`":
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = await mapFilter[int, string](iter1,
+          proc (i: int): Future[?string] {.async.} =
+            if i < 3:
+              return some($i)
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[string]
+
+    expect CatchableError:
+      for fut in iter2:
+        collected.add(await fut)
+
+    check:
+      collected == @["0", "1", "2"]
+      iter2.finished
+
+  test "Should propagate cancellation error immediately":
+    let
+      fut = newFuture[?string]("testasynciter")
+
+    let
+      iter1 = AsyncIter[int].new(0..<5).delayBy(10.millis)
+      iter2 = await mapFilter[int, string](iter1,
+          proc (i: int): Future[?string] {.async.} =
+            if i < 3:
+              return some($i)
+            else:
+              return await fut
+        )
+
+    proc cancelFut(): Future[void] {.async.} =
+      await sleepAsync(100.millis)
+      await fut.cancelAndWait()
+
+    asyncSpawn(cancelFut())
+
+    var collected: seq[string]
+
+    expect CancelledError:
+      for fut in iter2:
+        collected.add(await fut)
+
+    check:
+      collected == @["0", "1"]
+      iter2.finished

tests/codex/utils/testiter.nim

@@ -0,0 +1,129 @@
+import std/sugar
+
+import pkg/questionable
+import pkg/chronos
+import pkg/codex/utils/iter
+
+import ../../asynctest
+import ../helpers
+
+checksuite "Test Iter":
+
+  test "Should be finished":
+    let iter = Iter[int].empty()
+
+    check:
+      iter.finished == true
+
+  test "Should be iterable with `items`":
+    let iter = Iter.new(0..<5)
+
+    let items =
+      collect:
+        for v in iter:
+          v
+
+    check:
+      items == @[0, 1, 2, 3, 4]
+
+  test "Should be iterable with `pairs`":
+    let iter = Iter.new(0..<5)
+
+    let pairs =
+      collect:
+        for i, v in iter:
+          (i, v)
+
+    check:
+      pairs == @[(0, 0), (1, 1), (2, 2), (3, 3), (4, 4)]
+
+  test "Should map each item using `map`":
+    let iter = Iter.new(0..<5)
+      .map((i: int) => $i)
+
+    check:
+      iter.toSeq() == @["0", "1", "2", "3", "4"]
+
+  test "Should leave only odd items using `filter`":
+    let iter = Iter.new(0..<5)
+      .filter((i: int) => (i mod 2) == 1)
+
+    check:
+      iter.toSeq() == @[1, 3]
+
+  test "Should leave only odd items using `mapFilter`":
+    let
+      iter1 = Iter.new(0..<5)
+      iter2 = mapFilter[int, string](iter1,
+        proc(i: int): ?string =
+          if (i mod 2) == 1:
+            some($i)
+          else:
+            string.none
+      )
+
+    check:
+      iter2.toSeq() == @["1", "3"]
+
+  test "Should yield all items before err using `map`":
+    let
+      iter = Iter.new(0..<5)
+       .map(
+          proc (i: int): string =
+            if i < 3:
+              return $i
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[string]
+
+    expect CatchableError:
+      for i in iter:
+        collected.add(i)
+
+    check:
+      collected == @["0", "1", "2"]
+      iter.finished
+
+  test "Should yield all items before err using `filter`":
+    let
+      iter = Iter.new(0..<5)
+       .filter(
+          proc (i: int): bool =
+            if i < 3:
+              return true
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[int]
+
+    expect CatchableError:
+      for i in iter:
+        collected.add(i)
+
+    check:
+      collected == @[0, 1, 2]
+      iter.finished
+
+  test "Should yield all items before err using `mapFilter`":
+    let
+      iter1 = Iter.new(0..<5)
+      iter2 = mapFilter[int, string](iter1,
+          proc (i: int): ?string =
+            if i < 3:
+              return some($i)
+            else:
+              raise newException(CatchableError, "Some error")
+        )
+
+    var collected: seq[string]
+
+    expect CatchableError:
+      for i in iter2:
+        collected.add(i)
+
+    check:
+      collected == @["0", "1", "2"]
+      iter2.finished

vendor/nim-datastore

@@ -1 +1 @@
-Subproject commit f4989fcce5d74a648e7e2598a72a7b21948f4a85
+Subproject commit 3ab6b84a634a7b2ee8c0144f050bf5893cd47c17