checked-exceptions-cont

.github/actions/nimbus-build-system/action.yml

@@ -37,6 +37,7 @@ runs:
         sudo apt-get update -qq
         sudo DEBIAN_FRONTEND='noninteractive' apt-get install \
           --no-install-recommends -yq lcov
+        sudo apt install libpcre3
 
     - name: APT (Linux i386)
       if: inputs.os == 'linux' && inputs.cpu == 'i386'
@@ -83,7 +84,7 @@ runs:
 
     - name: Install gcc 14 on Linux
       # We don't want to install gcc 14 for coverage (Ubuntu 20.04)
-      if : ${{ inputs.os == 'linux' && inputs.coverage != 'true' }}
+      if: ${{ inputs.os == 'linux' && inputs.coverage != 'true' }}
       shell: ${{ inputs.shell }} {0}
       run: |
         # Skip for older Ubuntu versions
@@ -202,7 +203,7 @@ runs:
     - name: Restore Nim toolchain binaries from cache
       id: nim-cache
       uses: actions/cache@v4
-      if : ${{ inputs.coverage != 'true'  }}
+      if: ${{ inputs.coverage != 'true'  }}
       with:
         path: NimBinaries
         key: ${{ inputs.os }}-${{ inputs.cpu }}-nim-${{ inputs.nim_version }}-cache-${{ env.cache_nonce }}-${{ github.run_id }}

Makefile

@@ -233,6 +233,11 @@ format:
 	$(NPH) codex/
 	$(NPH) tests/
 
+format-global:
+	nph *.nim
+	nph codex/
+	nph tests/
+
 clean-nph:
 	rm -f $(NPH)
 

codex/erasure/erasure.nim

@@ -28,7 +28,6 @@
 import ../clock
 import ../blocktype as bt
 import ../utils
-import ../utils/asynciter
 import ../indexingstrategy
 import ../errors
 import ../utils/arrayutils
@@ -122,14 +121,15 @@
 
 proc getPendingBlocks(
     self: Erasure, manifest: Manifest, indices: seq[int]
-): AsyncIter[(?!bt.Block, int)] =
+): AsyncResultIter[(?!bt.Block, int)] =
   ## Get pending blocks iterator
   ##
-  var pendingBlocks: seq[Future[(?!bt.Block, int)]] = @[]
+
+  var pendingBlocks: seq[Future[(?!bt.Block, int)].Raising([CancelledError])] = @[]
 
   proc attachIndex(
-      fut: Future[?!bt.Block], i: int
-  ): Future[(?!bt.Block, int)] {.async.} =
+      fut: Future[?!bt.Block].Raising([CancelledError]), i: int
+  ): Future[(?!bt.Block, int)] {.async: (raises: [CancelledError]).} =
     ## avoids closure capture issues
     return (await fut, i)
 
@@ -141,20 +141,23 @@
   proc isFinished(): bool =
     pendingBlocks.len == 0
 
-  proc genNext(): Future[(?!bt.Block, int)] {.async.} =
-    let completedFut = await one(pendingBlocks)
-    if (let i = pendingBlocks.find(completedFut); i >= 0):
-      pendingBlocks.del(i)
-      return await completedFut
-    else:
-      let (_, index) = await completedFut
-      raise newException(
-        CatchableError,
-        "Future for block id not found, tree cid: " & $manifest.treeCid & ", index: " &
-          $index,
-      )
+  proc genNext(): Future[?!(?!bt.Block, int)] {.async: (raises: [CancelledError]).} =
+    try:
+      let completedFut = await one(pendingBlocks)
+      if (let i = pendingBlocks.find(completedFut); i >= 0):
+        pendingBlocks.del(i)
+        return success(await completedFut)
+      else:
+        let (_, index) = await completedFut
+        return failure(
+          "Future for block id not found, tree cid: " & $manifest.treeCid & ", index: " &
+            $index
+        )
+    except ValueError as err:
+      # ValueError is raised by `one` when the pendingBlocks is empty
+      return failure("iterator finished (pendingBlocks empty) but genNext() was called")
 
