Feat/overlay support

.github/workflows/ci-reusable.yml

@@ -20,6 +20,7 @@ env:
 jobs:
   build:
     strategy:
+      fail-fast: false
       matrix:
         include: ${{ fromJson(inputs.matrix) }}
 

.gitignore

@@ -44,3 +44,9 @@ docker/prometheus-data
 data/
 nimbledeps
 logs/
+.grepai/
+*.log
+CLAUDE.md
+AGENTS.md
+.claude
+.opencode

.gitmodules

@@ -96,12 +96,6 @@
 [submodule "vendor/nimble/sqlite3_abi"]
 	path = vendor/nimble/sqlite3_abi
 	url = https://github.com/arnetheduck/nim-sqlite3-abi
-[submodule "vendor/nimble/leveldbstatic"]
-	path = vendor/nimble/leveldbstatic
-	url = https://github.com/durability-labs/nim-leveldbstatic
-[submodule "vendor/nimble/datastore"]
-	path = vendor/nimble/datastore
-	url = https://github.com/durability-labs/nim-datastore
 [submodule "vendor/nimble/archivistdht"]
 	path = vendor/nimble/archivistdht
 	url = https://github.com/durability-labs/archivist-dht
@@ -159,3 +153,10 @@
 [submodule "vendor/nimble/zippy"]
 	path = vendor/nimble/zippy
 	url = https://github.com/guzba/zippy
+[submodule "vendor/nimble/nim-kvstore"]
+	path = vendor/nimble/nim-kvstore
+	url = git@github.com:durability-labs/nim-kvstore.git
+	url = https://github.com/durability-labs/nim-kvstore.git
+[submodule "vendor/nimble/threading"]
+	path = vendor/nimble/threading
+	url = https://github.com/nim-lang/threading.git

archivist.nim

@@ -147,6 +147,7 @@ when isMainModule:
       chronos.poll()
     except Exception as exc:
       error "Unhandled exception in async proc, aborting", msg = exc.msg
+      # raise exc # uncomment for stack trace
       quit QuitFailure
 
   try:
@@ -155,6 +156,7 @@ when isMainModule:
     waitFor shutdown
   except CatchableError as error:
     error "Archivist Node didn't shutdown correctly", error = error.msg
+    # raise exc # uncomment for stacktrace
     quit QuitFailure
 
   notice "Exited Archivist Node"

archivist.nim.cfg

@@ -4,3 +4,5 @@
 --define:"chronicles_sinks=textlines[dynamic],json[dynamic],textlines[dynamic]"
 # enable metrics collection
 --define:metrics
+# bypass Nim TLSF allocator to avoid cross-thread free accumulation
+--define:useMalloc

archivist.nimble

@@ -11,22 +11,18 @@ import "./vendor/nimble/deps.nims"
 before build:
   exec "nim vendor" / "nimble" / "install.nims"
 
-task update, "Sync submodules to pinned commits (safe)":
-  exec "git submodule sync --recursive"
-  exec "git submodule update --init --recursive"
-
-task updateUnsafe, "Reset and clean submodules to pinned commits (destructive)":
-  exec "git submodule sync --recursive"
-  exec "git submodule foreach --recursive 'git reset --hard'"
-  exec "git submodule foreach --recursive 'git clean -fdx'"
-  exec "git submodule update --init --recursive --force"
+before test:
+  exec "nim vendor" / "nimble" / "install.nims"
 
 task test, "Run node tests":
   exec "nim c -r tests" / "testNode"
 
 task testContracts, "Run contract tests":
   exec "nim c -r tests" / "testContracts"
 
+before testIntegration:
+  exec "nim vendor" / "nimble" / "install.nims"
+
 task testIntegration, "Run integration tests":
   exec "nim c" &
     " --define:release" &
@@ -52,6 +48,16 @@ task format, "Format code using NPH":
   exec findExe("nph") & " tests/"
   exec findExe("nph") & " tools/"
 
+task syncModules, "Sync submodules to pinned commits (safe)":
+  exec "git submodule sync --recursive"
+  exec "git submodule update --init --recursive"
+
+task syncUnsafe, "Reset and clean submodules to pinned commits (destructive)":
+  exec "git submodule sync --recursive"
+  exec "git submodule foreach --recursive 'git reset --hard'"
+  exec "git submodule foreach --recursive 'git clean -fdx'"
+  exec "git submodule update --init --recursive --force"
+
 task addDep, "Add vendored Nim dependency (git submodule)":
   addDepTask(thisDir())
 

archivist/archivist.nim

@@ -19,7 +19,7 @@ import pkg/confutils
 import pkg/confutils/defs
 import pkg/nitro
 import pkg/stew/io2
-import pkg/datastore
+import pkg/kvstore
 import pkg/ethers except Rng
 
 import ./node
@@ -168,8 +168,8 @@ proc new*(
     )
 
   let
