Merge branch 'master' into feature/blockexc-prevent-retransmit

.github/workflows/docker-reusable.yml

@@ -94,11 +94,11 @@ jobs:
           - target:
               os: linux
               arch: amd64
-            builder: ubuntu-22.04
+            builder: ubuntu-24.04
           - target:
               os: linux
               arch: arm64
-            builder: ubuntu-22.04-arm
+            builder: ubuntu-24.04-arm
 
     name: Build ${{ matrix.target.os }}/${{ matrix.target.arch }}
     runs-on: ${{ matrix.builder }}

.github/workflows/docs.yml

@@ -2,17 +2,17 @@ name: OpenAPI
 
 on:
   push:
-    branches:
-      - 'master'
+    tags:
+      - "v*.*.*"
     paths:
-      - 'openapi.yaml'
-      - '.github/workflows/docs.yml'
+      - "openapi.yaml"
+      - ".github/workflows/docs.yml"
   pull_request:
     branches:
-      - '**'
+      - "**"
     paths:
-      - 'openapi.yaml'
-      - '.github/workflows/docs.yml'
+      - "openapi.yaml"
+      - ".github/workflows/docs.yml"
 
 # Sets permissions of the GITHUB_TOKEN to allow deployment to GitHub Pages
 permissions:
@@ -40,7 +40,7 @@ jobs:
   deploy:
     name: Deploy
     runs-on: ubuntu-latest
-    if: github.ref == 'refs/heads/master'
+    if: startsWith(github.ref, 'refs/tags/')
     steps:
       - name: Checkout
         uses: actions/checkout@v4

build.nims

@@ -4,7 +4,6 @@ import std/os except commandLineParams
 
 ### Helper functions
 proc buildBinary(name: string, srcDir = "./", params = "", lang = "c") =
-
   if not dirExists "build":
     mkDir "build"
 
@@ -14,13 +13,15 @@ proc buildBinary(name: string, srcDir = "./", params = "", lang = "c") =
     for param in commandLineParams():
       extra_params &= " " & param
   else:
-    for i in 2..<paramCount():
+    for i in 2 ..< paramCount():
       extra_params &= " " & paramStr(i)
 
   let
     # Place build output in 'build' folder, even if name includes a longer path.
     outName = os.lastPathPart(name)
-    cmd = "nim " & lang & " --out:build/" & outName & " " & extra_params & " " & srcDir & name & ".nim"
+    cmd =
+      "nim " & lang & " --out:build/" & outName & " " & extra_params & " " & srcDir &
+      name & ".nim"
 
   exec(cmd)
 
@@ -29,7 +30,8 @@ proc test(name: string, srcDir = "tests/", params = "", lang = "c") =
   exec "build/" & name
 
 task codex, "build codex binary":
-  buildBinary "codex", params = "-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE"
+  buildBinary "codex",
+    params = "-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE"
 
 task toolsCirdl, "build tools/cirdl binary":
   buildBinary "tools/cirdl/cirdl"
@@ -41,7 +43,9 @@ task testContracts, "Build & run Codex Contract tests":
   test "testContracts"
 
 task testIntegration, "Run integration tests":
-  buildBinary "codex", params = "-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE -d:codex_enable_proof_failures=true"
+  buildBinary "codex",
+    params =
+      "-d:chronicles_runtime_filtering -d:chronicles_log_level=TRACE -d:codex_enable_proof_failures=true"
   test "testIntegration"
   # use params to enable logging from the integration test executable
   # test "testIntegration", params = "-d:chronicles_sinks=textlines[notimestamps,stdout],textlines[dynamic] " &
@@ -90,15 +94,25 @@ task coverage, "generates code coverage report":
 
   var nimSrcs = " "
   for f in walkDirRec("codex", {pcFile}):
-    if f.endswith(".nim"): nimSrcs.add " " & f.absolutePath.quoteShell()
+    if f.endswith(".nim"):
+      nimSrcs.add " " & f.absolutePath.quoteShell()
 
   echo "======== Running Tests ======== "
-  test "coverage", srcDir = "tests/", params = " --nimcache:nimcache/coverage -d:release -d:codex_enable_proof_failures=true"
+  test "coverage",
+    srcDir = "tests/",
+    params =
+      " --nimcache:nimcache/coverage -d:release -d:codex_enable_proof_failures=true"
   exec("rm nimcache/coverage/*.c")
-  rmDir("coverage"); mkDir("coverage")
+  rmDir("coverage")
+  mkDir("coverage")
   echo " ======== Running LCOV ======== "
