refactor: merging proving module into sales (#469)

Co-authored-by: Eric <5089238+emizzle@users.noreply.github.com>

codex/codex.nim

@@ -55,9 +55,9 @@ proc bootstrapInteractions(
     config: CodexConf,
     repo: RepoStore
 ): Future[Contracts] {.async.} =
-  ## bootstrap interactions and return contracts 
+  ## bootstrap interactions and return contracts
   ## using clients, hosts, validators pairings
-  ## 
+  ##
 
   if not config.persistence and not config.validator:
     if config.ethAccount.isSome:
@@ -86,18 +86,23 @@ proc bootstrapInteractions(
   var client: ?ClientInteractions
   var host: ?HostInteractions
   var validator: ?ValidatorInteractions
+
   if config.persistence:
-    let purchasing = Purchasing.new(market, clock)
+    # This is used for simulation purposes. Normal nodes won't be compiled with this flag
+    # and hence the proof failure will always be 0.
     when codex_enable_proof_failures:
-      let proving = if config.simulateProofFailures > 0:
+      let proofFailures = config.simulateProofFailures
-                      SimulatedProving.new(market, clock,
+      if proofFailures > 0:
-                                           config.simulateProofFailures)
+        warn "Enabling proof failure simulation!"
-                    else: Proving.new(market, clock)
     else:
-      let proving = Proving.new(market, clock)
+      let proofFailures = 0
-    let sales = Sales.new(market, clock, proving, repo)
+      if config.simulateProofFailures > 0:
+        warn "Proof failure simulation is not enabled for this build! Configuration ignored"
+
+    let purchasing = Purchasing.new(market, clock)
+    let sales = Sales.new(market, clock, repo, proofFailures)
     client = some ClientInteractions.new(clock, purchasing)
-    host = some HostInteractions.new(clock, sales, proving)
+    host = some HostInteractions.new(clock, sales)
   if config.validator:
     let validation = Validation.new(clock, market, config.validatorMaxSlots)
     validator = some ValidatorInteractions.new(clock, validation)

codex/conf.nim

@@ -251,7 +251,7 @@ type
           defaultValue: 0
           name: "simulate-proof-failures"
           hidden
-        .}: uint
+        .}: int
 
     of initNode:
       discard

codex/contracts/interactions/hostinteractions.nim

@@ -2,34 +2,28 @@ import pkg/ethers
 import pkg/chronicles
 
 import ../../sales
-import ../../proving
 import ./interactions
 
 export sales
-export proving
 export chronicles
 
 type
   HostInteractions* = ref object of ContractInteractions
     sales*: Sales
-    proving*: Proving
 
 proc new*(
     _: type HostInteractions,
     clock: OnChainClock,
-    sales: Sales,
+    sales: Sales
-    proving: Proving
 ): HostInteractions =
   ## Create a new HostInteractions instance
-  ## 
+  ##
-  HostInteractions(clock: clock, sales: sales, proving: proving)
+  HostInteractions(clock: clock, sales: sales)
 
 method start*(self: HostInteractions) {.async.} =
   await procCall ContractInteractions(self).start()
   await self.sales.start()
-  await self.proving.start()
 
 method stop*(self: HostInteractions) {.async.} =
   await self.sales.stop()
-  await self.proving.stop()
   await procCall ContractInteractions(self).start()

codex/node.nim

@@ -372,7 +372,7 @@ proc start*(node: CodexNodeRef) {.async.} =
       # TODO: remove data from local storage
       discard
 