-    discoveryStore = Datastore(
-      LevelDbDatastore.new(config.dataDir / ArchivistDhtProvidersNamespace).expect(
+    discoveryStore = KVStore(
+      SQLiteKVStore.new(config.dataDir / ArchivistDhtProvidersNamespace, tp).expect(
         "Should create discovery datastore!"
       )
     )
@@ -185,40 +185,41 @@ proc new*(
     wallet = WalletRef.new(EthPrivateKey.random())
     network = BlockExcNetwork.new(switch)
 
-    repoData =
+    repoData: KVStore =
       case config.repoKind
       of repoFS:
-        Datastore(
-          FSDatastore.new($config.dataDir, depth = 5).expect(
-            "Should create repo file data store!"
+        KVStore(
+          FSKVStore
+          .new(
+            $config.dataDir,
+            tp,
+            depth = 5,
+            directIO = config.fsDirectIO,
+            fsyncFile = config.fsFsyncFile,
+            fsyncDir = config.fsFsyncDir,
           )
+          .expect("Should create repo file data store!")
         )
       of repoSQLite:
-        Datastore(
-          SQLiteDatastore.new($config.dataDir).expect(
+        KVStore(
+          SQLiteKVStore.new($config.dataDir, tp).expect(
             "Should create repo SQLite data store!"
           )
         )
-      of repoLevelDb:
-        Datastore(
-          LevelDbDatastore.new($config.dataDir).expect(
-            "Should create repo LevelDB data store!"
-          )
-        )
 
     repoStore = RepoStore.new(
       repoDs = repoData,
-      metaDs = LevelDbDatastore.new(config.dataDir / ArchivistMetaNamespace).expect(
+      metaDs = SQLiteKVStore.new(config.dataDir / ArchivistMetaNamespace, tp).expect(
           "Should create metadata store!"
         ),
       quotaMaxBytes = config.storageQuota,
-      blockTtl = config.blockTtl,
+      overlayTtl = config.overlayTtl.seconds,
     )
 
     maintenance = BlockMaintainer.new(
       repoStore,
-      interval = config.blockMaintenanceInterval,
-      numberOfBlocksPerInterval = config.blockMaintenanceNumberOfBlocks,
+      interval = config.overlayMaintenanceInterval,
+      numberOfBlocksPerInterval = config.overlayMaintenanceNumberOfBlocks,
     )
 
     peerStore = PeerCtxStore.new()
@@ -240,6 +241,7 @@ proc new*(
     archivistNode = ArchivistNodeRef.new(
       switch = switch,
       networkStore = store,
+      repoStore = repoStore,
       engine = engine,
       discovery = discovery,
       prover = prover,

archivist/blockexchange/engine/advertiser.nim

@@ -83,12 +83,21 @@ proc advertiseBlock(b: Advertiser, cid: Cid) {.async: (raises: [CancelledError])
 proc advertiseLocalStoreLoop(b: Advertiser) {.async: (raises: []).} =
   try:
     while b.advertiserRunning:
-      if cidsIter =? await b.localStore.listBlocks(blockType = BlockType.Manifest):
-        trace "Advertiser begins iterating blocks..."
-        for c in cidsIter:
-          if cid =? await c:
-            await b.advertiseBlock(cid)
-        trace "Advertiser iterating blocks finished."
+      without cidsIter =? await b.localStore.listBlocks(blockType = BlockType.Manifest),
+        err:
+        trace "Error retrieving manifest iterator, advertising skipped!", err = err.msg
+        await sleepAsync(b.advertiseLocalStoreLoopSleep)
+        continue
+
+      defer:
+        if err =? (await cidsIter.dispose()).errorOption:
+          warn "Error disposing manifest iterator", err = err.msg
+
+      trace "Advertiser begins iterating blocks..."
+      for c in cidsIter:
+        if cid =? await c:
+          await b.advertiseBlock(cid)
+      trace "Advertiser iterating blocks finished."
 
       await sleepAsync(b.advertiseLocalStoreLoopSleep)
   except CancelledError:
@@ -126,7 +135,7 @@ proc start*(b: Advertiser) {.async: (raises: []).} =
 
   # The advertiser is expected to be started only once.
   if b.advertiserRunning:
-    raiseAssert "Advertiser can only be started once — this should not happen"
+    raiseAssert "Advertiser can only be started once - this should not happen"
 
   proc onBlock(cid: Cid) {.async: (raises: []).} =
     try:

archivist/blockexchange/engine/engine.nim

@@ -12,6 +12,7 @@ import std/sets
 import std/options
 import std/algorithm
 import std/sugar
+import std/tables
 
 import pkg/chronos
 import pkg/libp2p/[cid, switch, multihash, multicodec]
@@ -396,10 +397,17 @@ proc validateBlockDelivery(self: BlockExcEngine, bd: BlockDelivery): ?!void =
 
 proc blocksDeliveryHandler*(
     self: BlockExcEngine, peer: PeerId, blocksDelivery: seq[BlockDelivery]
-) {.async: (raises: []).} =
+) {.async: (raises: [CancelledError]).} =
+  # TODO: this should not be here, the engine should just resolve the future,
+  # and let the caller do the validation and storing
   trace "Received blocks from peer", peer, blocks = (blocksDelivery.mapIt(it.address))
 
-  var validatedBlocksDelivery: seq[BlockDelivery]
+  # Validate all deliveries and separate leaf vs non-leaf
+  var
+    validatedBlocksDelivery: seq[BlockDelivery]
+    leafByTree: Table[Cid, seq[(Block, Natural, ArchivistProof)]]
+    nonLeafDeliveries: seq[BlockDelivery]
+
   for bd in blocksDelivery:
     logScope:
       peer = peer
@@ -410,36 +418,54 @@ proc blocksDeliveryHandler*(
         warn "Block validation failed", msg = err.msg
         continue
 
-      if err =? (await self.localStore.putBlock(bd.blk)).errorOption:
-        error "Unable to store block", err = err.msg
-        continue
-
       if bd.address.leaf:
         without proof =? bd.proof:
           warn "Proof expected for a leaf block delivery"
           continue
-        if err =? (
-          await self.localStore.putCidAndProof(
-            bd.address.treeCid, bd.address.index, bd.blk.cid, proof
-          )
-        ).errorOption:
-          warn "Unable to store proof and cid for a block"
-          continue
+
+        leafByTree.withValue(bd.address.treeCid, slot):
+          slot[].add((bd.blk, bd.address.index, proof))
+        do:
+          leafByTree[bd.address.treeCid] = @[(bd.blk, bd.address.index, proof)]
+      else:
+        nonLeafDeliveries.add(bd)
     except CatchableError as exc:
       warn "Error handling block delivery", error = exc.msg
       continue
 
     validatedBlocksDelivery.add(bd)
 
+  # Batch write leaf blocks grouped by treeCid
+  for treeCid, items in leafByTree:
+    if err =? (await self.localStore.putBlocks(treeCid, items)).errorOption:
+      error "Unable to store leaf blocks", treeCid, err = err.msg
+      # Remove failed leaves from validatedBlocksDelivery
+      validatedBlocksDelivery.keepItIf(
+        not (it.address.leaf and it.address.treeCid == treeCid)
+      )
+
+  # Write non-leaf blocks sequentially - this should only be manifests after #94
+  for bd in nonLeafDeliveries:
+    without isManifest =? bd.blk.cid.isManifest, err:
+      error "Received a non-leaf block that isn't a manifest!", err = err.msg
+      validatedBlocksDelivery.keepItIf(it.address.cid != bd.address.cid)
+      continue
+
+    # TODO: The putBlock here should be replace by something like -
+    # storeManifestBlock(...)
+    if err =? (await self.localStore.putBlock(bd.blk)).errorOption:
+      error "Unable to store block", err = err.msg
+      validatedBlocksDelivery.keepItIf(it.address.cid != bd.address.cid)
+
   archivist_block_exchange_blocks_received.inc(validatedBlocksDelivery.len.int64)
 
   let peerCtx = self.peers.get(peer)
   if peerCtx != nil:
-    if err =? catch(await self.payForBlocks(peerCtx, blocksDelivery)).errorOption:
+    if err =? catchAsync(await self.payForBlocks(peerCtx, blocksDelivery)).errorOption:
       warn "Error paying for blocks", err = err.msg
       return
 
-  if err =? catch(await self.resolveBlocks(validatedBlocksDelivery)).errorOption:
+  if err =? catchAsync(await self.resolveBlocks(validatedBlocksDelivery)).errorOption:
     warn "Error resolving blocks", err = err.msg
     return
 
@@ -470,7 +496,11 @@ proc wantListHandler*(
         let
           have =
             try:
-              await e.address in self.localStore
+              if e.address.leaf:
+                (await self.localStore.hasBlock(e.address.treeCid, e.address.index)) |?
+                  false
+              else:
+                (await self.localStore.hasBlock(e.address.cid)) |? false
             except CatchableError:
               # TODO: should not be necessary once we have proper exception tracking on the BlockStore interface
               false
@@ -612,13 +642,13 @@ proc taskHandler*(
     proc localLookup(e: WantListEntry): Future[?!BlockDelivery] {.async.} =
       if e.address.leaf:
         (await self.localStore.getBlockAndProof(e.address.treeCid, e.address.index)).map(
-          (blkAndProof: (Block, ArchivistProof)) =>
+          (blkAndProof: (Natural, Block, ArchivistProof)) =>
             BlockDelivery(
-              address: e.address, blk: blkAndProof[0], proof: blkAndProof[1].some
+              address: e.address, blk: blkAndProof[1], proof: blkAndProof[2].some
             )
         )
       else:
-        (await self.localStore.getBlock(e.address)).map(
+        (await self.localStore.getBlock(e.address.cid)).map(
           (blk: Block) =>
             BlockDelivery(address: e.address, blk: blk, proof: ArchivistProof.none)
         )