Wire up prover (#736)

* wire prover into node

* stricter case object checks

* return correct proof

* misc renames

* adding usefull traces

* fix nodes and tolerance to match expected params

* format challenges in logs

* add circom compat to solidity groth16 convertion

* update

* bump time to give nodes time to load with all circom artifacts

* misc

* misc

* use correct dataset geometry in erasure

* make errors more searchable

* use parens around `=? (await...)` calls

* styling

* styling

* use push raises

* fix to match constructor arguments

* merge master

* merge master

* integration: fix proof parameters for a test

Increased times due to ZK proof generation.
Increased storage requirement because we're now hosting
5 slots instead of 1.

* sales: calculate initial proof at start of period

reason: this ensures that the period (and therefore
the challenge) doesn't change while we're calculating
the proof

* integration: fix proof parameters for tests

Increased times due to waiting on next period.
Fixed data to be of right size.
Updated expected payout due to hosting 5 slots.

* sales: wait for stable proof challenge

When the block pointer is nearing the
wrap-around point, we wait another period
before calculating a proof.

* fix merge conflict

---------

Co-authored-by: Dmitriy Ryajov <[email protected]>
Co-authored-by: Eric <[email protected]>

Author: 3 people

codex/blockexchange/engine/engine.nim

@@ -129,7 +129,6 @@ proc sendWantHave(
   for p in peers:
     if p notin excluded:
       if address notin p.peerHave:
-        trace " wantHave > ", peer = p.id
         await b.network.request.sendWantList(
           p.id,
           @[address],
@@ -145,11 +144,18 @@ proc sendWantBlock(
     @[address],
     wantType = WantType.WantBlock) # we want this remote to send us a block
 
-proc monitorBlockHandle(b: BlockExcEngine, handle: Future[Block], address: BlockAddress, peerId: PeerId) {.async.} =
+proc monitorBlockHandle(
+  b: BlockExcEngine,
+  handle: Future[Block],
+  address: BlockAddress,
+  peerId: PeerId) {.async.} =
+
   try:
     trace "Monitoring block handle", address, peerId
     discard await handle
     trace "Block handle success", address, peerId
+  except CancelledError as exc:
+    trace "Block handle cancelled", address, peerId
   except CatchableError as exc:
     trace "Error block handle, disconnecting peer", address, exc = exc.msg, peerId
 
@@ -167,38 +173,38 @@ proc monitorBlockHandle(b: BlockExcEngine, handle: Future[Block], address: Block
 proc requestBlock*(
   b: BlockExcEngine,
   address: BlockAddress,
-  timeout = DefaultBlockTimeout): Future[Block] {.async.} =
+  timeout = DefaultBlockTimeout): Future[?!Block] {.async.} =
   let blockFuture = b.pendingBlocks.getWantHandle(address, timeout)
 
   if b.pendingBlocks.isInFlight(address):
-    return await blockFuture
-
-  let peers = b.peers.selectCheapest(address)
-  if peers.len == 0:
-    b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])
-
-  let maybePeer =
-    if peers.len > 0:
-      peers[hash(address) mod peers.len].some
-    elif b.peers.len > 0:
-      toSeq(b.peers)[hash(address) mod b.peers.len].some
-    else:
-      BlockExcPeerCtx.none
+    success await blockFuture
+  else:
+    let peers = b.peers.selectCheapest(address)
+    if peers.len == 0:
+      b.discovery.queueFindBlocksReq(@[address.cidOrTreeCid])
+
+    let maybePeer =
+      if peers.len > 0:
+        peers[hash(address) mod peers.len].some
+      elif b.peers.len > 0:
+        toSeq(b.peers)[hash(address) mod b.peers.len].some
+      else:
+        BlockExcPeerCtx.none
 
-  if peer =? maybePeer:
-    asyncSpawn b.monitorBlockHandle(blockFuture, address, peer.id)
-    b.pendingBlocks.setInFlight(address)
-    await b.sendWantBlock(address, peer)
-    codex_block_exchange_want_block_lists_sent.inc()
-    await b.sendWantHave(address, @[peer], toSeq(b.peers))
-    codex_block_exchange_want_have_lists_sent.inc()
+    if peer =? maybePeer:
+      asyncSpawn b.monitorBlockHandle(blockFuture, address, peer.id)
+      b.pendingBlocks.setInFlight(address)
+      await b.sendWantBlock(address, peer)
+      codex_block_exchange_want_block_lists_sent.inc()
+      await b.sendWantHave(address, @[peer], toSeq(b.peers))
+      codex_block_exchange_want_have_lists_sent.inc()
 
-  return await blockFuture
+    success await blockFuture
 
 proc requestBlock*(
   b: BlockExcEngine,
   cid: Cid,
-  timeout = DefaultBlockTimeout): Future[Block] =
+  timeout = DefaultBlockTimeout): Future[?!Block] =
   b.requestBlock(BlockAddress.init(cid))
 
 proc blockPresenceHandler*(
@@ -298,7 +304,10 @@ proc resolveBlocks*(b: BlockExcEngine, blocksDelivery: seq[BlockDelivery]) {.asy
   b.discovery.queueProvideBlocksReq(cids.toSeq)
 
 proc resolveBlocks*(b: BlockExcEngine, blocks: seq[Block]) {.async.} =
-  await b.resolveBlocks(blocks.mapIt(BlockDelivery(blk: it, address: BlockAddress(leaf: false, cid: it.cid))))
+  await b.resolveBlocks(
+    blocks.mapIt(
+      BlockDelivery(blk: it, address: BlockAddress(leaf: false, cid: it.cid)
+  )))
 
 proc payForBlocks(engine: BlockExcEngine,
                   peer: BlockExcPeerCtx,
@@ -315,8 +324,7 @@ proc payForBlocks(engine: BlockExcEngine,
 
 proc validateBlockDelivery(
   b: BlockExcEngine,
-  bd: BlockDelivery
-): ?!void =
+  bd: BlockDelivery): ?!void =
   if bd.address notin b.pendingBlocks:
     return failure("Received block is not currently a pending block")
 

codex/blockexchange/engine/pendingblocks.nim

@@ -44,11 +44,10 @@ proc updatePendingBlockGauge(p: PendingBlocksManager) =
   codex_block_exchange_pending_block_requests.set(p.blocks.len.int64)
 
 proc getWantHandle*(
-    p: PendingBlocksManager,
-    address: BlockAddress,
-    timeout = DefaultBlockTimeout,
-    inFlight = false
-): Future[Block] {.async.} =
+  p: PendingBlocksManager,
+  address: BlockAddress,
+  timeout = DefaultBlockTimeout,
+  inFlight = false): Future[Block] {.async.} =
   ## Add an event for a block
   ##
 
@@ -75,17 +74,15 @@ proc getWantHandle*(
     p.updatePendingBlockGauge()
 
 proc getWantHandle*(
-    p: PendingBlocksManager,
-    cid: Cid,
-    timeout = DefaultBlockTimeout,
-    inFlight = false
-): Future[Block] =
+  p: PendingBlocksManager,
+  cid: Cid,
+  timeout = DefaultBlockTimeout,
+  inFlight = false): Future[Block] =
   p.getWantHandle(BlockAddress.init(cid), timeout, inFlight)
 
 proc resolve*(
   p: PendingBlocksManager,
-  blocksDelivery: seq[BlockDelivery]
-  ) {.gcsafe, raises: [].} =
+  blocksDelivery: seq[BlockDelivery]) {.gcsafe, raises: [].} =
   ## Resolve pending blocks
   ##
 
@@ -108,16 +105,23 @@ proc resolve*(
     do:
       warn "Attempting to resolve block that's not currently a pending block", address = bd.address
 
-proc setInFlight*(p: PendingBlocksManager,
-                  address: BlockAddress,
-                  inFlight = true) =
+proc setInFlight*(
+  p: PendingBlocksManager,
+  address: BlockAddress,
+  inFlight = true) =
+  ## Set inflight status for a block
+  ##
+
   p.blocks.withValue(address, pending):
     pending[].inFlight = inFlight
     trace "Setting inflight", address, inFlight = pending[].inFlight
 
-proc isInFlight*(p: PendingBlocksManager,
-                 address: BlockAddress,
-                ): bool =
+proc isInFlight*(
+  p: PendingBlocksManager,
+  address: BlockAddress): bool =
+  ## Check if a block is in flight
+  ##
+
   p.blocks.withValue(address, pending):
     result = pending[].inFlight
     trace "Getting inflight", address, inFlight = result
@@ -145,7 +149,6 @@ iterator wantListCids*(p: PendingBlocksManager): Cid =
       yieldedCids.incl(cid)
       yield cid
 
-
 iterator wantHandles*(p: PendingBlocksManager): Future[Block] =
   for v in p.blocks.values:
     yield v.handle

codex/erasure/backends/leopard.nim

@@ -22,10 +22,9 @@ type
     decoder*: Option[LeoDecoder]
 
 method encode*(
-    self: LeoEncoderBackend,
-    data,
-    parity: var openArray[seq[byte]]
-): Result[void, cstring] =
+  self: LeoEncoderBackend,
+  data,
+  parity: var openArray[seq[byte]]): Result[void, cstring] =
   ## Encode data using Leopard backend
 
   if parity.len == 0:
@@ -43,11 +42,10 @@ method encode*(
   encoder.encode(data, parity)
 
 method decode*(
-    self: LeoDecoderBackend,
-    data,
-    parity,
-    recovered: var openArray[seq[byte]]
-): Result[void, cstring] =
+  self: LeoDecoderBackend,
+  data,
+  parity,
+  recovered: var openArray[seq[byte]]): Result[void, cstring] =
   ## Decode data using given Leopard backend
 
   var decoder =
@@ -71,26 +69,24 @@ method release*(self: LeoDecoderBackend) =
     self.decoder.get().free()
 
 proc new*(
-    T: type LeoEncoderBackend,
-    blockSize,
-    buffers,
-    parity: int
-): LeoEncoderBackend =
+  T: type LeoEncoderBackend,
+  blockSize,
+  buffers,
+  parity: int): LeoEncoderBackend =
   ## Create an instance of an Leopard Encoder backend