-    hostContracts.proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    hostContracts.sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       # TODO: generate proof
       return @[42'u8]
 

codex/proving.nim

@@ -1,5 +0,0 @@
-import ./proving/proving
-import ./proving/simulated
-
-export proving
-export simulated

codex/proving/proving.nim

@@ -1,101 +0,0 @@
-import std/sets
-import pkg/upraises
-import pkg/questionable
-import pkg/chronicles
-import ../market
-import ../clock
-
-export sets
-
-logScope:
-  topics = "marketplace proving"
-
-type
-  Proving* = ref object of RootObj
-    market*: Market
-    clock: Clock
-    loop: ?Future[void]
-    slots*: HashSet[Slot]
-    onProve: ?OnProve
-  OnProve* = proc(slot: Slot): Future[seq[byte]] {.gcsafe, upraises: [].}
-
-func new*(T: type Proving, market: Market, clock: Clock): T =
-  T(market: market, clock: clock)
-
-proc onProve*(proving: Proving): ?OnProve =
-  proving.onProve
-
-proc `onProve=`*(proving: Proving, callback: OnProve) =
-  proving.onProve = some callback
-
-func add*(proving: Proving, slot: Slot) =
-  proving.slots.incl(slot)
-
-proc getCurrentPeriod*(proving: Proving): Future[Period] {.async.} =
-  let periodicity = await proving.market.periodicity()
-  return periodicity.periodOf(proving.clock.now().u256)
-
-proc waitUntilPeriod(proving: Proving, period: Period) {.async.} =
-  let periodicity = await proving.market.periodicity()
-  await proving.clock.waitUntil(periodicity.periodStart(period).truncate(int64))
-
-proc removeEndedContracts(proving: Proving) {.async.} =
-  var ended: HashSet[Slot]
-  for slot in proving.slots:
-    let state = await proving.market.slotState(slot.id)
-    if state == SlotState.Finished:
-      debug "Collecting finished slot's reward", slot = $slot.id
-      await proving.market.freeSlot(slot.id)
-
-    if state != SlotState.Filled:
-      debug "Request ended, cleaning up slot", slot = $slot.id
-      ended.incl(slot)
-  proving.slots.excl(ended)
-
-method prove*(proving: Proving, slot: Slot) {.base, async.} =
-  logScope:
-    currentPeriod = await proving.getCurrentPeriod()
-
-  without onProve =? proving.onProve:
-    raiseAssert "onProve callback not set"
-  try:
-    debug "Proving slot"
-    let proof = await onProve(slot)
-    trace "submitting proof", currentPeriod = await proving.getCurrentPeriod()
-    await proving.market.submitProof(slot.id, proof)
-  except CatchableError as e:
-    error "Submitting proof failed", msg = e.msg
-
-proc run(proving: Proving) {.async.} =
-  try:
-    while true:
-      let currentPeriod = await proving.getCurrentPeriod()
-      debug "Proving for new period", period = currentPeriod
-      await proving.removeEndedContracts()
-      for slot in proving.slots:
-        let id = slot.id
-        if (await proving.market.isProofRequired(id)) or
-           (await proving.market.willProofBeRequired(id)):
-          asyncSpawn proving.prove(slot)
-      await proving.waitUntilPeriod(currentPeriod + 1)
-  except CancelledError:
-    discard
-  except CatchableError as e:
-    error "Proving failed", msg = e.msg
-
-proc start*(proving: Proving) {.async.} =
-  if proving.loop.isSome:
-    return
-
-  proving.loop = some proving.run()
-
-proc stop*(proving: Proving) {.async.} =
-  if loop =? proving.loop:
-    proving.loop = Future[void].none
-    if not loop.finished:
-      await loop.cancelAndWait()
-
-proc subscribeProofSubmission*(proving: Proving,
-                               callback: OnProofSubmitted):
-                              Future[Subscription] =
-  proving.market.subscribeProofSubmission(callback)

codex/proving/simulated.nim

@@ -1,47 +0,0 @@
-import ../conf
-when codex_enable_proof_failures:
-
-  import std/strutils
-  import pkg/chronicles
-  import pkg/ethers
-  import pkg/ethers/testing
-  import ../market
-  import ../clock
-  import ./proving
-
-  type
-    SimulatedProving* = ref object of Proving
-      failEveryNProofs: uint
-      proofCount: uint
-
-  logScope:
-    topics = "simulated proving"
-
-  func new*(_: type SimulatedProving,
-            market: Market,
-            clock: Clock,
-            failEveryNProofs: uint): SimulatedProving =
-
-    let p = SimulatedProving.new(market, clock)
-    p.failEveryNProofs = failEveryNProofs
-    return p
-
-  proc onSubmitProofError(error: ref CatchableError, period: UInt256) =
-    error "Submitting invalid proof failed", period, msg = error.msg
-
-  method prove(proving: SimulatedProving, slot: Slot) {.async.} =
-    let period = await proving.getCurrentPeriod()
-    proving.proofCount += 1
-    if proving.failEveryNProofs > 0'u and
-      proving.proofCount mod proving.failEveryNProofs == 0'u:
-      proving.proofCount = 0
-      try:
-        trace "submitting INVALID proof", currentPeriod = await proving.getCurrentPeriod()
-        await proving.market.submitProof(slot.id, newSeq[byte](0))
-      except ProviderError as e:
-        if not e.revertReason.contains("Invalid proof"):
-          onSubmitProofError(e, period)
-      except CatchableError as e:
-        onSubmitProofError(e, period)
-    else:
-      await procCall Proving(proving).prove(slot)

codex/sales.nim

@@ -7,7 +7,6 @@ import pkg/chronicles
 import pkg/datastore
 import ./market
 import ./clock
-import ./proving
 import ./stores
 import ./contracts/requests
 import ./contracts/marketplace
@@ -61,26 +60,37 @@ proc `onClear=`*(sales: Sales, onClear: OnClear) =
 proc `onSale=`*(sales: Sales, callback: OnSale) =
   sales.context.onSale = some callback
 
+proc `onProve=`*(sales: Sales, callback: OnProve) =
+  sales.context.onProve = some callback
+
 proc onStore*(sales: Sales): ?OnStore = sales.context.onStore
 
 proc onClear*(sales: Sales): ?OnClear = sales.context.onClear
 
 proc onSale*(sales: Sales): ?OnSale = sales.context.onSale
 
+proc onProve*(sales: Sales): ?OnProve = sales.context.onProve
+
 func new*(_: type Sales,
           market: Market,
           clock: Clock,
-          proving: Proving,
           repo: RepoStore): Sales =
+  Sales.new(market, clock, repo, 0)
+
+func new*(_: type Sales,
+          market: Market,
+          clock: Clock,
+          repo: RepoStore,
+          simulateProofFailures: int): Sales =
 
   let reservations = Reservations.new(repo)
   Sales(
     context: SalesContext(
       market: market,
       clock: clock,
-      proving: proving,
       reservations: reservations,
-      slotQueue: SlotQueue.new(reservations)
+      slotQueue: SlotQueue.new(reservations),
+      simulateProofFailures: simulateProofFailures
     ),
     trackedFutures: TrackedFutures.new(),
     subscriptions: @[]
@@ -392,9 +402,9 @@ proc unsubscribe(sales: Sales) {.async.} =
       error "Unable to unsubscribe from subscription", error = e.msg
 
 proc start*(sales: Sales) {.async.} =
+  await sales.load()
   await sales.startSlotQueue()
   await sales.subscribe()
-  await sales.load()
 
 proc stop*(sales: Sales) {.async.} =
   trace "stopping sales"

codex/sales/salescontext.nim

@@ -4,7 +4,6 @@ import pkg/upraises
 import ../node/batch
 import ../market
 import ../clock
-import ../proving
 import ./slotqueue
 import ./reservations
 
@@ -16,15 +15,15 @@ type
     onClear*: ?OnClear
     onSale*: ?OnSale
     onCleanUp*: OnCleanUp
-    proving*: Proving
+    onProve*: ?OnProve
     reservations*: Reservations
     slotQueue*: SlotQueue
+    simulateProofFailures*: int
 
   OnStore* = proc(request: StorageRequest,
                   slot: UInt256,
                   onBatch: BatchProc): Future[?!void] {.gcsafe, upraises: [].}
-  OnProve* = proc(request: StorageRequest,
+  OnProve* = proc(slot: Slot): Future[seq[byte]] {.gcsafe, upraises: [].}
-                  slot: UInt256): Future[seq[byte]] {.gcsafe, upraises: [].}
   OnClear* = proc(request: StorageRequest,
                   slotIndex: UInt256) {.gcsafe, upraises: [].}
   OnSale* = proc(request: StorageRequest,

codex/sales/statemachine.nim

@@ -4,13 +4,11 @@ import ../errors
 import ../utils/asyncstatemachine
 import ../market
 import ../clock
-import ../proving
 import ../contracts/requests
 
 export market
 export clock
 export asyncstatemachine
-export proving
 
 type
   SaleState* = ref object of State

codex/sales/states/cancelled.nim

@@ -1,7 +1,11 @@
+import pkg/chronicles
 import ../statemachine
 import ./errorhandling
 import ./errored
 
+logScope:
+    topics = "marketplace sales cancelled"
+
 type
   SaleCancelled* = ref object of ErrorHandlingState
   SaleCancelledError* = object of CatchableError

codex/sales/states/downloading.nim

@@ -9,7 +9,7 @@ import ./errorhandling
 import ./cancelled
 import ./failed
 import ./filled
-import ./proving
+import ./initialproving
 import ./errored
 
 type
@@ -80,4 +80,4 @@ method run*(state: SaleDownloading, machine: Machine): Future[?State] {.async.}
   trace "Download complete"
 
   markUnused(availability.id)
-  return some State(SaleProving())
+  return some State(SaleInitialProving())

codex/sales/states/failed.nim

@@ -1,7 +1,11 @@
+import pkg/chronicles
 import ../statemachine
 import ./errorhandling
 import ./errored
 
+logScope:
+    topics = "marketplace sales failed"
+
 type
   SaleFailed* = ref object of ErrorHandlingState
   SaleFailedError* = object of SaleError

codex/sales/states/filled.nim

@@ -1,11 +1,17 @@
 import pkg/questionable
+import pkg/chronicles
+import ../../conf
 import ../statemachine
 import ../salesagent
 import ./errorhandling
 import ./errored
-import ./finished
 import ./cancelled
 import ./failed
+import ./proving
+import ./provingsimulated
+
+logScope:
+    topics = "marketplace sales filled"
 
 type
   SaleFilled* = ref object of ErrorHandlingState
@@ -21,7 +27,8 @@ method `$`*(state: SaleFilled): string = "SaleFilled"
 
 method run*(state: SaleFilled, machine: Machine): Future[?State] {.async.} =
   let data = SalesAgent(machine).data
-  let market = SalesAgent(machine).context.market
+  let context = SalesAgent(machine).context
+  let market = context.market
 
   without slotIndex =? data.slotIndex:
     raiseAssert("no slot index assigned")
@@ -29,7 +36,19 @@ method run*(state: SaleFilled, machine: Machine): Future[?State] {.async.} =
   let host = await market.getHost(data.requestId, slotIndex)
   let me = await market.getSigner()
   if host == me.some:
-    return some State(SaleFinished())
+    info "Slot succesfully filled", requestId = $data.requestId, slotIndex
+
+    if request =? data.request and slotIndex =? data.slotIndex and
+        onSale =? context.onSale:
+      onSale(request, slotIndex)
+
+    when codex_enable_proof_failures:
+      if context.simulateProofFailures > 0:
+        info "Proving with failure rate", rate = context.simulateProofFailures
+        return some State(SaleProvingSimulated(failEveryNProofs: context.simulateProofFailures))
+
+    return some State(SaleProving())
+
   else:
     let error = newException(HostMismatchError, "Slot filled by other host")
     return some State(SaleErrored(error: error))

codex/sales/states/finished.nim

@@ -25,16 +25,13 @@ method run*(state: SaleFinished, machine: Machine): Future[?State] {.async.} =
   let data = agent.data
   let context = agent.context
 
-  debug "Request succesfully filled", requestId = $data.requestId
+  without request =? data.request:
+    raiseAssert "no sale request"
 
-  if request =? data.request and
+  without slotIndex =? data.slotIndex:
-      slotIndex =? data.slotIndex:
+    raiseAssert("no slot index assigned")
-    let slot = Slot(request: request, slotIndex: slotIndex)
-    debug "Adding slot to proving list", slotId = $slot.id
-    context.proving.add(slot)
 
-    if onSale =? context.onSale:
+  info "Slot finished and paid out", requestId = $data.requestId, slotIndex
-      onSale(request, slotIndex)
 
   if onCleanUp =? context.onCleanUp:
     await onCleanUp()

codex/sales/states/ignored.nim

@@ -1,8 +1,12 @@
+import pkg/chronicles
 import pkg/chronos
 import ../statemachine
 import ../salesagent
 import ./errorhandling
 
+logScope:
+    topics = "marketplace sales ignored"
+
 type
   SaleIgnored* = ref object of ErrorHandlingState
 

codex/sales/states/initialproving.nim

@@ -0,0 +1,40 @@
+import pkg/chronicles
+import ../statemachine
+import ../salesagent
+import ./errorhandling
+import ./filling
+import ./cancelled
+import ./failed
+
+logScope:
+    topics = "marketplace sales initial-proving"
+
+type
+  SaleInitialProving* = ref object of ErrorHandlingState
+
+method `$`*(state: SaleInitialProving): string = "SaleInitialProving"
+
+method onCancelled*(state: SaleInitialProving, request: StorageRequest): ?State =
+  return some State(SaleCancelled())
+
+method onFailed*(state: SaleInitialProving, request: StorageRequest): ?State =
+  return some State(SaleFailed())
+
+method run*(state: SaleInitialProving, machine: Machine): Future[?State] {.async.} =
+  let data = SalesAgent(machine).data
+  let context = SalesAgent(machine).context
+
+  without request =? data.request:
+    raiseAssert "no sale request"
+
+  without onProve =? context.onProve:
+    raiseAssert "onProve callback not set"
+
+  without slotIndex =? data.slotIndex:
+    raiseAssert("no slot index assigned")
+
+  debug "Generating initial proof", requestId = $data.requestId
+  let proof = await onProve(Slot(request: request, slotIndex: slotIndex))
+  debug "Finished proof calculation", requestId = $data.requestId
+
+  return some State(SaleFilling(proof: proof))

codex/sales/states/payout.nim

@@ -0,0 +1,38 @@
+import pkg/chronicles
+import ../../market
+import ../statemachine
+import ../salesagent
+import ./errorhandling
+import ./cancelled
+import ./failed
+import ./finished
+
+logScope:
+    topics = "marketplace sales payout"
+
+type
+  SalePayout* = ref object of ErrorHandlingState
+
+method `$`*(state: SalePayout): string = "SalePayout"
+
+method onCancelled*(state: SalePayout, request: StorageRequest): ?State =
+  return some State(SaleCancelled())
+
+method onFailed*(state: SalePayout, request: StorageRequest): ?State =
+  return some State(SaleFailed())
+
+method run(state: SalePayout, machine: Machine): Future[?State] {.async.} =
+  let data = SalesAgent(machine).data
+  let market = SalesAgent(machine).context.market
+
+  without request =? data.request:
+    raiseAssert "no sale request"
+
+  without slotIndex =? data.slotIndex:
+    raiseAssert("no slot index assigned")
+
+  let slot = Slot(request: request, slotIndex: slotIndex)
+  debug "Collecting finished slot's reward",  requestId = $data.requestId, slotIndex
+  await market.freeSlot(slot.id)
+
+  return some State(SaleFinished())

codex/sales/states/preparing.nim

@@ -15,7 +15,7 @@ type
   SalePreparing* = ref object of ErrorHandlingState
 
 logScope:
-    topics = "sales preparing"
+    topics = "marketplace sales preparing"
 
 method `$`*(state: SalePreparing): string = "SalePreparing"
 

codex/sales/states/proving.nim

@@ -1,29 +1,106 @@
+import std/options
 import pkg/chronicles
+import ../../clock
 import ../statemachine
 import ../salesagent
+import ../salescontext
 import ./errorhandling
-import ./filling
 import ./cancelled
 import ./failed
-import ./filled
+import ./errored
+import ./payout
 
 logScope:
     topics = "marketplace sales proving"
 
 type
+  SlotNotFilledError* = object of CatchableError
   SaleProving* = ref object of ErrorHandlingState
+    loop: Future[void]
+
+method prove*(
+  state: SaleProving,
+  slot: Slot,
+  onProve: OnProve,
+  market: Market,
+  currentPeriod: Period
+) {.base, async.} =
+  try:
+    let proof = await onProve(slot)
+    debug "Submitting proof", currentPeriod = currentPeriod, slotId = $slot.id
+    await market.submitProof(slot.id, proof)
+  except CatchableError as e:
+    error "Submitting proof failed", msg = e.msg
+
+proc proveLoop(
+  state: SaleProving,
+  market: Market,
+  clock: Clock,
+  request: StorageRequest,
+  slotIndex: UInt256,
+  onProve: OnProve
+) {.async.} =
+
+  let slot = Slot(request: request, slotIndex: slotIndex)
+  let slotId = slot.id
+
+  logScope:
+    period = currentPeriod
+    requestId = $request.id
+    slotIndex
+    slotId = $slot.id
+
+  proc getCurrentPeriod(): Future[Period] {.async.} =
+    let periodicity = await market.periodicity()
+    return periodicity.periodOf(clock.now().u256)
+
+  proc waitUntilPeriod(period: Period) {.async.} =
+    let periodicity = await market.periodicity()
+    await clock.waitUntil(periodicity.periodStart(period).truncate(int64))
+
+  while true:
+    let currentPeriod = await getCurrentPeriod()
+    let slotState = await market.slotState(slot.id)
+    if slotState == SlotState.Finished:
+      debug "Slot reached finished state", period = currentPeriod
+      return
+
+    if slotState != SlotState.Filled:
+      raise newException(SlotNotFilledError, "Slot is not in Filled state!")
+
+    debug "Proving for new period", period = currentPeriod
+
+    if (await market.isProofRequired(slotId)) or (await market.willProofBeRequired(slotId)):
+      debug "Proof is required", period = currentPeriod
+      await state.prove(slot, onProve, market, currentPeriod)
+
+    await waitUntilPeriod(currentPeriod + 1)
 
 method `$`*(state: SaleProving): string = "SaleProving"
 
 method onCancelled*(state: SaleProving, request: StorageRequest): ?State =
+  if not state.loop.isNil:
+    if not state.loop.finished:
+      try:
+        waitFor state.loop.cancelAndWait()
+      except CatchableError as e:
+        error "Error during cancelation of prooving loop", msg = e.msg
+
+    state.loop = nil
+
   return some State(SaleCancelled())
 
 method onFailed*(state: SaleProving, request: StorageRequest): ?State =
-  return some State(SaleFailed())
+  if not state.loop.isNil:
+    if not state.loop.finished:
+      try:
+        waitFor state.loop.cancelAndWait()
+      except CatchableError as e:
+        error "Error during cancelation of prooving loop", msg = e.msg
 
-method onSlotFilled*(state: SaleProving, requestId: RequestId,
+    state.loop = nil
-                     slotIndex: UInt256): ?State =
+
-  return some State(SaleFilled())
+  return some State(SaleFailed())
 
 method run*(state: SaleProving, machine: Machine): Future[?State] {.async.} =
   let data = SalesAgent(machine).data
@@ -32,14 +109,39 @@ method run*(state: SaleProving, machine: Machine): Future[?State] {.async.} =
   without request =? data.request:
     raiseAssert "no sale request"
 
-  without onProve =? context.proving.onProve:
+  without onProve =? context.onProve:
     raiseAssert "onProve callback not set"
 
   without slotIndex =? data.slotIndex:
     raiseAssert("no slot index assigned")
 
-  debug "Start proof generation", requestId = $data.requestId
+  without market =? context.market:
-  let proof = await onProve(Slot(request: request, slotIndex: slotIndex))
+    raiseAssert("market not set")
-  debug "Finished proof generation", requestId = $data.requestId
 
-  return some State(SaleFilling(proof: proof))
+  without clock =? context.clock:
+    raiseAssert("clock not set")
+
+  debug "Start proving", requestId = $data.requestId, slotIndex
+  try:
+    let loop = state.proveLoop(market, clock, request, slotIndex, onProve)
+    state.loop = loop
+    await loop
+  except CancelledError:
+    discard
+  except CatchableError as e:
+    error "Proving failed", msg = e.msg
+    return some State(SaleErrored(error: e))
+  finally:
+    # Cleanup of the proving loop
+    debug "Stopping proving.", requestId = $data.requestId, slotIndex
+
+    if not state.loop.isNil:
+        if not state.loop.finished:
+          try:
+            await state.loop.cancelAndWait()
+          except CatchableError as e:
+            error "Error during cancelation of prooving loop", msg = e.msg
+
+        state.loop = nil
+
+  return some State(SalePayout())

codex/sales/states/provingsimulated.nim

@@ -0,0 +1,41 @@
+import ../../conf
+when codex_enable_proof_failures:
+  import std/strutils
+  import pkg/chronicles
+  import pkg/stint
+  import pkg/ethers
+  import pkg/ethers/testing
+
+  import ../../contracts/requests
+  import ../../market
+  import ../salescontext
+  import ./proving
+
+  logScope:
+      topics = "marketplace sales simulated-proving"
+
+  type
+    SaleProvingSimulated* = ref object of SaleProving
+      failEveryNProofs*: int
+      proofCount: int
+
+  proc onSubmitProofError(error: ref CatchableError, period: UInt256, slotId: SlotId) =
+    error "Submitting invalid proof failed", period = period, slotId = $slotId, msg = error.msg
+
+  method prove*(state: SaleProvingSimulated, slot: Slot, onProve: OnProve, market: Market, currentPeriod: Period) {.async.} =
+    trace "Processing proving in simulated mode"
+    state.proofCount += 1
+    if state.failEveryNProofs > 0 and
+      state.proofCount mod state.failEveryNProofs == 0:
+      state.proofCount = 0
+
+      try:
+        warn "Submitting INVALID proof", period = currentPeriod, slotId = slot.id
+        await market.submitProof(slot.id, newSeq[byte](0))
+      except ProviderError as e:
+        if not e.revertReason.contains("Invalid proof"):
+          onSubmitProofError(e, currentPeriod, slot.id)
+      except CatchableError as e:
+        onSubmitProofError(e, currentPeriod, slot.id)
+    else:
+      await procCall SaleProving(state).prove(slot, onProve, market, currentPeriod)

codex/sales/states/unknown.nim

@@ -1,3 +1,4 @@
+import pkg/chronicles
 import ../statemachine
 import ../salesagent
 import ./filled
@@ -6,6 +7,9 @@ import ./failed
 import ./errored
 import ./cancelled
 
+logScope:
+    topics = "marketplace sales unknown"
+
 type
   SaleUnknown* = ref object of SaleState
   SaleUnknownError* = object of CatchableError

docs/TWOCLIENTTEST.md

@@ -137,9 +137,9 @@ If your file is downloaded and identical to the file you uploaded, then this man
 curl --location 'http://localhost:8081/api/codex/v1/sales/availability' \
 --header 'Content-Type: application/json' \
 --data '{
-    "size": "0xF4240",
+    "size": "1000000",
-    "duration": "0xE10",
+    "duration": "3600",
-    "minPrice": "0x3E8"
+    "minPrice": "1000"
 }'
 ```
 
@@ -151,18 +151,27 @@ This informs your node that you are available to store 1MB of data for a duratio
 curl --location 'http://localhost:8080/api/codex/v1/storage/request/<CID>' \
 --header 'Content-Type: application/json' \
 --data '{
-    "reward": "0x400",
+    "reward": "1024",
-    "duration": "0x78",
+    "duration": "120",
-    "proofProbability": "0x10"
+    "proofProbability": "8"
 }'
 ```
 
 This creates a storage Request for `<CID>` (that you have to fill in) for
 duration of 2 minutes and with reward of 1024 tokens. It expects hosts to
-provide a storage proof once every 16 periods on average.
+provide a storage proof once every 8 periods on average.
 
 It returns Request ID which you can then use to query for the Request's state as follows:
 
 ```bash
 curl --location 'http://localhost:8080/api/codex/v1/storage/purchases/<RequestID>'
 ```
+
+## Notes
+
+When using the Ganache blockchain, there are some deviations from the expected behavior, mainly linked to how blocks are mined, which affects certain functionalities in the Sales module.
+Therefore, if you are manually testing processes such as payout collection after a request is finished or proof submissions, you need to mine some blocks manually for it to work correctly. You can do this by using the following curl command:
+
+```bash
+$ curl -H "Content-Type: application/json" -X POST --data '{"jsonrpc":"2.0","method":"evm_mine","params":[],"id":67}' 127.0.0.1:8545
+```

tests/codex/sales/states/testfilled.nim

@@ -5,6 +5,7 @@ import pkg/codex/sales/salesagent
 import pkg/codex/sales/salescontext
 import pkg/codex/sales/states/filled
 import pkg/codex/sales/states/errored
+import pkg/codex/sales/states/proving
 import pkg/codex/sales/states/finished
 import ../../helpers/mockmarket
 import ../../examples
@@ -33,11 +34,11 @@ checksuite "sales state 'filled'":
                           StorageRequest.none)
     state = SaleFilled.new()
 
-  test "switches to finished state when slot is filled by me":
+  test "switches to proving state when slot is filled by me":
     slot.host = await market.getSigner()
     market.filled = @[slot]
     let next = await state.run(agent)
-    check !next of SaleFinished
+    check !next of SaleProving
 
   test "switches to error state when slot is filled by another host":
     slot.host = Address.example

tests/codex/sales/states/testinitialproving.nim

@@ -0,0 +1,44 @@
+import pkg/asynctest
+import pkg/questionable
+import pkg/chronos
+import pkg/codex/contracts/requests
+import pkg/codex/sales/states/initialproving
+import pkg/codex/sales/states/cancelled
+import pkg/codex/sales/states/failed
+import pkg/codex/sales/states/filling
+import pkg/codex/sales/salesagent
+import pkg/codex/sales/salescontext
+import ../../examples
+import ../../helpers
+
+asyncchecksuite "sales state 'initialproving'":
+
+  let proof = exampleProof()
+  let request = StorageRequest.example
+  let slotIndex = (request.ask.slots div 2).u256
+
+  var state: SaleInitialProving
+  var agent: SalesAgent
+
+  setup:
+    let onProve = proc (slot: Slot): Future[seq[byte]] {.async.} =
+                            return proof
+    let context = SalesContext(onProve: onProve.some)
+    agent = newSalesAgent(context,
+                          request.id,
+                          slotIndex,
+                          request.some)
+    state = SaleInitialProving.new()
+
+  test "switches to cancelled state when request expires":
+    let next = state.onCancelled(request)
+    check !next of SaleCancelled
+
+  test "switches to failed state when request fails":
+    let next = state.onFailed(request)
+    check !next of SaleFailed
+
+  test "switches to filling state when initial proving is complete":
+    let next = await state.run(agent)
+    check !next of SaleFilling
+    check SaleFilling(!next).proof == proof

tests/codex/sales/states/testproving.nim

@@ -1,22 +1,45 @@
-import std/unittest
+import pkg/asynctest
+import pkg/chronos
 import pkg/questionable
 import pkg/codex/contracts/requests
 import pkg/codex/sales/states/proving
 import pkg/codex/sales/states/cancelled
 import pkg/codex/sales/states/failed
-import pkg/codex/sales/states/filled
+import pkg/codex/sales/states/payout
+import pkg/codex/sales/salesagent
+import pkg/codex/sales/salescontext
 import ../../examples
 import ../../helpers
+import ../../helpers/mockmarket
+import ../../helpers/mockclock
 
-checksuite "sales state 'proving'":
+asyncchecksuite "sales state 'proving'":
 
-  let request = StorageRequest.example
+  let slot = Slot.example
-  let slotIndex = (request.ask.slots div 2).u256
+  let request = slot.request
+  let proof = exampleProof()
+
+  var market: MockMarket
+  var clock: MockClock
+  var agent: SalesAgent
   var state: SaleProving
 
   setup:
+    clock = MockClock.new()
+    market = MockMarket.new()
+    let onProve = proc (slot: Slot): Future[seq[byte]] {.async.} =
+                        return proof
+    let context = SalesContext(market: market, clock: clock, onProve: onProve.some)
+    agent = newSalesAgent(context,
+                          request.id,
+                          slot.slotIndex,
+                          request.some)
     state = SaleProving.new()
 
+  proc advanceToNextPeriod(market: Market) {.async.} =
+    let periodicity = await market.periodicity()
+    clock.advance(periodicity.seconds.truncate(int64))
+
   test "switches to cancelled state when request expires":
     let next = state.onCancelled(request)
     check !next of SaleCancelled
@@ -25,7 +48,35 @@ checksuite "sales state 'proving'":
     let next = state.onFailed(request)
     check !next of SaleFailed
 
-  test "switches to filled state when slot is filled":
+  test "submits proofs":
-    let next = state.onSlotFilled(request.id, slotIndex)
+    var receivedIds: seq[SlotId]
-    check !next of SaleFilled
+    var receivedProofs: seq[seq[byte]]
+
+    proc onProofSubmission(id: SlotId, proof: seq[byte]) =
+      receivedIds.add(id)
+      receivedProofs.add(proof)
+
+    let subscription = await market.subscribeProofSubmission(onProofSubmission)
+    market.slotState[slot.id] = SlotState.Filled
+
+    let future = state.run(agent)
+
+    market.setProofRequired(slot.id, true)
+    await market.advanceToNextPeriod()
+
+    check eventuallyCheck receivedIds == @[slot.id] and receivedProofs == @[proof]
+
+    await future.cancelAndWait()
+    await subscription.unsubscribe()
+
+  test "switches to payout state when request is finished":
+    market.slotState[slot.id] = SlotState.Filled
+
+    let future = state.run(agent)
+
+    market.slotState[slot.id] = SlotState.Finished
+    await market.advanceToNextPeriod()
+
+    check eventuallyCheck future.finished
+    check !(future.read()) of SalePayout
 

tests/codex/sales/states/testsimulatedproving.nim

@@ -0,0 +1,96 @@
+import pkg/asynctest
+import pkg/chronos
+import pkg/questionable
+import pkg/codex/contracts/requests
+import pkg/codex/sales/states/provingsimulated
+import pkg/codex/sales/states/proving
+import pkg/codex/sales/states/cancelled
+import pkg/codex/sales/states/failed
+import pkg/codex/sales/states/payout
+import pkg/codex/sales/salesagent
+import pkg/codex/sales/salescontext
+import ../../examples
+import ../../helpers
+import ../../helpers/mockmarket
+import ../../helpers/mockclock
+
+asyncchecksuite "sales state 'simulated-proving'":
+
+  let slot = Slot.example
+  let request = slot.request
+  let proof = exampleProof()
+  let failEveryNProofs = 3
+  let totalProofs = 6
+
+  var market: MockMarket
+  var clock: MockClock
+  var agent: SalesAgent
+  var state: SaleProvingSimulated
+
+  var proofSubmitted: Future[void] = newFuture[void]("proofSubmitted")
+  var submitted: seq[seq[byte]]
+  var subscription: Subscription
+
+  setup:
+    clock = MockClock.new()
+
+    proc onProofSubmission(id: SlotId, proof: seq[byte]) =
+      submitted.add(proof)
+      proofSubmitted.complete()
+      proofSubmitted = newFuture[void]("proofSubmitted")
+
+    market = MockMarket.new()
+    market.slotState[slot.id] = SlotState.Filled
+    market.setProofRequired(slot.id, true)
+    subscription = await market.subscribeProofSubmission(onProofSubmission)
+
+    let onProve = proc (slot: Slot): Future[seq[byte]] {.async.} =
+                        return proof
+    let context = SalesContext(market: market, clock: clock, onProve: onProve.some)
+    agent = newSalesAgent(context,
+                          request.id,
+                          slot.slotIndex,
+                          request.some)
+    state = SaleProvingSimulated.new()
+    state.failEveryNProofs = failEveryNProofs
+
+  teardown:
+    await subscription.unsubscribe()
+
+  proc advanceToNextPeriod(market: Market) {.async.} =
+    let periodicity = await market.periodicity()
+    clock.advance(periodicity.seconds.truncate(int64))
+
+  proc waitForProvingRounds(market: Market, rounds: int) {.async.} =
+    var rnds = rounds - 1 # proof round runs prior to advancing
+    while rnds > 0:
+      await market.advanceToNextPeriod()
+      await proofSubmitted
+      rnds -= 1
+
+  test "switches to cancelled state when request expires":
+    let next = state.onCancelled(request)
+    check !next of SaleCancelled
+
+  test "switches to failed state when request fails":
+    let next = state.onFailed(request)
+    check !next of SaleFailed
+
+  test "submits invalid proof every 3 proofs":
+    let future = state.run(agent)
+
+    await market.waitForProvingRounds(totalProofs)
+    check submitted == @[proof, proof, @[], proof, proof, @[]]
+
+    await future.cancelAndWait()
+
+  test "switches to payout state when request is finished":
+    market.slotState[slot.id] = SlotState.Filled
+
+    let future = state.run(agent)
+
+    market.slotState[slot.id] = SlotState.Finished
+    await market.advanceToNextPeriod()
+
+    check eventuallyCheck future.finished
+    check !(future.read()) of SalePayout

tests/codex/sales/testsales.nim

@@ -13,7 +13,6 @@ import pkg/codex/sales/salescontext
 import pkg/codex/sales/reservations
 import pkg/codex/sales/slotqueue
 import pkg/codex/stores/repostore
-import pkg/codex/proving
 import pkg/codex/blocktype as bt
 import pkg/codex/node
 import ../helpers/mockmarket
@@ -29,7 +28,6 @@ asyncchecksuite "Sales - start":
   var sales: Sales
   var market: MockMarket
   var clock: MockClock
-  var proving: Proving
   var reservations: Reservations
   var repo: RepoStore
   var queue: SlotQueue
@@ -52,19 +50,18 @@ asyncchecksuite "Sales - start":
 
     market = MockMarket.new()
     clock = MockClock.new()
-    proving = Proving.new()
     let repoDs = SQLiteDatastore.new(Memory).tryGet()
     let metaDs = SQLiteDatastore.new(Memory).tryGet()
     repo = RepoStore.new(repoDs, metaDs)
     await repo.start()
-    sales = Sales.new(market, clock, proving, repo)
+    sales = Sales.new(market, clock, repo)
     reservations = sales.context.reservations
     sales.onStore = proc(request: StorageRequest,
                          slot: UInt256,
                          onBatch: BatchProc): Future[?!void] {.async.} =
       return success()
     queue = sales.context.slotQueue
-    proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       return proof
     itemsProcessed = @[]
     request.expiry = (clock.now() + 42).u256
@@ -119,7 +116,6 @@ asyncchecksuite "Sales":
   var sales: Sales
   var market: MockMarket
   var clock: MockClock
-  var proving: Proving
   var reservations: Reservations
   var repo: RepoStore
   var queue: SlotQueue
@@ -152,19 +148,18 @@ asyncchecksuite "Sales":
     market.activeSlots[me] = @[]
 
     clock = MockClock.new()
-    proving = Proving.new()
     let repoDs = SQLiteDatastore.new(Memory).tryGet()
     let metaDs = SQLiteDatastore.new(Memory).tryGet()
     repo = RepoStore.new(repoDs, metaDs)
     await repo.start()
-    sales = Sales.new(market, clock, proving, repo)
+    sales = Sales.new(market, clock, repo)
     reservations = sales.context.reservations
     sales.onStore = proc(request: StorageRequest,
                          slot: UInt256,
                          onBatch: BatchProc): Future[?!void] {.async.} =
       return success()
     queue = sales.context.slotQueue
-    proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       return proof
     await sales.start()
     request.expiry = (clock.now() + 42).u256
@@ -363,7 +358,7 @@ asyncchecksuite "Sales":
 
   test "handles errors during state run":
     var saleFailed = false
-    proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       # raise exception so machine.onError is called
       raise newException(ValueError, "some error")
 
@@ -389,7 +384,7 @@ asyncchecksuite "Sales":
   test "generates proof of storage":
     var provingRequest: StorageRequest
     var provingSlot: UInt256
-    proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       provingRequest = slot.request
       provingSlot = slot.slotIndex
     check isOk await reservations.reserve(availability)
@@ -425,7 +420,7 @@ asyncchecksuite "Sales":
   test "calls onClear when storage becomes available again":
     # fail the proof intentionally to trigger `agent.finish(success=false)`,
     # which then calls the onClear callback
-    proving.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
+    sales.onProve = proc(slot: Slot): Future[seq[byte]] {.async.} =
       raise newException(IOError, "proof failed")
     var clearedRequest: StorageRequest
     var clearedSlotIndex: UInt256
@@ -462,17 +457,6 @@ asyncchecksuite "Sales":
     clock.set(request.expiry.truncate(int64))
     check eventually (await reservations.allAvailabilities) == @[availability]
 
-  test "adds proving for slot when slot is filled":
-    var soldSlotIndex: UInt256
-    sales.onSale = proc(request: StorageRequest,
-                        slotIndex: UInt256) =
-      soldSlotIndex = slotIndex
-    check proving.slots.len == 0
-    check isOk await reservations.reserve(availability)
-    await market.requestStorage(request)
-    check eventuallyCheck proving.slots.len == 1
-    check proving.slots.contains(Slot(request: request, slotIndex: soldSlotIndex))
-
   test "loads active slots from market":
     let me = await market.getSigner()
 

tests/codex/sales/teststates.nim

@@ -2,7 +2,9 @@ import ./states/testunknown
 import ./states/testdownloading
 import ./states/testfilling
 import ./states/testfinished
-import ./states/testproving
+import ./states/testinitialproving
 import ./states/testfilled
+import ./states/testproving
+import ./states/testsimulatedproving
 
 {.warning[UnusedImport]: off.}

tests/codex/testproving.nim

@@ -1,188 +0,0 @@
-import std/sequtils
-import pkg/asynctest
-import pkg/chronos
-import pkg/codex/proving
-import ./helpers/mockmarket
-import ./helpers/mockclock
-import ./helpers/eventually
-import ./examples
-import ./helpers
-
-asyncchecksuite "Proving":
-
-  var proving: Proving
-  var market: MockMarket
-  var clock: MockClock
-
-  setup:
-    market = MockMarket.new()
-    clock = MockClock.new()
-    proving = Proving.new(market, clock)
-    await proving.start()
-
-  teardown:
-    await proving.stop()
-
-  proc advanceToNextPeriod(market: MockMarket) {.async.} =
-    let periodicity = await market.periodicity()
-    clock.advance(periodicity.seconds.truncate(int64))
-
-  test "maintains a list of slots to watch":
-    let slot1, slot2 = Slot.example
-    check proving.slots.len == 0
-    proving.add(slot1)
-    check proving.slots.contains(slot1)
-    proving.add(slot2)
-    check proving.slots.contains(slot1)
-    check proving.slots.contains(slot2)
-
-  test "removes duplicate slots":
-    let slot = Slot.example
-    proving.add(slot)
-    proving.add(slot)
-    check proving.slots.len == 1
-
-  test "invokes callback when proof is required":
-    let slot = Slot.example
-    proving.add(slot)
-    var called: bool
-    proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
-      called = true
-    proving.onProve = onProve
-    market.slotState[slot.id] = SlotState.Filled
-    market.setProofRequired(slot.id, true)
-    await market.advanceToNextPeriod()
-    check eventually called
-
-  test "callback receives slot for which proof is required":
-    let slot1, slot2 = Slot.example
-    proving.add(slot1)
-    proving.add(slot2)
-    var callbackSlots: seq[Slot]
-    proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
-      callbackSlots.add(slot)
-    proving.onProve = onProve
-    market.slotState[slot1.id] = SlotState.Filled
-    market.slotState[slot2.id] = SlotState.Filled
-    market.setProofRequired(slot1.id, true)
-    await market.advanceToNextPeriod()
-    check eventually callbackSlots == @[slot1]
-    market.setProofRequired(slot1.id, false)
-    market.setProofRequired(slot2.id, true)
-    await market.advanceToNextPeriod()
-    check eventually callbackSlots == @[slot1, slot2]
-
-  test "invokes callback when proof is about to be required":
-    let slot = Slot.example
-    proving.add(slot)
-    var called: bool
-    proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
-      called = true
-    proving.onProve = onProve
-    market.setProofRequired(slot.id, false)
-    market.setProofToBeRequired(slot.id, true)
-    market.slotState[slot.id] = SlotState.Filled
-    await market.advanceToNextPeriod()
-    check eventually called
-
-  test "stops watching when slot is finished":
-    let slot = Slot.example
-    proving.add(slot)
-    market.setProofEnd(slot.id, clock.now().u256)
-    await market.advanceToNextPeriod()
-    var called: bool
-    proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
-      called = true
-    proving.onProve = onProve
-    market.setProofRequired(slot.id, true)
-    await market.advanceToNextPeriod()
-    market.slotState[slot.id] = SlotState.Finished
-    check eventually (not proving.slots.contains(slot))
-    check not called
-
-  test "submits proofs":
-    let slot = Slot.example
-    let proof = exampleProof()
-
-    proving.onProve = proc (slot: Slot): Future[seq[byte]] {.async.} =
-      return proof
-
-    var receivedIds: seq[SlotId]
-    var receivedProofs: seq[seq[byte]]
-
-    proc onProofSubmission(id: SlotId, proof: seq[byte]) =
-      receivedIds.add(id)
-      receivedProofs.add(proof)
-
-    let subscription = await proving.subscribeProofSubmission(onProofSubmission)
-
-    proving.add(slot)
-    market.slotState[slot.id] = SlotState.Filled
-    market.setProofRequired(slot.id, true)
-    await market.advanceToNextPeriod()
-
-    check eventually receivedIds == @[slot.id] and receivedProofs == @[proof]
-
-    await subscription.unsubscribe()
-
-suite "Simulated proving":
-
-  var proving: SimulatedProving
-  var subscription: Subscription
-  var market: MockMarket
-  var clock: MockClock
-  var submitted: seq[seq[byte]]
-  var proof: seq[byte]
-  let slot = Slot.example
-  var proofSubmitted: Future[void]
-
-  setup:
-    proof = exampleProof()
-    submitted = @[]
-    market = MockMarket.new()
-    clock = MockClock.new()
-    proofSubmitted = newFuture[void]("proofSubmitted")
-
-  teardown:
-    await subscription.unsubscribe()
-    await proving.stop()
-
-  proc newSimulatedProving(failEveryNProofs: uint) {.async.} =
-    proc onProofSubmission(id: SlotId, proof: seq[byte]) =
-      submitted.add(proof)
-      proofSubmitted.complete()
-      proofSubmitted = newFuture[void]("proofSubmitted")
-
-    proving = SimulatedProving.new(market, clock, failEveryNProofs)
-    proving.onProve = proc (slot: Slot): Future[seq[byte]] {.async.} =
-      return proof
-    subscription = await proving.subscribeProofSubmission(onProofSubmission)
-    proving.add(slot)
-    market.slotState[slot.id] = SlotState.Filled
-    market.setProofRequired(slot.id, true)
-    await proving.start()
-
-  proc advanceToNextPeriod(market: Market) {.async.} =
-    let periodicity = await market.periodicity()
-    clock.advance(periodicity.seconds.truncate(int64))
-
-  proc waitForProvingRounds(market: Market, rounds: uint) {.async.} =
-    var rnds = rounds - 1 # proof round runs prior to advancing
-    while rnds > 0:
-      await market.advanceToNextPeriod()
-      await proofSubmitted
-      rnds -= 1
-
-  test "submits invalid proof every 3 proofs":
-    let failEveryNProofs = 3'u
-    let totalProofs = 6'u
-    await newSimulatedProving(failEveryNProofs)
-    await market.waitForProvingRounds(totalProofs)
-    check submitted == @[proof, proof, @[], proof, proof, @[]]
-
-  test "does not submit invalid proofs when failEveryNProofs is 0":
-    let failEveryNProofs = 0'u
-    let totalProofs = 6'u
-    await newSimulatedProving(failEveryNProofs)
-    await market.waitForProvingRounds(totalProofs)
-    check submitted == proof.repeat(totalProofs)

tests/testCodex.nim

@@ -9,7 +9,6 @@ import ./codex/teststorestream
 import ./codex/testpurchasing
 import ./codex/testsales
 import ./codex/testerasure
-import ./codex/testproving
 import ./codex/testutils
 import ./codex/testclock
 import ./codex/testsystemclock