-  AsyncIter[(?!bt.Block, int)].new(genNext, isFinished)
+  AsyncResultIter[(?!bt.Block, int)].new(genNext, isFinished)
 
 proc prepareEncodingData(
     self: Erasure,
@@ -164,7 +167,7 @@
     data: ref seq[seq[byte]],
     cids: ref seq[Cid],
     emptyBlock: seq[byte],
-): Future[?!Natural] {.async.} =
+): Future[?!Natural] {.async: (raises: [CancelledError, IndexingError]).} =
   ## Prepare data for encoding
   ##
 
@@ -178,7 +181,8 @@
 
   var resolved = 0
   for fut in pendingBlocksIter:
-    let (blkOrErr, idx) = await fut
+    without (blkOrErr, idx) =? (await fut), err:
+      return failure(err)
     without blk =? blkOrErr, err:
       warn "Failed retrieving a block", treeCid = manifest.treeCid, idx, msg = err.msg
       return failure(err)
@@ -208,7 +212,7 @@
     parityData: ref seq[seq[byte]],
     cids: ref seq[Cid],
     emptyBlock: seq[byte],
-): Future[?!(Natural, Natural)] {.async.} =
+): Future[?!(Natural, Natural)] {.async: (raises: [CancelledError, IndexingError]).} =
   ## Prepare data for decoding
   ## `encoded`    - the encoded manifest
   ## `step`       - the current step
@@ -235,7 +239,8 @@
     if resolved >= encoded.ecK:
       break
 
-    let (blkOrErr, idx) = await fut
+    without (blkOrErr, idx) =? (await fut), err:
+      return failure(err)
     without blk =? blkOrErr, err:
       trace "Failed retrieving a block", idx, treeCid = encoded.treeCid, msg = err.msg
       continue
@@ -362,7 +367,7 @@
 
 proc encodeData(
     self: Erasure, manifest: Manifest, params: EncodingParams
-): Future[?!Manifest] {.async.} =
+): Future[?!Manifest] {.async: (raises: [CancelledError]).} =
   ## Encode blocks pointed to by the protected manifest
   ##
   ## `manifest` - the manifest to encode
@@ -403,15 +408,12 @@
 
       trace "Erasure coding data", data = data[].len
 
-      try:
-        if err =? (
-          await self.asyncEncode(
-            manifest.blockSize.int, params.ecK, params.ecM, data, parity
-          )
-        ).errorOption:
-          return failure(err)
-      except CancelledError as exc:
-        raise exc
+      if err =? (
+        await self.asyncEncode(
+          manifest.blockSize.int, params.ecK, params.ecM, data, parity
+        )
+      ).errorOption:
+        return failure(err)
 
       var idx = params.rounded + step
       for j in 0 ..< params.ecM:
@@ -451,17 +453,17 @@
   except CancelledError as exc:
     trace "Erasure coding encoding cancelled"
     raise exc # cancellation needs to be propagated
-  except CatchableError as exc:
-    trace "Erasure coding encoding error", exc = exc.msg
-    return failure(exc)
+  except IndexingError as err:
+    trace "Erasure coding encoding indexing error", error = err.msg
+    return failure(err)
 
 proc encode*(
     self: Erasure,
     manifest: Manifest,
     blocks: Natural,
     parity: Natural,
     strategy = SteppedStrategy,
-): Future[?!Manifest] {.async.} =
+): Future[?!Manifest] {.async: (raises: [CancelledError]).} =
   ## Encode a manifest into one that is erasure protected.
   ##
   ## `manifest`   - the original manifest to be encoded
@@ -554,7 +556,7 @@
 
 proc decodeInternal(
     self: Erasure, encoded: Manifest
-): Future[?!(ref seq[Cid], seq[Natural])] {.async.} =
+): Future[?!(ref seq[Cid], seq[Natural])] {.async: (raises: [CancelledError]).} =
   logScope:
     steps = encoded.steps
     rounded_blocks = encoded.rounded
@@ -597,15 +599,12 @@
         continue
 
       trace "Erasure decoding data"