-  ## 
+  ##
   LeoEncoderBackend(
     blockSize: blockSize,
     buffers: buffers,
     parity: parity)
 
 proc new*(
-    T: type LeoDecoderBackend,
-    blockSize,
-    buffers,
-    parity: int
-): LeoDecoderBackend =
+  T: type LeoDecoderBackend,
+  blockSize,
+  buffers,
+  parity: int): LeoDecoderBackend =
   ## Create an instance of an Leopard Decoder backend
-  ## 
+  ##
   LeoDecoderBackend(
     blockSize: blockSize,
     buffers: buffers,

codex/erasure/erasure.nim

@@ -241,7 +241,7 @@ proc init*(
 
   let
     rounded = roundUp(manifest.blocksCount, ecK)
-    steps = divUp(manifest.blocksCount, ecK)
+    steps = divUp(rounded, ecK)
     blocksCount = rounded + (steps * ecM)
 
   success EncodingParams(

codex/logutils.nim

@@ -239,3 +239,5 @@ formatIt(LogFormat.textLines, Cid): shortLog($it)
 formatIt(LogFormat.json, Cid): $it
 formatIt(UInt256): $it
 formatIt(MultiAddress): $it
+formatIt(LogFormat.textLines, array[32, byte]): it.short0xHexLog
+formatIt(LogFormat.json, array[32, byte]): it.to0xHex

codex/node.nim

@@ -556,6 +556,8 @@ proc onStore(
     trace "Unable to build slot", err = err.msg
     return failure(err)
 
+  trace "Slot successfully retrieved and reconstructed"
+
   if cid =? slotRoot.toSlotCid() and cid != manifest.slotRoots[slotIdx.int]:
     trace "Slot root mismatch", manifest = manifest.slotRoots[slotIdx.int], recovered = slotRoot.toSlotCid()
     return failure(newException(CodexError, "Slot root mismatch"))
@@ -593,25 +595,32 @@ proc onProve(
       error "Unable to fetch manifest for cid", err = err.msg
       return failure(err)
 
-    without builder =? Poseidon2Builder.new(self.networkStore.localStore, manifest), err:
-      error "Unable to create slots builder", err = err.msg
-      return failure(err)
+    when defined(verify_circuit):
+      without (inputs, proof) =? await prover.prove(slotIdx, manifest, challenge), err:
+        error "Unable to generate proof", err = err.msg
+        return failure(err)
 
-    without sampler =? DataSampler.new(slotIdx, self.networkStore.localStore, builder), err:
-      error "Unable to create data sampler", err = err.msg
-      return failure(err)
+      without checked =? await prover.verify(proof, inputs), err:
+        error "Unable to verify proof", err = err.msg
+        return failure(err)
 
-    without proofInput =? await sampler.getProofInput(challenge, nSamples = 3), err:
-      error "Unable to get proof input for slot", err = err.msg
-      return failure(err)
+      if not checked:
+        error "Proof verification failed"
+        return failure("Proof verification failed")
 
-  # Todo: send proofInput to circuit. Get proof. (Profit, repeat.)
+      trace "Proof verified successfully"
+    else:
+      without (_, proof) =? await prover.prove(slotIdx, manifest, challenge), err:
+        error "Unable to generate proof", err = err.msg
+        return failure(err)
 
-  # For now: dummy proof that is not all zero's, so that it is accepted by the
-  # dummy verifier:
-  var proof = Groth16Proof.default
-  proof.a.x = 42.u256
-  success(proof)
+    let groth16Proof = proof.toGroth16Proof()
+    trace "Proof generated successfully", groth16Proof
+
+    success groth16Proof
+  else:
+    warn "Prover not enabled"
+    failure "Prover not enabled"
 
 proc onExpiryUpdate(
   self: CodexNodeRef,

codex/sales/reservations.nim

@@ -325,8 +325,9 @@ proc createReservation*(
     return failure(error)
 
   if availability.size < slotSize:
-    let error = newException(BytesOutOfBoundsError, "trying to reserve an " &
-      "amount of bytes that is greater than the total size of the Availability")
+    let error = newException(
+      BytesOutOfBoundsError,
+      "trying to reserve an amount of bytes that is greater than the total size of the Availability")
     return failure(error)
 
   if createResErr =? (await self.update(reservation)).errorOption:
@@ -427,9 +428,9 @@ proc release*(
     return failure(error)
 
   if reservation.size < bytes.u256:
-    let error = newException(BytesOutOfBoundsError,
-      "trying to release an amount of bytes that is greater than the total " &
-      "size of the Reservation")
+    let error = newException(
+      BytesOutOfBoundsError,
+      "trying to release an amount of bytes that is greater than the total size of the Reservation")
     return failure(error)
 
   if releaseErr =? (await self.repo.release(bytes)).errorOption:
@@ -540,7 +541,7 @@ proc findAvailability*(
 
         return some availability
 
-      trace "availiability did not match",
+      trace "availability did not match",
         size, availsize = availability.size,
         duration, availDuration = availability.duration,
         minPrice, availMinPrice = availability.minPrice,