-  exec("lcov --capture --directory nimcache/coverage --output-file coverage/coverage.info")
-  exec("lcov --extract coverage/coverage.info --output-file coverage/coverage.f.info " & nimSrcs)
+  exec(
+    "lcov --capture --directory nimcache/coverage --output-file coverage/coverage.info"
+  )
+  exec(
+    "lcov --extract coverage/coverage.info --output-file coverage/coverage.f.info " &
+      nimSrcs
+  )
   echo " ======== Generating HTML coverage report ======== "
   exec("genhtml coverage/coverage.f.info --output-directory coverage/report ")
   echo " ======== Coverage report Done ======== "

codex.nim

@@ -38,33 +38,35 @@ when isMainModule:
   when defined(posix):
     import system/ansi_c
 
-  type
-    CodexStatus {.pure.} = enum
-      Stopped,
-      Stopping,
-      Running
+  type CodexStatus {.pure.} = enum
+    Stopped
+    Stopping
+    Running
 
   let config = CodexConf.load(
     version = codexFullVersion,
     envVarsPrefix = "codex",
-    secondarySources = proc (config: CodexConf, sources: auto) {.gcsafe, raises: [ConfigurationError].} =
-            if configFile =? config.configFile:
-              sources.addConfigFile(Toml, configFile)
+    secondarySources = proc(
+        config: CodexConf, sources: auto
+    ) {.gcsafe, raises: [ConfigurationError].} =
+      if configFile =? config.configFile:
+        sources.addConfigFile(Toml, configFile)
+    ,
   )
   config.setupLogging()
   config.setupMetrics()
 
-  if not(checkAndCreateDataDir((config.dataDir).string)):
+  if not (checkAndCreateDataDir((config.dataDir).string)):
     # We are unable to access/create data folder or data folder's
     # permissions are insecure.
     quit QuitFailure
 
-  if config.prover() and not(checkAndCreateDataDir((config.circuitDir).string)):
+  if config.prover() and not (checkAndCreateDataDir((config.circuitDir).string)):
     quit QuitFailure
 
   trace "Data dir initialized", dir = $config.dataDir
 
-  if not(checkAndCreateDataDir((config.dataDir / "repo"))):
+  if not (checkAndCreateDataDir((config.dataDir / "repo"))):
     # We are unable to access/create data folder or data folder's
     # permissions are insecure.
     quit QuitFailure
@@ -83,11 +85,12 @@ when isMainModule:
         config.dataDir / config.netPrivKeyFile
 
     privateKey = setupKey(keyPath).expect("Should setup private key!")
-    server = try:
-      CodexServer.new(config, privateKey)
-    except Exception as exc:
-      error "Failed to start Codex", msg = exc.msg
-      quit QuitFailure
+    server =
+      try:
+        CodexServer.new(config, privateKey)
+      except Exception as exc:
+        error "Failed to start Codex", msg = exc.msg
+        quit QuitFailure
 
   ## Ctrl+C handling
   proc doShutdown() =
@@ -101,7 +104,9 @@ when isMainModule:
       # workaround for https://github.com/nim-lang/Nim/issues/4057
       try:
         setupForeignThreadGc()
-      except Exception as exc: raiseAssert exc.msg # shouldn't happen
+      except Exception as exc:
+        raiseAssert exc.msg
+        # shouldn't happen
     notice "Shutting down after having received SIGINT"
 
     doShutdown()

codex/blockexchange/engine/engine.nim