-      try:
-        if err =? (
-          await self.asyncDecode(
-            encoded.blockSize.int, encoded.ecK, encoded.ecM, data, parityData, recovered
-          )
-        ).errorOption:
-          return failure(err)
-      except CancelledError as exc:
-        raise exc
+      if err =? (
+        await self.asyncDecode(
+          encoded.blockSize.int, encoded.ecK, encoded.ecM, data, parityData, recovered
+        )
+      ).errorOption:
+        return failure(err)
 
       for i in 0 ..< encoded.ecK:
         let idx = i * encoded.steps + step
@@ -630,9 +629,9 @@
   except CancelledError as exc:
     trace "Erasure coding decoding cancelled"
     raise exc # cancellation needs to be propagated
-  except CatchableError as exc:
-    trace "Erasure coding decoding error", exc = exc.msg
-    return failure(exc)
+  except IndexingError as err:
+    trace "Erasure coding decoding indexing error", error = err.msg
+    return failure(err)
   finally:
     decoder.release()
 

codex/indexingstrategy.nim

@@ -1,6 +1,6 @@
 import ./errors
 import ./utils
-import ./utils/asynciter
+import ./utils/iter
 
 {.push raises: [].}
 

codex/node.nim

@@ -44,7 +44,6 @@ import ./indexingstrategy
 import ./utils
 import ./errors
 import ./logutils
-import ./utils/asynciter
 import ./utils/trackedfutures
 
 export logutils

codex/slots/builder/builder.nim

@@ -24,12 +24,11 @@ import ../../utils
 import ../../stores
 import ../../manifest
 import ../../merkletree
-import ../../utils/asynciter
 import ../../indexingstrategy
 
 import ../converters
 
-export converters, asynciter
+export converters, iter
 
 logScope:
   topics = "codex slotsbuilder"

codex/stores/blockstore.nim

@@ -154,7 +154,7 @@ method hasBlock*(
 
 method listBlocks*(
     self: BlockStore, blockType = BlockType.Manifest
-): Future[?!SafeAsyncIter[Cid]] {.base, async: (raises: [CancelledError]), gcsafe.} =
+): Future[?!AsyncResultIter[Cid]] {.base, async: (raises: [CancelledError]), gcsafe.} =
   ## Get the list of blocks in the BlockStore. This is an intensive operation
   ##
 

codex/stores/cachestore.nim

@@ -139,7 +139,7 @@ func cids(self: CacheStore): (iterator (): Cid {.gcsafe, raises: [].}) =
 
 method listBlocks*(
     self: CacheStore, blockType = BlockType.Manifest
-): Future[?!SafeAsyncIter[Cid]] {.async: (raises: [CancelledError]).} =
+): Future[?!AsyncResultIter[Cid]] {.async: (raises: [CancelledError]).} =
   ## Get the list of blocks in the BlockStore. This is an intensive operation
   ##
 
@@ -152,7 +152,7 @@ method listBlocks*(
     success(cids())
 
   let iter = await (
-    SafeAsyncIter[Cid].new(genNext, isFinished).filter(
+    AsyncResultIter[Cid].new(genNext, isFinished).filter(
       proc(cid: ?!Cid): Future[bool] {.async: (raises: [CancelledError]).} =
         without cid =? cid, err:
           trace "Cannot get Cid from the iterator", err = err.msg

codex/stores/maintenance.nim

@@ -18,7 +18,7 @@ import pkg/questionable/results
 
 import ./repostore
 import ../utils/timer
-import ../utils/safeasynciter
+import ../utils/asyncresultiter
 import ../clock
 import ../logutils
 import ../systemclock

codex/stores/networkstore.nim

@@ -19,7 +19,7 @@
 import ../logutils
 import ../merkletree
 import ../utils/asyncheapqueue
-import ../utils/safeasynciter
+import ../utils/asyncresultiter
 import ./blockstore
 
 export blockstore, blockexchange, asyncheapqueue
@@ -127,7 +127,7 @@
 
 method listBlocks*(
     self: NetworkStore, blockType = BlockType.Manifest
-): Future[?!SafeAsyncIter[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
+): Future[?!AsyncResultIter[Cid]] {.async: (raw: true, raises: [CancelledError]).} =
   self.localStore.listBlocks(blockType)
 
 method delBlock*(