@@ -402,7 +402,7 @@ proc wantListHandler*(b: BlockExcEngine, peer: PeerId, wantList: WantList) {.asy
         have = await e.address in b.localStore
         price = @(b.pricing.get(Pricing(price: 0.u256)).price.toBytesBE)
 
-      case e.wantType:
+      case e.wantType
       of WantType.WantHave:
         if have:
           presence.add(

codex/clock.nim

@@ -40,5 +40,8 @@ proc toSecondsSince1970*(bytes: seq[byte]): SecondsSince1970 =
   let asUint = uint64.fromBytes(bytes)
   cast[int64](asUint)
 
+proc toSecondsSince1970*(num: uint64): SecondsSince1970 =
+  cast[int64](num)
+
 proc toSecondsSince1970*(bigint: UInt256): SecondsSince1970 =
   bigint.truncate(int64)

codex/codex.nim

@@ -11,8 +11,10 @@ import std/sequtils
 import std/strutils
 import std/os
 import std/tables
+import std/cpuinfo
 
 import pkg/chronos
+import pkg/taskpools
 import pkg/presto
 import pkg/libp2p
 import pkg/confutils
@@ -107,7 +109,9 @@ proc bootstrapInteractions(s: CodexServer): Future[void] {.async.} =
       quit QuitFailure
 
     let marketplace = Marketplace.new(marketplaceAddress, signer)
-    let market = OnChainMarket.new(marketplace, config.rewardRecipient)
+    let market = OnChainMarket.new(
+      marketplace, config.rewardRecipient, config.marketplaceRequestCacheSize
+    )
     let clock = OnChainClock.new(provider)
 
     var client: ?ClientInteractions
@@ -194,7 +198,18 @@ proc new*(
     .withTcpTransport({ServerFlags.ReuseAddr})
     .build()
 
-  var cache: CacheStore = nil
+  var
+    cache: CacheStore = nil
+    taskpool: Taskpool
+
+  try:
+    if config.numThreads == ThreadCount(0):
+      taskpool = Taskpool.new(numThreads = min(countProcessors(), 16))
+    else:
+      taskpool = Taskpool.new(numThreads = int(config.numThreads))
+    info "Threadpool started", numThreads = taskpool.numThreads
+  except CatchableError as exc:
+    raiseAssert("Failure in taskpool initialization:" & exc.msg)
 
   if config.cacheSize > 0'nb:
     cache = CacheStore.new(cacheSize = config.cacheSize)
@@ -286,6 +301,7 @@ proc new*(
       engine = engine,
       discovery = discovery,
       prover = prover,
+      taskPool = taskpool,
     )
 
     restServer = RestServerRef

codex/conf.nim

@@ -44,14 +44,19 @@ import ./utils
 import ./nat
 import ./utils/natutils
 
+from ./contracts/config import DefaultRequestCacheSize
 from ./validationconfig import MaxSlots, ValidationGroups
 
 export units, net, codextypes, logutils, completeCmdArg, parseCmdArg, NatConfig
 export ValidationGroups, MaxSlots
 
 export
   DefaultQuotaBytes, DefaultBlockTtl, DefaultBlockMaintenanceInterval,
-  DefaultNumberOfBlocksToMaintainPerInterval
+  DefaultNumberOfBlocksToMaintainPerInterval, DefaultRequestCacheSize
+
+type ThreadCount* = distinct Natural
+
+proc `==`*(a, b: ThreadCount): bool {.borrow.}
 
 proc defaultDataDir*(): string =
   let dataDir =
@@ -71,6 +76,7 @@ const
 
   DefaultDataDir* = defaultDataDir()
   DefaultCircuitDir* = defaultDataDir() / "circuits"
+  DefaultThreadCount* = ThreadCount(0)
 
 type
   StartUpCmd* {.pure.} = enum
@@ -184,6 +190,13 @@ type
       name: "max-peers"
     .}: int
 
+    numThreads* {.
+      desc:
+        "Number of worker threads (\"0\" = use as many threads as there are CPU cores available)",
+      defaultValue: DefaultThreadCount,
+      name: "num-threads"
+    .}: ThreadCount
+
     agentString* {.
       defaultValue: "Codex",
       desc: "Node agent string which is used as identifier in network",
@@ -347,6 +360,16 @@ type
         name: "reward-recipient"
       .}: Option[EthAddress]
 
+      marketplaceRequestCacheSize* {.
+        desc:
+          "Maximum number of StorageRequests kept in memory." &
+          "Reduces fetching of StorageRequest data from the contract.",
+        defaultValue: DefaultRequestCacheSize,
+        defaultValueDesc: $DefaultRequestCacheSize,
+        name: "request-cache-size",
+        hidden
+      .}: uint16
+
       case persistenceCmd* {.defaultValue: noCmd, command.}: PersistenceCmd
       of PersistenceCmd.prover:
         circuitDir* {.
@@ -482,6 +505,13 @@ proc parseCmdArg*(
     quit QuitFailure
   ma
 
+proc parseCmdArg*(T: type ThreadCount, input: string): T {.upraises: [ValueError].} =
+  let count = parseInt(input)
+  if count != 0 and count < 2:
+    warn "Invalid number of threads", input = input
+    quit QuitFailure
+  ThreadCount(count)
+
 proc parseCmdArg*(T: type SignedPeerRecord, uri: string): T =
   var res: SignedPeerRecord
   try:
@@ -579,6 +609,15 @@ proc readValue*(
     quit QuitFailure
   val = NBytes(value)
 
+proc readValue*(
+    r: var TomlReader, val: var ThreadCount
+) {.upraises: [SerializationError, IOError].} =
+  var str = r.readValue(string)
+  try:
+    val = parseCmdArg(ThreadCount, str)
+  except CatchableError as err:
+    raise newException(SerializationError, err.msg)
+
 proc readValue*(
     r: var TomlReader, val: var Duration
 ) {.upraises: [SerializationError, IOError].} =
@@ -609,6 +648,9 @@ proc completeCmdArg*(T: type NBytes, val: string): seq[string] =
 proc completeCmdArg*(T: type Duration, val: string): seq[string] =
   discard
 
+proc completeCmdArg*(T: type ThreadCount, val: string): seq[string] =
+  discard
+
 # silly chronicles, colors is a compile-time property
 proc stripAnsi*(v: string): string =
   var