Validator (#387)

* [contracts] Add SlotFreed event

* [integration] allow test node to be stopped twice

* [cli] add --validator option

* [contracts] remove dead code

* [contracts] instantiate OnChainMarket and OnChainClock only once

* [contracts] add Validation

* [sales] remove duplicate import

* [market] add missing import

* [market] subscribe to all SlotFilled events

* [market] add freeSlot()

* [sales] fix warnings

* [market] subscribe to SlotFreed events

* [contracts] fix warning

* [validator] keep track of filled slots

* [validation] remove slots that have ended

* [proving] absorb Proofs into Market

Both Proofs and Market are abstractions around
the Marketplace contract, having them separately
is more trouble than it's worth at the moment.

* [market] add markProofAsMissing()

* [clock] speed up waiting for clock in tests

* [validator] mark proofs as missing

* [timer] fix error on node shutdown

* [cli] handle --persistence and --validator separately

* [market] allow retrieval of proof timeout value

* [validator] do not subscribe to SlotFreed events

Freed slots are already handled in
removeSlotsThatHaveEnded(), and onSlotsFreed()
interfered with its iterator.

* [validator] Start validation at the start of a new period

To decrease the likelihood that we hit the validation timeout.

* [validator] do not mark proofs as missing after timeout

* [market] check whether proof can be marked as missing

* [validator] simplify validation

Simulate a transaction to mark proof as missing, instead
of trying to keep track of all the conditions that may
lead to a proof being marked as missing.

* [build] use nim-ethers PR #40

Uses "pending" blocktag instead of "latest" blocktag
for better simulation of transactions before sending
them.

https://github.com/status-im/nim-ethers/pull/40

* [integration] integration test for validator

* [validator] monitor a maximum number of slots

Adds cli parameter --validator-max-slots.

* [market] fix missing collateral argument

After rebasing, add the new argument to fillSlot calls.

* [build] update to nim-ethers 0.2.5

* [validator] use Set instead of Table to keep track of slots

* [validator] add logging

* [validator] add test for slot failure

* [market] use "pending" blocktag to use more up to date block time

* [contracts] remove unused import

* [validator] fix: wait until after period ends

The smart contract checks that 'end < block.timestamp',
so we need to wait until the block timestamp is greater
than the period end.

codex/clock.nim

@@ -10,7 +10,7 @@ type
 method now*(clock: Clock): SecondsSince1970 {.base, upraises: [].} =
   raiseAssert "not implemented"
 
-proc waitUntil*(clock: Clock, time: SecondsSince1970) {.async.} =
+method waitUntil*(clock: Clock, time: SecondsSince1970) {.base,async.} =
   while clock.now() < time:
     await sleepAsync(1.seconds)
 

codex/codex.nim

@@ -32,6 +32,7 @@ import ./utils/fileutils
 import ./erasure
 import ./discovery
 import ./contracts
+import ./contracts/clock
 import ./utils/addrutils
 import ./namespaces
 
@@ -100,13 +101,16 @@ proc new(_: type Contracts,
   config: CodexConf,
   repo: RepoStore): Contracts =
 
-  if not config.persistence:
+  if not config.persistence and not config.validator:
     if config.ethAccount.isSome:
-      warn "Ethereum account was set, but persistence is not enabled"
+      warn "Ethereum account was set, but neither persistence nor validator is enabled"
     return
 
   without account =? config.ethAccount:
+    if config.persistence:
       error "Persistence enabled, but no Ethereum account was set"
+    if config.validator:
+      error "Validator enabled, but no Ethereum account was set"
     quit QuitFailure
 
   var deploy: Deployment
@@ -123,17 +127,26 @@ proc new(_: type Contracts,
     error "Marketplace contract address not found in deployment file"
     quit QuitFailure
 
-  # TODO: at some point there may be cli options that enable client-only or host-only
+  let provider = JsonRpcProvider.new(config.ethProvider)
-  # operation, and both client AND host will not necessarily need to be instantiated
+  let signer = provider.getSigner(account)
-  let client = ClientInteractions.new(config.ethProvider,
+  let marketplace = Marketplace.new(marketplaceAddress, signer)
-                                      account,
+  let market = OnChainMarket.new(marketplace)
-                                      marketplaceAddress)
+  let clock = OnChainClock.new(provider)
-  let host = HostInteractions.new(config.ethProvider,
-                                  account,
-                                  repo,
-                                  marketplaceAddress)
 
-  (client.option, host.option)
+  var client: ?ClientInteractions
+  var host: ?HostInteractions
+  var validator: ?ValidatorInteractions
+  if config.persistence:
+    let purchasing = Purchasing.new(market, clock)
+    let proving = Proving.new(market, clock)
+    let sales = Sales.new(market, clock, proving, repo)
+    client = some ClientInteractions.new(clock, purchasing)
+    host = some HostInteractions.new(clock, sales, proving)
+  if config.validator:
+    let validation = Validation.new(clock, market, config.validatorMaxSlots)
+    validator = some ValidatorInteractions.new(clock, validation)
+
+  (client, host, validator)
 
 proc new*(T: type CodexServer, config: CodexConf, privateKey: CodexPrivateKey): T =
 

codex/conf.nim

@@ -218,6 +218,18 @@ type
         name: "eth-deployment"
       .}: Option[string]
 
+      validator* {.
+        desc: "Enables validator, requires an Ethereum node"
+        defaultValue: false
+        name: "validator"
+      .}: bool
+
+      validatorMaxSlots* {.
+        desc: "Maximum number of slots that the validator monitors"
+        defaultValue: 1000
+        name: "validator-max-slots"
+      .}: int
+
     of initNode:
       discard
 

codex/contracts.nim

@@ -2,12 +2,10 @@ import contracts/requests
 import contracts/marketplace
 import contracts/deployment
 import contracts/market
-import contracts/proofs
 import contracts/interactions
 
 export requests
 export marketplace
 export deployment
 export market
-export proofs
 export interactions

codex/contracts/interactions.nim

@@ -1,5 +1,9 @@
 import ./interactions/interactions
 import ./interactions/hostinteractions
 import ./interactions/clientinteractions
+import ./interactions/validatorinteractions
 
-export interactions, hostinteractions, clientinteractions
+export interactions
+export hostinteractions
+export clientinteractions
+export validatorinteractions

codex/contracts/interactions/clientinteractions.nim

@@ -1,12 +1,8 @@
 import pkg/ethers
 import pkg/chronicles
-import pkg/questionable
-import pkg/questionable/results
 
 import ../../purchasing
-import ../marketplace
 import ../market
-import ../proofs
 import ../clock
 import ./interactions
 
@@ -18,16 +14,9 @@ type
     purchasing*: Purchasing
 
 proc new*(_: type ClientInteractions,
-          providerUrl: string,
+          clock: OnChainClock,
-          account: Address,
+          purchasing: Purchasing): ClientInteractions =
-          contractAddress: Address): ?!ClientInteractions =
+  ClientInteractions(clock: clock, purchasing: purchasing)
-
-  without prepared =? prepare(providerUrl, account, contractAddress), error:
-    return failure(error)
-
-  let c = ClientInteractions.new(prepared.clock)
-  c.purchasing = Purchasing.new(prepared.market, prepared.clock)
-  return success(c)
 
 proc start*(self: ClientInteractions) {.async.} =
   await procCall ContractInteractions(self).start()

codex/contracts/interactions/hostinteractions.nim

@@ -1,12 +1,8 @@
 import pkg/ethers
 import pkg/chronicles
-import pkg/questionable
-import pkg/questionable/results
 
 import ../../sales
 import ../../proving
-import ../../stores
-import ../proofs
 import ./interactions
 
 export sales
@@ -19,21 +15,10 @@ type
     proving*: Proving
 
 proc new*(_: type HostInteractions,
-  providerUrl: string,
+          clock: OnChainClock,
-  account: Address,
+          sales: Sales,
-  repo: RepoStore,
+          proving: Proving): HostInteractions =
-  contractAddress: Address): ?!HostInteractions =
+  HostInteractions(clock: clock, sales: sales, proving: proving)
-
-  without prepared =? prepare(providerUrl, account, contractAddress), error:
-    return failure(error)
-
-  let proofs = OnChainProofs.new(prepared.contract)
-  let proving = Proving.new(proofs, prepared.clock)
-
-  let h = HostInteractions.new(prepared.clock)
-  h.sales = Sales.new(prepared.market, prepared.clock, proving, repo)
-  h.proving = proving
-  return success(h)
 
 method start*(self: HostInteractions) {.async.} =
   await procCall ContractInteractions(self).start()

codex/contracts/interactions/interactions.nim

@@ -1,33 +1,13 @@
 import pkg/ethers
-import ../../errors
 import ../clock
 import ../marketplace
 import ../market
 
+export clock
+
 type
   ContractInteractions* = ref object of RootObj
-    clock: OnChainClock
+    clock*: OnChainClock
-  ContractInteractionsError* = object of CodexError
-  ReadDeploymentFileFailureError* = object of ContractInteractionsError
-  ContractAddressError* = object of ContractInteractionsError
-
-proc new*(T: type ContractInteractions,
-          clock: OnChainClock): T =
-  T(clock: clock)
-
-proc prepare*(
-  providerUrl: string = "ws://localhost:8545",
-  account, contractAddress: Address):
-  ?!tuple[contract: Marketplace, market: OnChainMarket, clock: OnChainClock] =
-
-  let provider = JsonRpcProvider.new(providerUrl)
-  let signer = provider.getSigner(account)
-
-  let contract = Marketplace.new(contractAddress, signer)
-  let market = OnChainMarket.new(contract)
-  let clock = OnChainClock.new(signer.provider)
-
-  return success((contract, market, clock))
 
 method start*(self: ContractInteractions) {.async, base.} =
   await self.clock.start()

codex/contracts/interactions/validatorinteractions.nim

@@ -0,0 +1,21 @@
+import ./interactions
+import ../../validation
+
+export validation
+
+type
+  ValidatorInteractions* = ref object of ContractInteractions
+    validation: Validation
+
+proc new*(_: type ValidatorInteractions,
+          clock: OnChainClock,
+          validation: Validation): ValidatorInteractions =
+  ValidatorInteractions(clock: clock, validation: validation)
+
+proc start*(self: ValidatorInteractions) {.async.} =
+  await procCall ContractInteractions(self).start()
+  await self.validation.start()
+
+proc stop*(self: ValidatorInteractions) {.async.} =
+  await self.validation.stop()
+  await procCall ContractInteractions(self).stop()

codex/contracts/market.nim

@@ -39,6 +39,15 @@ proc approveFunds(market: OnChainMarket, amount: UInt256) {.async.} =
 method getSigner*(market: OnChainMarket): Future[Address] {.async.} =
   return await market.signer.getAddress()
 
+method periodicity*(market: OnChainMarket): Future[Periodicity] {.async.} =
+  let config = await market.contract.config()
+  let period = config.proofs.period
+  return Periodicity(seconds: period)
+
+method proofTimeout*(market: OnChainMarket): Future[UInt256] {.async.} =
+  let config = await market.contract.config()
+  return config.proofs.timeout
+
 method myRequests*(market: OnChainMarket): Future[seq[RequestId]] {.async.} =
   return await market.contract.myRequests
 
@@ -104,10 +113,54 @@ method fillSlot(market: OnChainMarket,
   await market.approveFunds(collateral)
   await market.contract.fillSlot(requestId, slotIndex, proof)
 
+method freeSlot*(market: OnChainMarket, slotId: SlotId) {.async.} =
+  await market.contract.freeSlot(slotId)
+
 method withdrawFunds(market: OnChainMarket,
                      requestId: RequestId) {.async.} =
   await market.contract.withdrawFunds(requestId)
 
+method isProofRequired*(market: OnChainMarket,
+                        id: SlotId): Future[bool] {.async.} =
+  try:
+    return await market.contract.isProofRequired(id)
+  except ProviderError as e:
+    if e.revertReason.contains("Slot is free"):
+      return false
+    raise e
+
+method willProofBeRequired*(market: OnChainMarket,
+                            id: SlotId): Future[bool] {.async.} =
+  try:
+    return await market.contract.willProofBeRequired(id)
+  except ProviderError as e:
+    if e.revertReason.contains("Slot is free"):
+      return false
+    raise e
+
+method submitProof*(market: OnChainMarket,
+                    id: SlotId,
+                    proof: seq[byte]) {.async.} =
+  await market.contract.submitProof(id, proof)
+
+method markProofAsMissing*(market: OnChainMarket,
+                           id: SlotId,
+                           period: Period) {.async.} =
+  await market.contract.markProofAsMissing(id, period)
+
+method canProofBeMarkedAsMissing*(market: OnChainMarket,
+                                  id: SlotId,
+                                  period: Period): Future[bool] {.async.} =
+  let provider = market.contract.provider
+  let contractWithoutSigner = market.contract.connect(provider)
+  let overrides = CallOverrides(blockTag: some BlockTag.pending)
+  try:
+    await contractWithoutSigner.markProofAsMissing(id, period, overrides)
+    return true
+  except EthersError as e:
+    trace "Proof can not be marked as missing", msg = e.msg
+    return false
+
 method subscribeRequests(market: OnChainMarket,
                          callback: OnRequest):
                         Future[MarketSubscription] {.async.} =
@@ -116,15 +169,30 @@ method subscribeRequests(market: OnChainMarket,
   let subscription = await market.contract.subscribe(StorageRequested, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
+method subscribeSlotFilled*(market: OnChainMarket,
+                            callback: OnSlotFilled):
+                           Future[MarketSubscription] {.async.} =
+  proc onEvent(event: SlotFilled) {.upraises:[].} =
+    callback(event.requestId, event.slotIndex)
+  let subscription = await market.contract.subscribe(SlotFilled, onEvent)
+  return OnChainMarketSubscription(eventSubscription: subscription)
+
 method subscribeSlotFilled*(market: OnChainMarket,
                             requestId: RequestId,
                             slotIndex: UInt256,
                             callback: OnSlotFilled):
                            Future[MarketSubscription] {.async.} =
-  proc onEvent(event: SlotFilled) {.upraises:[].} =
+  proc onSlotFilled(eventRequestId: RequestId, eventSlotIndex: UInt256) =
-    if event.requestId == requestId and event.slotIndex == slotIndex:
+    if eventRequestId == requestId and eventSlotIndex == slotIndex:
-      callback(event.requestId, event.slotIndex)
+      callback(requestId, slotIndex)
-  let subscription = await market.contract.subscribe(SlotFilled, onEvent)
+  return await market.subscribeSlotFilled(onSlotFilled)
+
+method subscribeSlotFreed*(market: OnChainMarket,
+                           callback: OnSlotFreed):
+                          Future[MarketSubscription] {.async.} =
+  proc onEvent(event: SlotFreed) {.upraises:[].} =
+    callback(event.slotId)
+  let subscription = await market.contract.subscribe(SlotFreed, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
 method subscribeFulfillment(market: OnChainMarket,
@@ -157,5 +225,13 @@ method subscribeRequestFailed*(market: OnChainMarket,
   let subscription = await market.contract.subscribe(RequestFailed, onEvent)
   return OnChainMarketSubscription(eventSubscription: subscription)
 
+method subscribeProofSubmission*(market: OnChainMarket,
+                                 callback: OnProofSubmitted):
+                                Future[MarketSubscription] {.async.} =
+  proc onEvent(event: ProofSubmitted) {.upraises: [].} =
+    callback(event.id, event.proof)
+  let subscription = await market.contract.subscribe(ProofSubmitted, onEvent)
+  return OnChainMarketSubscription(eventSubscription: subscription)
+
 method unsubscribe*(subscription: OnChainMarketSubscription) {.async.} =
   await subscription.eventSubscription.unsubscribe()

codex/contracts/marketplace.nim

@@ -11,6 +11,7 @@ export stint
 export ethers
 export erc20
 export config
+export requests
 
 type
   Marketplace* = ref object of Contract
@@ -21,6 +22,9 @@ type
     requestId* {.indexed.}: RequestId
     slotIndex* {.indexed.}: UInt256
     slotId*: SlotId
+  SlotFreed* = object of Event
+    requestId* {.indexed.}: RequestId
+    slotId*: SlotId
   RequestFulfilled* = object of Event
     requestId* {.indexed.}: RequestId
   RequestCancelled* = object of Event

codex/contracts/proofs.nim

@@ -1,64 +0,0 @@
-import std/strutils
-import pkg/ethers
-import pkg/ethers/testing
-import ../storageproofs/timing/proofs
-import ./marketplace
-
-export proofs
-
-type
-  OnChainProofs* = ref object of Proofs
-    marketplace: Marketplace
-    pollInterval*: Duration
-  ProofsSubscription = proofs.Subscription
-  EventSubscription = ethers.Subscription
-  OnChainProofsSubscription = ref object of ProofsSubscription
-    eventSubscription: EventSubscription
-
-const DefaultPollInterval = 3.seconds
-
-proc new*(_: type OnChainProofs, marketplace: Marketplace): OnChainProofs =
-  OnChainProofs(marketplace: marketplace, pollInterval: DefaultPollInterval)
-
-method periodicity*(proofs: OnChainProofs): Future[Periodicity] {.async.} =
-  let config = await proofs.marketplace.config()
-  let period = config.proofs.period
-  return Periodicity(seconds: period)
-
-method isProofRequired*(proofs: OnChainProofs,
-                        id: SlotId): Future[bool] {.async.} =
-  try:
-    return await proofs.marketplace.isProofRequired(id)
-  except ProviderError as e:
-    if e.revertReason.contains("Slot is free"):
-      return false
-    raise e
-
-method willProofBeRequired*(proofs: OnChainProofs,
-                            id: SlotId): Future[bool] {.async.} =
-  try:
-    return await proofs.marketplace.willProofBeRequired(id)
-  except ProviderError as e:
-    if e.revertReason.contains("Slot is free"):
-      return false
-    raise e
-
-method slotState*(proofs: OnChainProofs,
-                  id: SlotId): Future[SlotState] {.async.} =
-  return await proofs.marketplace.slotState(id)
-
-method submitProof*(proofs: OnChainProofs,
-                    id: SlotId,
-                    proof: seq[byte]) {.async.} =
-  await proofs.marketplace.submitProof(id, proof)
-
-method subscribeProofSubmission*(proofs: OnChainProofs,
-                                 callback: OnProofSubmitted):
-                                Future[ProofsSubscription] {.async.} =
-  proc onEvent(event: ProofSubmitted) {.upraises: [].} =
-    callback(event.id, event.proof)
-  let subscription = await proofs.marketplace.subscribe(ProofSubmitted, onEvent)
-  return OnChainProofsSubscription(eventSubscription: subscription)
-
-method unsubscribe*(subscription: OnChainProofsSubscription) {.async, upraises:[].} =
-  await subscription.eventSubscription.unsubscribe()

codex/market.nim

@@ -4,11 +4,13 @@ import pkg/questionable
 import pkg/ethers/erc20
 import ./contracts/requests
 import ./clock
+import ./periods
 
 export chronos
 export questionable
 export requests
 export SecondsSince1970
+export periods
 
 type
   Market* = ref object of RootObj
@@ -16,12 +18,20 @@ type
   OnRequest* = proc(id: RequestId, ask: StorageAsk) {.gcsafe, upraises:[].}
   OnFulfillment* = proc(requestId: RequestId) {.gcsafe, upraises: [].}
   OnSlotFilled* = proc(requestId: RequestId, slotIndex: UInt256) {.gcsafe, upraises:[].}
+  OnSlotFreed* = proc(slotId: SlotId) {.gcsafe, upraises: [].}
   OnRequestCancelled* = proc(requestId: RequestId) {.gcsafe, upraises:[].}
   OnRequestFailed* = proc(requestId: RequestId) {.gcsafe, upraises:[].}
+  OnProofSubmitted* = proc(id: SlotId, proof: seq[byte]) {.gcsafe, upraises:[].}
 
 method getSigner*(market: Market): Future[Address] {.base, async.} =
   raiseAssert("not implemented")
 
+method periodicity*(market: Market): Future[Periodicity] {.base, async.} =
+  raiseAssert("not implemented")
+
+method proofTimeout*(market: Market): Future[UInt256] {.base, async.} =
+  raiseAssert("not implemented")
+
 method requestStorage*(market: Market,
                        request: StorageRequest) {.base, async.} =
   raiseAssert("not implemented")
@@ -67,6 +77,9 @@ method fillSlot*(market: Market,
                  collateral: UInt256) {.base, async.} =
   raiseAssert("not implemented")
 
+method freeSlot*(market: Market, slotId: SlotId) {.base, async.} =
+  raiseAssert("not implemented")
+
 method withdrawFunds*(market: Market,
                       requestId: RequestId) {.base, async.} =
   raiseAssert("not implemented")
@@ -76,12 +89,40 @@ method subscribeRequests*(market: Market,
                          Future[Subscription] {.base, async.} =
   raiseAssert("not implemented")
 
+method isProofRequired*(market: Market,
+                        id: SlotId): Future[bool] {.base, async.} =
+  raiseAssert("not implemented")
+
+method willProofBeRequired*(market: Market,
+                            id: SlotId): Future[bool] {.base, async.} =
+  raiseAssert("not implemented")
+
+method submitProof*(market: Market,
+                    id: SlotId,
+                    proof: seq[byte]) {.base, async.} =
+  raiseAssert("not implemented")
+
+method markProofAsMissing*(market: Market,
+                           id: SlotId,
+                           period: Period) {.base, async.} =
+  raiseAssert("not implemented")
+
+method canProofBeMarkedAsMissing*(market: Market,
+                                  id: SlotId,
+                                  period: Period): Future[bool] {.base, async.} =
+  raiseAssert("not implemented")
+
 method subscribeFulfillment*(market: Market,
                              requestId: RequestId,
                              callback: OnFulfillment):
                             Future[Subscription] {.base, async.} =
   raiseAssert("not implemented")
 
+method subscribeSlotFilled*(market: Market,
+                            callback: OnSlotFilled):
+                           Future[Subscription] {.base, async.} =
+  raiseAssert("not implemented")
+
 method subscribeSlotFilled*(market: Market,
                             requestId: RequestId,
                             slotIndex: UInt256,
@@ -89,6 +130,11 @@ method subscribeSlotFilled*(market: Market,
                            Future[Subscription] {.base, async.} =
   raiseAssert("not implemented")
 
+method subscribeSlotFreed*(market: Market,
+                           callback: OnSlotFreed):
+                          Future[Subscription] {.base, async.} =
+  raiseAssert("not implemented")
+
 method subscribeRequestCancelled*(market: Market,
                                   requestId: RequestId,
                                   callback: OnRequestCancelled):
@@ -101,5 +147,10 @@ method subscribeRequestFailed*(market: Market,
                              Future[Subscription] {.base, async.} =
   raiseAssert("not implemented")
 
+method subscribeProofSubmission*(market: Market,
+                                 callback: OnProofSubmitted):
+                                Future[Subscription] {.base, async.} =
+  raiseAssert("not implemented")
+
 method unsubscribe*(subscription: Subscription) {.base, async, upraises:[].} =
   raiseAssert("not implemented")

codex/node.nim

@@ -47,6 +47,7 @@ type
   Contracts* = tuple
     client: ?ClientInteractions
     host: ?HostInteractions
+    validator: ?ValidatorInteractions
 
   CodexNodeRef* = ref object
     switch*: Switch
@@ -316,7 +317,7 @@ proc new*(
   engine: BlockExcEngine,
   erasure: Erasure,
   discovery: Discovery,
-  contracts: Contracts = (ClientInteractions.none, HostInteractions.none)): T =
+  contracts = Contracts.default): T =
   T(
     switch: switch,
     blockStore: store,
@@ -389,6 +390,13 @@ proc start*(node: CodexNodeRef) {.async.} =
       error "Unable to start client contract interactions: ", error=error.msg
       node.contracts.client = ClientInteractions.none
 
+  if validatorContracts =? node.contracts.validator:
+    try:
+      await validatorContracts.start()
+    except CatchableError as error:
+      error "Unable to start validator contract interactions: ", error=error.msg
+      node.contracts.validator = ValidatorInteractions.none
+
   node.networkId = node.switch.peerInfo.peerId
   notice "Started codex node", id = $node.networkId, addrs = node.switch.peerInfo.addrs
 

codex/proving.nim

@@ -2,23 +2,22 @@ import std/sets
 import pkg/upraises
 import pkg/questionable
 import pkg/chronicles
-import ./storageproofs
+import ./market
 import ./clock
 
 export sets
-export storageproofs
 
 type
   Proving* = ref object
-    proofs: Proofs
+    market: Market
     clock: Clock
     loop: ?Future[void]
     slots*: HashSet[Slot]
     onProve: ?OnProve
   OnProve* = proc(slot: Slot): Future[seq[byte]] {.gcsafe, upraises: [].}
 
-func new*(_: type Proving, proofs: Proofs, clock: Clock): Proving =
+func new*(_: type Proving, market: Market, clock: Clock): Proving =
-  Proving(proofs: proofs, clock: clock)
+  Proving(market: market, clock: clock)
 
 proc onProve*(proving: Proving): ?OnProve =
   proving.onProve
@@ -30,17 +29,17 @@ func add*(proving: Proving, slot: Slot) =
   proving.slots.incl(slot)
 
 proc getCurrentPeriod(proving: Proving): Future[Period] {.async.} =
-  let periodicity = await proving.proofs.periodicity()
+  let periodicity = await proving.market.periodicity()
   return periodicity.periodOf(proving.clock.now().u256)
 
 proc waitUntilPeriod(proving: Proving, period: Period) {.async.} =
-  let periodicity = await proving.proofs.periodicity()
+  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.proofs.slotState(slot.id)
+    let state = await proving.market.slotState(slot.id)
     if state != SlotState.Filled:
       ended.incl(slot)
   proving.slots.excl(ended)
@@ -50,7 +49,7 @@ proc prove(proving: Proving, slot: Slot) {.async.} =
     raiseAssert "onProve callback not set"
   try:
     let proof = await onProve(slot)
-    await proving.proofs.submitProof(slot.id, proof)
+    await proving.market.submitProof(slot.id, proof)
   except CatchableError as e:
     error "Submitting proof failed", msg = e.msg
 
@@ -61,8 +60,8 @@ proc run(proving: Proving) {.async.} =
       await proving.removeEndedContracts()
       for slot in proving.slots:
         let id = slot.id
-        if (await proving.proofs.isProofRequired(id)) or
+        if (await proving.market.isProofRequired(id)) or
-           (await proving.proofs.willProofBeRequired(id)):
+           (await proving.market.willProofBeRequired(id)):
           asyncSpawn proving.prove(slot)
       await proving.waitUntilPeriod(currentPeriod + 1)
   except CancelledError:
@@ -85,4 +84,4 @@ proc stop*(proving: Proving) {.async.} =
 proc subscribeProofSubmission*(proving: Proving,
                                callback: OnProofSubmitted):
                               Future[Subscription] =
-  proving.proofs.subscribeProofSubmission(callback)
+  proving.market.subscribeProofSubmission(callback)

codex/sales.nim

@@ -4,7 +4,6 @@ import pkg/upraises
 import pkg/stint
 import pkg/chronicles
 import pkg/datastore
-import pkg/upraises
 import ./rng
 import ./market
 import ./clock

codex/storageproofs.nim

@@ -1,7 +1,6 @@
 import ./storageproofs/por
-import ./storageproofs/timing
 import ./storageproofs/stpstore
 import ./storageproofs/stpnetwork
 import ./storageproofs/stpproto
 
-export por, timing, stpstore, stpnetwork, stpproto
+export por, stpstore, stpnetwork, stpproto

codex/storageproofs/storageproofs.nim

@@ -22,9 +22,8 @@ import ./por
 import ./stpnetwork
 import ./stpproto
 import ./stpstore
-import ./timing
 
-export stpnetwork, stpstore, por, timing, stpproto
+export stpnetwork, stpstore, por, stpproto
 
 type
   StorageProofs* = object

codex/storageproofs/timing.nim

@@ -1,4 +0,0 @@
-import ./timing/periods
-import ./timing/proofs
-
-export periods, proofs

codex/storageproofs/timing/proofs.nim

@@ -1,44 +0,0 @@
-import pkg/chronos
-import pkg/stint
-import pkg/upraises
-import ./periods
-import ../../contracts/requests
-
-export chronos
-export stint
-export periods
-export requests
-
-type
-  Proofs* = ref object of RootObj
-  Subscription* = ref object of RootObj
-  OnProofSubmitted* = proc(id: SlotId, proof: seq[byte]) {.gcsafe, upraises:[].}
-
-method periodicity*(proofs: Proofs):
-                   Future[Periodicity] {.base, async.} =
-  raiseAssert("not implemented")
-
-method isProofRequired*(proofs: Proofs,
-                        id: SlotId): Future[bool] {.base, async.} =
-  raiseAssert("not implemented")
-
-method willProofBeRequired*(proofs: Proofs,
-                            id: SlotId): Future[bool] {.base, async.} =
-  raiseAssert("not implemented")
-
-method slotState*(proofs: Proofs,
-                  id: SlotId): Future[SlotState] {.base, async.} =
-  raiseAssert("not implemented")
-
-method submitProof*(proofs: Proofs,
-                    id: SlotId,
-                    proof: seq[byte]) {.base, async.} =
-  raiseAssert("not implemented")
-
-method subscribeProofSubmission*(proofs: Proofs,
-                                 callback: OnProofSubmitted):
-                                Future[Subscription] {.base, async.} =
-  raiseAssert("not implemented")
-
-method unsubscribe*(subscription: Subscription) {.base, async, upraises:[].} =
-  raiseAssert("not implemented")

codex/utils/timer.nim

@@ -32,6 +32,8 @@ proc timerLoop(timer: Timer) {.async.} =
     while true:
       await timer.callback()
       await sleepAsync(timer.interval)
+  except CancelledError:
+    raise
   except CatchableError as exc:
     error "Timer caught unhandled exception: ", name=timer.name, msg=exc.msg
 

codex/validation.nim

@@ -0,0 +1,102 @@
+import std/sets
+import std/sequtils
+import pkg/chronos
+import pkg/chronicles
+import ./market
+import ./clock
+
+export market
+export sets
+
+type
+  Validation* = ref object
+    slots: HashSet[SlotId]
+    maxSlots: int
+    clock: Clock
+    market: Market
+    subscriptions: seq[Subscription]
+    running: Future[void]
+    periodicity: Periodicity
+    proofTimeout: UInt256
+
+logScope:
+  topics = "codex validator"
+
+proc new*(_: type Validation,
+          clock: Clock,
+          market: Market,
+          maxSlots: int): Validation =
+  Validation(clock: clock, market: market, maxSlots: maxSlots)
+
+proc slots*(validation: Validation): seq[SlotId] =
+  validation.slots.toSeq
+
+proc getCurrentPeriod(validation: Validation): UInt256 =
+  return validation.periodicity.periodOf(validation.clock.now().u256)
+
+proc waitUntilNextPeriod(validation: Validation) {.async.} =
+  let period = validation.getCurrentPeriod()
+  let periodEnd = validation.periodicity.periodEnd(period)
+  trace "Waiting until next period", currentPeriod = period
+  await validation.clock.waitUntil(periodEnd.truncate(int64) + 1)
+
+proc subscribeSlotFilled(validation: Validation) {.async.} =
+  proc onSlotFilled(requestId: RequestId, slotIndex: UInt256) =
+    let slotId = slotId(requestId, slotIndex)
+    if slotId notin validation.slots:
+      if validation.slots.len < validation.maxSlots:
+        trace "Adding slot", slotId = $slotId
+        validation.slots.incl(slotId)
+  let subscription = await validation.market.subscribeSlotFilled(onSlotFilled)
+  validation.subscriptions.add(subscription)
+
+proc removeSlotsThatHaveEnded(validation: Validation) {.async.} =
+  var ended: HashSet[SlotId]
+  for slotId in validation.slots:
+    let state = await validation.market.slotState(slotId)
+    if state != SlotState.Filled:
+      trace "Removing slot", slot = $slotId
+      ended.incl(slotId)
+  validation.slots.excl(ended)
+
+proc markProofAsMissing(validation: Validation,
+                        slotId: SlotId,
+                        period: Period) {.async.} =
+  try:
+    if await validation.market.canProofBeMarkedAsMissing(slotId, period):
+      trace "Marking proof as missing", slotId = $slotId, period = period
+      await validation.market.markProofAsMissing(slotId, period)
+  except CancelledError:
+    raise
+  except CatchableError as e:
+    debug "Marking proof as missing failed", msg = e.msg
+
+proc markProofsAsMissing(validation: Validation) {.async.} =
+  for slotId in validation.slots:
+    let previousPeriod = validation.getCurrentPeriod() - 1
+    await validation.markProofAsMissing(slotId, previousPeriod)
+
+proc run(validation: Validation) {.async.} =
+  trace "Validation started"
+  try:
+    while true:
+      await validation.waitUntilNextPeriod()
+      await validation.removeSlotsThatHaveEnded()
+      await validation.markProofsAsMissing()
+  except CancelledError:
+    trace "Validation stopped"
+    discard
+  except CatchableError as e:
+    error "Validation failed", msg = e.msg
+
+proc start*(validation: Validation) {.async.} =
+  validation.periodicity = await validation.market.periodicity()
+  validation.proofTimeout = await validation.market.proofTimeout()
+  await validation.subscribeSlotFilled()
+  validation.running = validation.run()
+
+proc stop*(validation: Validation) {.async.} =
+  await validation.running.cancelAndWait()
+  while validation.subscriptions.len > 0:
+    let subscription = validation.subscriptions.pop()
+    await subscription.unsubscribe()

tests/codex/helpers/mockclock.nim

@@ -1,4 +1,5 @@
 import std/times
+import pkg/chronos
 import codex/clock
 
 export clock
@@ -6,16 +7,33 @@ export clock
 type
   MockClock* = ref object of Clock
     time: SecondsSince1970
+    waiting: seq[Waiting]
+  Waiting = ref object
+    until: SecondsSince1970
+    future: Future[void]
 
 func new*(_: type MockClock,
           time: SecondsSince1970 = getTime().toUnix): MockClock =
   MockClock(time: time)
 
-func set*(clock: MockClock, time: SecondsSince1970) =
+proc set*(clock: MockClock, time: SecondsSince1970) =
   clock.time = time
+  var index = 0
+  while index < clock.waiting.len:
+    if clock.waiting[index].until <= clock.time:
+      clock.waiting[index].future.complete()
+      clock.waiting.del(index)
+    else:
+      inc index
 
-func advance*(clock: MockClock, seconds: int64) =
+proc advance*(clock: MockClock, seconds: int64) =
-  clock.time += seconds
+  clock.set(clock.time + seconds)
 
 method now*(clock: MockClock): SecondsSince1970 =
   clock.time
+
+method waitUntil*(clock: MockClock, time: SecondsSince1970) {.async.} =
+  if time > clock.now():
+    let future = newFuture[void]()
+    clock.waiting.add(Waiting(until: time, future: future))
+    await future

tests/codex/helpers/mockmarket.nim

@@ -1,15 +1,18 @@
 import std/sequtils
 import std/tables
 import std/hashes
+import std/sets
 import pkg/codex/market
 import pkg/codex/contracts/requests
 import pkg/codex/contracts/config
+import ../examples
 
 export market
 export tables
 
 type
   MockMarket* = ref object of Market
+    periodicity: Periodicity
     activeRequests*: Table[Address, seq[RequestId]]
     activeSlots*: Table[Address, seq[SlotId]]
     requested*: seq[StorageRequest]
@@ -18,10 +21,16 @@ type
     slotState*: Table[SlotId, SlotState]
     fulfilled*: seq[Fulfillment]
     filled*: seq[MockSlot]
+    freed*: seq[SlotId]
+    markedAsMissingProofs*: seq[SlotId]
+    canBeMarkedAsMissing: HashSet[SlotId]
     withdrawn*: seq[RequestId]
+    proofsRequired: HashSet[SlotId]
+    proofsToBeRequired: HashSet[SlotId]
+    proofEnds: Table[SlotId, UInt256]
     signer: Address
     subscriptions: Subscriptions
-    config: MarketplaceConfig
+    config*: MarketplaceConfig
   Fulfillment* = object
     requestId*: RequestId
     proof*: seq[byte]
@@ -35,8 +44,10 @@ type
     onRequest: seq[RequestSubscription]
     onFulfillment: seq[FulfillmentSubscription]
     onSlotFilled: seq[SlotFilledSubscription]
+    onSlotFreed: seq[SlotFreedSubscription]
     onRequestCancelled: seq[RequestCancelledSubscription]
     onRequestFailed: seq[RequestFailedSubscription]
+    onProofSubmitted: seq[ProofSubmittedSubscription]
   RequestSubscription* = ref object of Subscription
     market: MockMarket
     callback: OnRequest
@@ -46,9 +57,12 @@ type
     callback: OnFulfillment
   SlotFilledSubscription* = ref object of Subscription
     market: MockMarket
-    requestId: RequestId
+    requestId: ?RequestId
-    slotIndex: UInt256
+    slotIndex: ?UInt256
     callback: OnSlotFilled
+  SlotFreedSubscription* = ref object of Subscription
+    market: MockMarket
+    callback: OnSlotFreed
   RequestCancelledSubscription* = ref object of Subscription
     market: MockMarket
     requestId: RequestId
@@ -57,6 +71,9 @@ type
     market: MockMarket
     requestId: RequestId
     callback: OnRequestCancelled
+  ProofSubmittedSubscription = ref object of Subscription
+    market: MockMarket
+    callback: OnProofSubmitted
 
 proc hash*(address: Address): Hash =
   hash(address.toArray)
@@ -83,6 +100,12 @@ proc new*(_: type MockMarket): MockMarket =
 method getSigner*(market: MockMarket): Future[Address] {.async.} =
   return market.signer
 
+method periodicity*(mock: MockMarket): Future[Periodicity] {.async.} =
+  return Periodicity(seconds: mock.config.proofs.period)
+
+method proofTimeout*(market: MockMarket): Future[UInt256] {.async.} =
+  return market.config.proofs.timeout
+
 method requestStorage*(market: MockMarket, request: StorageRequest) {.async.} =
   market.requested.add(request)
   var subscriptions = market.subscriptions.onRequest
@@ -139,10 +162,20 @@ proc emitSlotFilled*(market: MockMarket,
                      slotIndex: UInt256) =
   var subscriptions = market.subscriptions.onSlotFilled
   for subscription in subscriptions:
-    if subscription.requestId == requestId and
+    let requestMatches =
-       subscription.slotIndex == slotIndex:
+      subscription.requestId.isNone or
+      subscription.requestId == some requestId
+    let slotMatches =
+      subscription.slotIndex.isNone or
+      subscription.slotIndex == some slotIndex
+    if requestMatches and slotMatches:
       subscription.callback(requestId, slotIndex)
 
+proc emitSlotFreed*(market: MockMarket, slotId: SlotId) =
+  var subscriptions = market.subscriptions.onSlotFreed
+  for subscription in subscriptions:
+    subscription.callback(slotId)
+
 proc emitRequestCancelled*(market: MockMarket,
                      requestId: RequestId) =
   var subscriptions = market.subscriptions.onRequestCancelled
@@ -174,6 +207,7 @@ proc fillSlot*(market: MockMarket,
     host: host
   )
   market.filled.add(slot)
+  market.slotState[slotId(slot.requestId, slot.slotIndex)] = SlotState.Filled
   market.emitSlotFilled(requestId, slotIndex)
 
 method fillSlot*(market: MockMarket,
@@ -183,11 +217,58 @@ method fillSlot*(market: MockMarket,
                  collateral: UInt256) {.async.} =
   market.fillSlot(requestId, slotIndex, proof, market.signer)
 
+method freeSlot*(market: MockMarket, slotId: SlotId) {.async.} =
+  market.freed.add(slotId)
+  market.emitSlotFreed(slotId)
+
 method withdrawFunds*(market: MockMarket,
                       requestId: RequestId) {.async.} =
   market.withdrawn.add(requestId)
   market.emitRequestCancelled(requestId)
 
+proc setProofRequired*(mock: MockMarket, id: SlotId, required: bool) =
+  if required:
+    mock.proofsRequired.incl(id)
+  else:
+    mock.proofsRequired.excl(id)
+
+method isProofRequired*(mock: MockMarket,
+                        id: SlotId): Future[bool] {.async.} =
+  return mock.proofsRequired.contains(id)
+
+proc setProofToBeRequired*(mock: MockMarket, id: SlotId, required: bool) =
+  if required:
+    mock.proofsToBeRequired.incl(id)
+  else:
+    mock.proofsToBeRequired.excl(id)
+
+method willProofBeRequired*(mock: MockMarket,
+                            id: SlotId): Future[bool] {.async.} =
+  return mock.proofsToBeRequired.contains(id)
+
+proc setProofEnd*(mock: MockMarket, id: SlotId, proofEnd: UInt256) =
+  mock.proofEnds[id] = proofEnd
+
+method submitProof*(mock: MockMarket, id: SlotId, proof: seq[byte]) {.async.} =
+  for subscription in mock.subscriptions.onProofSubmitted:
+    subscription.callback(id, proof)
+
+method markProofAsMissing*(market: MockMarket,
+                           id: SlotId,
+                           period: Period) {.async.} =
+  market.markedAsMissingProofs.add(id)
+
+proc setCanProofBeMarkedAsMissing*(mock: MockMarket, id: SlotId, required: bool) =
+  if required:
+    mock.canBeMarkedAsMissing.incl(id)
+  else:
+    mock.canBeMarkedAsMissing.excl(id)
+
+method canProofBeMarkedAsMissing*(market: MockMarket,
+                                  id: SlotId,
+                                  period: Period): Future[bool] {.async.} =
+  return market.canBeMarkedAsMissing.contains(id)
+
 method subscribeRequests*(market: MockMarket,
                           callback: OnRequest):
                          Future[Subscription] {.async.} =
@@ -210,6 +291,13 @@ method subscribeFulfillment*(market: MockMarket,
   market.subscriptions.onFulfillment.add(subscription)
   return subscription
 
+method subscribeSlotFilled*(market: MockMarket,
+                            callback: OnSlotFilled):
+                           Future[Subscription] {.async.} =
+  let subscription = SlotFilledSubscription(market: market, callback: callback)
+  market.subscriptions.onSlotFilled.add(subscription)
+  return subscription
+
 method subscribeSlotFilled*(market: MockMarket,
                             requestId: RequestId,
                             slotIndex: UInt256,
@@ -217,13 +305,20 @@ method subscribeSlotFilled*(market: MockMarket,
                            Future[Subscription] {.async.} =
   let subscription = SlotFilledSubscription(
     market: market,
-    requestId: requestId,
+    requestId: some requestId,
-    slotIndex: slotIndex,
+    slotIndex: some slotIndex,
     callback: callback
   )
   market.subscriptions.onSlotFilled.add(subscription)
   return subscription
 
+method subscribeSlotFreed*(market: MockMarket,
+                           callback: OnSlotFreed):
+                          Future[Subscription] {.async.} =
+  let subscription = SlotFreedSubscription(market: market, callback: callback)
+  market.subscriptions.onSlotFreed.add(subscription)
+  return subscription
+
 method subscribeRequestCancelled*(market: MockMarket,
                             requestId: RequestId,
                             callback: OnRequestCancelled):
@@ -248,6 +343,16 @@ method subscribeRequestFailed*(market: MockMarket,
   market.subscriptions.onRequestFailed.add(subscription)
   return subscription
 
+method subscribeProofSubmission*(mock: MockMarket,
+                                 callback: OnProofSubmitted):
+                                Future[Subscription] {.async.} =
+  let subscription = ProofSubmittedSubscription(
+    market: mock,
+    callback: callback
+  )
+  mock.subscriptions.onProofSubmitted.add(subscription)
+  return subscription
+
 method unsubscribe*(subscription: RequestSubscription) {.async.} =
   subscription.market.subscriptions.onRequest.keepItIf(it != subscription)
 
@@ -257,8 +362,14 @@ method unsubscribe*(subscription: FulfillmentSubscription) {.async.} =
 method unsubscribe*(subscription: SlotFilledSubscription) {.async.} =
   subscription.market.subscriptions.onSlotFilled.keepItIf(it != subscription)
 
+method unsubscribe*(subscription: SlotFreedSubscription) {.async.} =
+  subscription.market.subscriptions.onSlotFreed.keepItIf(it != subscription)
+
 method unsubscribe*(subscription: RequestCancelledSubscription) {.async.} =
   subscription.market.subscriptions.onRequestCancelled.keepItIf(it != subscription)
 
 method unsubscribe*(subscription: RequestFailedSubscription) {.async.} =
   subscription.market.subscriptions.onRequestFailed.keepItIf(it != subscription)
+
+method unsubscribe*(subscription: ProofSubmittedSubscription) {.async.} =
+  subscription.market.subscriptions.onProofSubmitted.keepItIf(it != subscription)

tests/codex/helpers/mockproofs.nim

@@ -1,74 +0,0 @@
-import std/sets
-import std/tables
-import std/sequtils
-import pkg/upraises
-import pkg/codex/storageproofs
-
-type
-  MockProofs* = ref object of Proofs
-    periodicity: Periodicity
-    proofsRequired: HashSet[SlotId]
-    proofsToBeRequired: HashSet[SlotId]
-    proofEnds: Table[SlotId, UInt256]
-    subscriptions: seq[MockSubscription]
-    slotStates: Table[SlotId, SlotState]
-  MockSubscription* = ref object of Subscription
-    proofs: MockProofs
-    callback: OnProofSubmitted
-
-const DefaultPeriodLength = 10.u256
-
-func new*(_: type MockProofs): MockProofs =
-  MockProofs(periodicity: Periodicity(seconds: DefaultPeriodLength))
-
-func setPeriodicity*(mock: MockProofs, periodicity: Periodicity) =
-  mock.periodicity = periodicity
-
-method periodicity*(mock: MockProofs): Future[Periodicity] {.async.} =
-  return mock.periodicity
-
-proc setProofRequired*(mock: MockProofs, id: SlotId, required: bool) =
-  if required:
-    mock.proofsRequired.incl(id)
-  else:
-    mock.proofsRequired.excl(id)
-
-method isProofRequired*(mock: MockProofs,
-                        id: SlotId): Future[bool] {.async.} =
-  return mock.proofsRequired.contains(id)
-
-proc setProofToBeRequired*(mock: MockProofs, id: SlotId, required: bool) =
-  if required:
-    mock.proofsToBeRequired.incl(id)
-  else:
-    mock.proofsToBeRequired.excl(id)
-
-method willProofBeRequired*(mock: MockProofs,
-                            id: SlotId): Future[bool] {.async.} =
-  return mock.proofsToBeRequired.contains(id)
-
-proc setProofEnd*(mock: MockProofs, id: SlotId, proofEnd: UInt256) =
-  mock.proofEnds[id] = proofEnd
-
-method submitProof*(mock: MockProofs,
-                    id: SlotId,
-                    proof: seq[byte]) {.async.} =
-  for subscription in mock.subscriptions:
-    subscription.callback(id, proof)
-
-method subscribeProofSubmission*(mock: MockProofs,
-                                 callback: OnProofSubmitted):
-                                Future[Subscription] {.async.} =
-  let subscription = MockSubscription(proofs: mock, callback: callback)
-  mock.subscriptions.add(subscription)
-  return subscription
-
-method unsubscribe*(subscription: MockSubscription) {.async, upraises:[].} =
-  subscription.proofs.subscriptions.keepItIf(it != subscription)
-
-method slotState*(mock: MockProofs,
-                  slotId: SlotId): Future[SlotState] {.async.} =
-  return mock.slotStates[slotId]
-
-proc setSlotState*(mock: MockProofs, slotId: SlotId, state: SlotState) =
-  mock.slotStates[slotId] = state

tests/codex/sales/testreservations.nim

@@ -6,7 +6,6 @@ import pkg/asynctest
 import pkg/datastore
 import pkg/json_serialization
 import pkg/json_serialization/std/options
-import pkg/stew/byteutils
 
 import pkg/codex/stores
 import pkg/codex/sales

tests/codex/sales/testsales.nim

@@ -146,7 +146,6 @@ suite "Sales":
   test "retrieves and stores data locally":
     var storingRequest: StorageRequest
     var storingSlot: UInt256
-    var storingAvailability: Availability
     sales.onStore = proc(request: StorageRequest,
                          slot: UInt256,
                          onBatch: BatchProc): Future[?!void] {.async.} =

tests/codex/testproving.nim

@@ -1,7 +1,7 @@
 import pkg/asynctest
 import pkg/chronos
 import pkg/codex/proving
-import ./helpers/mockproofs
+import ./helpers/mockmarket
 import ./helpers/mockclock
 import ./helpers/eventually
 import ./examples
@@ -9,20 +9,20 @@ import ./examples
 suite "Proving":
 
   var proving: Proving
-  var proofs: MockProofs
+  var market: MockMarket
   var clock: MockClock
 
   setup:
-    proofs = MockProofs.new()
+    market = MockMarket.new()
     clock = MockClock.new()
-    proving = Proving.new(proofs, clock)
+    proving = Proving.new(market, clock)
     await proving.start()
 
   teardown:
     await proving.stop()
 
-  proc advanceToNextPeriod(proofs: MockProofs) {.async.} =
+  proc advanceToNextPeriod(market: MockMarket) {.async.} =
-    let periodicity = await proofs.periodicity()
+    let periodicity = await market.periodicity()
     clock.advance(periodicity.seconds.truncate(int64))
 
   test "maintains a list of slots to watch":
@@ -47,9 +47,9 @@ suite "Proving":
     proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
       called = true
     proving.onProve = onProve
-    proofs.setSlotState(slot.id, SlotState.Filled)
+    market.slotState[slot.id] = SlotState.Filled
-    proofs.setProofRequired(slot.id, true)
+    market.setProofRequired(slot.id, true)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
     check eventually called
 
   test "callback receives slot for which proof is required":
@@ -60,14 +60,14 @@ suite "Proving":
     proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
       callbackSlots.add(slot)
     proving.onProve = onProve
-    proofs.setSlotState(slot1.id, SlotState.Filled)
+    market.slotState[slot1.id] = SlotState.Filled
-    proofs.setSlotState(slot2.id, SlotState.Filled)
+    market.slotState[slot2.id] = SlotState.Filled
-    proofs.setProofRequired(slot1.id, true)
+    market.setProofRequired(slot1.id, true)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
     check eventually callbackSlots == @[slot1]
-    proofs.setProofRequired(slot1.id, false)
+    market.setProofRequired(slot1.id, false)
-    proofs.setProofRequired(slot2.id, true)
+    market.setProofRequired(slot2.id, true)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
     check eventually callbackSlots == @[slot1, slot2]
 
   test "invokes callback when proof is about to be required":
@@ -77,24 +77,24 @@ suite "Proving":
     proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
       called = true
     proving.onProve = onProve
-    proofs.setProofRequired(slot.id, false)
+    market.setProofRequired(slot.id, false)
-    proofs.setProofToBeRequired(slot.id, true)
+    market.setProofToBeRequired(slot.id, true)
-    proofs.setSlotState(slot.id, SlotState.Filled)
+    market.slotState[slot.id] = SlotState.Filled
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
     check eventually called
 
   test "stops watching when slot is finished":
     let slot = Slot.example
     proving.add(slot)
-    proofs.setProofEnd(slot.id, clock.now().u256)
+    market.setProofEnd(slot.id, clock.now().u256)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
     var called: bool
     proc onProve(slot: Slot): Future[seq[byte]] {.async.} =
       called = true
     proving.onProve = onProve
-    proofs.setProofRequired(slot.id, true)
+    market.setProofRequired(slot.id, true)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
-    proofs.setSlotState(slot.id, SlotState.Finished)
+    market.slotState[slot.id] = SlotState.Finished
     check eventually (not proving.slots.contains(slot))
     check not called
 
@@ -115,9 +115,9 @@ suite "Proving":
     let subscription = await proving.subscribeProofSubmission(onProofSubmission)
 
     proving.add(slot)
-    proofs.setSlotState(slot.id, SlotState.Filled)
+    market.slotState[slot.id] = SlotState.Filled
-    proofs.setProofRequired(slot.id, true)
+    market.setProofRequired(slot.id, true)
-    await proofs.advanceToNextPeriod()
+    await market.advanceToNextPeriod()
 
     check eventually receivedIds == @[slot.id] and receivedProofs == @[proof]
 

tests/codex/testvalidation.nim

@@ -0,0 +1,65 @@
+import pkg/asynctest
+import pkg/chronos
+
+import codex/validation
+import ./helpers/mockmarket
+import ./helpers/mockclock
+import ./helpers/eventually
+import ./examples
+
+suite "validation":
+
+  let period = 10
+  let timeout = 5
+  let maxSlots = 100
+  let slot = Slot.example
+  let collateral = slot.request.ask.collateral
+
+  var validation: Validation
+  var market: MockMarket
+  var clock: MockClock
+
+  setup:
+    market = MockMarket.new()
+    clock = MockClock.new()
+    validation = Validation.new(clock, market, maxSlots)
+    market.config.proofs.period = period.u256
+    market.config.proofs.timeout = timeout.u256
+    await validation.start()
+
+  teardown:
+    await validation.stop()
+
+  test "the list of slots that it's monitoring is empty initially":
+    check validation.slots.len == 0
+
+  test "when a slot is filled on chain, it is added to the list":
+    await market.fillSlot(slot.request.id, slot.slotIndex, @[], collateral)
+    check validation.slots == @[slot.id]
+
+  for state in [SlotState.Finished, SlotState.Failed]:
+    test "when slot state changes, it is removed from the list":
+      await market.fillSlot(slot.request.id, slot.slotIndex, @[], collateral)
+      market.slotState[slot.id] = state
+      clock.advance(period)
+      check eventually validation.slots.len == 0
+
+  test "when a proof is missed, it is marked as missing":
+    await market.fillSlot(slot.request.id, slot.slotIndex, @[], collateral)
+    market.setCanProofBeMarkedAsMissing(slot.id, true)
+    clock.advance(period)
+    await sleepAsync(1.millis)
+    check market.markedAsMissingProofs.contains(slot.id)
+
+  test "when a proof can not be marked as missing, it will not be marked":
+    await market.fillSlot(slot.request.id, slot.slotIndex, @[], collateral)
+    market.setCanProofBeMarkedAsMissing(slot.id, false)
+    clock.advance(period)
+    await sleepAsync(1.millis)
+    check market.markedAsMissingProofs.len == 0
+
+  test "it does not monitor more than the maximum number of slots":
+    for _ in 0..<maxSlots + 1:
+      let slot = Slot.example
+      await market.fillSlot(slot.request.id, slot.slotIndex, @[], collateral)
+    check validation.slots.len == maxSlots

tests/contracts/testContracts.nim

@@ -3,7 +3,6 @@ import pkg/chronos
 import pkg/ethers/testing
 import pkg/ethers/erc20
 import codex/contracts
-import codex/storageproofs
 import ../ethertest
 import ./examples
 import ./time

tests/contracts/testDeployment.nim

@@ -7,7 +7,7 @@ suite "Deployment":
   let deploymentFile = "vendor" / "codex-contracts-eth" / "deployment-localhost.json"
 
   test "can be instantiated with a deployment file":
-    let deployment = Deployment.init(deploymentFile)
+    discard Deployment.init(deploymentFile)
 
   test "contract address can be retreived from deployment json":
     let deployment = Deployment.init(deploymentFile)

tests/contracts/testInteractions.nim

@@ -1,46 +0,0 @@
-import std/os
-import pkg/datastore
-import pkg/codex/contracts
-import pkg/codex/stores
-import ../ethertest
-import ./examples
-
-ethersuite "Marketplace Contract Interactions - Client":
-
-  let url = "http://localhost:8545"
-  let account = Address.example
-  let contractAddress = Address.example
-
-  test "can be instantiated with a provider url, account, and contract address":
-    check ClientInteractions.new(url, account, contractAddress).isSuccess
-
-  test "provides purchasing":
-    let client = !ClientInteractions.new(url, account, contractAddress)
-    check client.purchasing != nil
-
-ethersuite "Marketplace Contract Interactions - Host":
-
-  let url = "http://localhost:8545"
-  let account = Address.example
-  let contractAddress = Address.example
-
-  var
-    repo: RepoStore
-    repoDs: Datastore
-    metaDs: Datastore
-
-  setup:
-    repoDs = SQLiteDatastore.new(Memory).tryGet()
-    metaDs = SQLiteDatastore.new(Memory).tryGet()
-    repo = RepoStore.new(repoDs, metaDs)
-
-  test "can be instantiated with a provider url, account, repo, and contract address":
-    check HostInteractions.new(url, account, repo, contractAddress).isSuccess
-
-  test "provides sales":
-    let host = !HostInteractions.new(url, account, repo, contractAddress)
-    check host.sales != nil
-
-  test "provides proving":
-    let host = !HostInteractions.new(url, account, repo, contractAddress)
-    check host.proving != nil

tests/contracts/testMarket.nim

@@ -2,7 +2,6 @@ import std/options
 import pkg/chronos
 import pkg/stew/byteutils
 import codex/contracts
-import codex/storageproofs
 import ../ethertest
 import ./examples
 import ./time
@@ -29,14 +28,17 @@ ethersuite "On-Chain Market":
 
     slotIndex = (request.ask.slots div 2).u256
 
-  proc waitUntilProofRequired(slotId: SlotId) {.async.} =
+  proc advanceToNextPeriod() {.async.} =
     let currentPeriod = periodicity.periodOf(await provider.currentTime())
-    await provider.advanceTimeTo(periodicity.periodEnd(currentPeriod))
+    await provider.advanceTimeTo(periodicity.periodEnd(currentPeriod) + 1)
+
+  proc waitUntilProofRequired(slotId: SlotId) {.async.} =
+    await advanceToNextPeriod()
     while not (
       (await marketplace.isProofRequired(slotId)) and
       (await marketplace.getPointer(slotId)) < 250
     ):
-      await provider.advanceTime(periodicity.seconds)
+      await advanceToNextPeriod()
 
   test "fails to instantiate when contract does not have a signer":
     let storageWithoutSigner = marketplace.connect(provider)
@@ -46,6 +48,17 @@ ethersuite "On-Chain Market":
   test "knows signer address":
     check (await market.getSigner()) == (await provider.getSigner().getAddress())
 
+  test "can retrieve proof periodicity":
+    let periodicity = await market.periodicity()
+    let config = await marketplace.config()
+    let periodLength = config.proofs.period
+    check periodicity.seconds == periodLength
+
+  test "can retrieve proof timeout":
+    let proofTimeout = await market.proofTimeout()
+    let config = await marketplace.config()
+    check proofTimeout == config.proofs.timeout
+
   test "supports marketplace requests":
     await market.requestStorage(request)
 
@@ -82,14 +95,48 @@ ethersuite "On-Chain Market":
     await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
     check (await market.getHost(request.id, slotIndex)) == some accounts[0]
 
-  test "support slot filled subscriptions":
+  test "supports freeing a slot":
+    await market.requestStorage(request)
+    await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
+    await market.freeSlot(slotId(request.id, slotIndex))
+    check (await market.getHost(request.id, slotIndex)) == none Address
+
+  test "supports checking whether proof is required now":
+    check (await market.isProofRequired(slotId(request.id, slotIndex))) == false
+
+  test "supports checking whether proof is required soon":
+    check (await market.willProofBeRequired(slotId(request.id, slotIndex))) == false
+
+  test "submits proofs":
+    await market.submitProof(slotId(request.id, slotIndex), proof)
+
+  test "marks a proof as missing":
+    let slotId = slotId(request, slotIndex)
+    await market.requestStorage(request)
+    await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
+    await waitUntilProofRequired(slotId)
+    let missingPeriod = periodicity.periodOf(await provider.currentTime())
+    await advanceToNextPeriod()
+    await market.markProofAsMissing(slotId, missingPeriod)
+    check (await marketplace.missingProofs(slotId)) == 1
+
+  test "can check whether a proof can be marked as missing":
+    let slotId = slotId(request, slotIndex)
+    await market.requestStorage(request)
+    await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
+    await waitUntilProofRequired(slotId)
+    let missingPeriod = periodicity.periodOf(await provider.currentTime())
+    await advanceToNextPeriod()
+    check (await market.canProofBeMarkedAsMissing(slotId, missingPeriod)) == true
+
+  test "supports slot filled subscriptions":
     await market.requestStorage(request)
     var receivedIds: seq[RequestId]
     var receivedSlotIndices: seq[UInt256]
     proc onSlotFilled(id: RequestId, slotIndex: UInt256) =
       receivedIds.add(id)
       receivedSlotIndices.add(slotIndex)
-    let subscription = await market.subscribeSlotFilled(request.id, slotIndex, onSlotFilled)
+    let subscription = await market.subscribeSlotFilled(onSlotFilled)
     await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
     check receivedIds == @[request.id]
     check receivedSlotIndices == @[slotIndex]
@@ -108,6 +155,17 @@ ethersuite "On-Chain Market":
     check receivedSlotIndices == @[slotIndex]
     await subscription.unsubscribe()
 
+  test "supports slot freed subscriptions":
+    await market.requestStorage(request)
+    await market.fillSlot(request.id, slotIndex, proof, request.ask.collateral)
+    var receivedIds: seq[SlotId]
+    proc onSlotFreed(id: SlotId) =
+      receivedIds.add(id)
+    let subscription = await market.subscribeSlotFreed(onSlotFreed)
+    await market.freeSlot(slotId(request.id, slotIndex))
+    check receivedIds == @[slotId(request.id, slotIndex)]
+    await subscription.unsubscribe()
+
   test "support fulfillment subscriptions":
     await market.requestStorage(request)
     var receivedIds: seq[RequestId]
@@ -172,7 +230,7 @@ ethersuite "On-Chain Market":
           break
         await waitUntilProofRequired(slotId)
         let missingPeriod = periodicity.periodOf(await provider.currentTime())
-        await provider.advanceTime(periodicity.seconds)
+        await advanceToNextPeriod()
         await marketplace.markProofAsMissing(slotId, missingPeriod)
     check receivedIds == @[request.id]
     await subscription.unsubscribe()
@@ -195,6 +253,23 @@ ethersuite "On-Chain Market":
     check receivedIds == @[request.id]
     await subscription.unsubscribe()
 
+  test "supports proof submission subscriptions":
+    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 market.subscribeProofSubmission(onProofSubmission)
+
+    await market.submitProof(slotId(request.id, slotIndex), proof)
+
+    check receivedIds == @[slotId(request.id, slotIndex)]
+    check receivedProofs == @[proof]
+
+    await subscription.unsubscribe()
+
   test "request is none when unknown":
     check isNone await market.getRequest(request.id)
 

tests/contracts/testProofs.nim

@@ -1,57 +0,0 @@
-import codex/contracts
-import ../ethertest
-import ./examples
-
-ethersuite "On-Chain Proofs":
-
-  let contractId = SlotId.example
-  let proof = exampleProof()
-
-  var proofs: OnChainProofs
-  var marketplace: Marketplace
-
-  setup:
-    let deployment = Deployment.init()
-    marketplace = Marketplace.new(!deployment.address(Marketplace), provider.getSigner())
-    proofs = OnChainProofs.new(marketplace)
-
-  test "can retrieve proof periodicity":
-    let periodicity = await proofs.periodicity()
-    let config = await marketplace.config()
-    let periodLength = config.proofs.period
-    check periodicity.seconds == periodLength
-
-  test "supports checking whether proof is required now":
-    check (await proofs.isProofRequired(contractId)) == false
-
-  test "supports checking whether proof is required soon":
-    check (await proofs.willProofBeRequired(contractId)) == false
-
-  test "retrieves correct slot state when request is unknown":
-    check (await proofs.slotState(SlotId.example)) == SlotState.Free
-
-  test "submits proofs":
-    await proofs.submitProof(contractId, proof)
-
-  test "supports proof submission subscriptions":
-    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 proofs.subscribeProofSubmission(onProofSubmission)
-
-    await proofs.submitProof(contractId, proof)
-
-    check receivedIds == @[contractId]
-    check receivedProofs == @[proof]
-
-    await subscription.unsubscribe()
-
-  test "proof not required when slot is empty":
-    check not await proofs.isProofRequired(contractId)
-
-  test "proof will not be required when slot is empty":
-    check not await proofs.willProofBeRequired(contractId)

tests/examples.nim

@@ -1,7 +1,7 @@
 import std/random
 import std/sequtils
 import std/times
-import pkg/codex/proving
+import pkg/codex/contracts/requests
 import pkg/stint
 
 proc example*[T: SomeInteger](_: type T): T =

tests/integration/nodes.nim

@@ -38,10 +38,11 @@ proc startNode*(args: openArray[string], debug = false): NodeProcess =
   node
 
 proc stop*(node: NodeProcess) =
-  let process = node.process
+  if node.process != nil:
-  process.terminate()
+    node.process.terminate()
-  discard process.waitForExit(timeout=5_000)
+    discard node.process.waitForExit(timeout=5_000)
-  process.close()
+    node.process.close()
+    node.process = nil
 
 proc restart*(node: NodeProcess) =
   node.stop()

tests/integration/testproofs.nim

@@ -1,45 +1,97 @@
+import std/os
 import codex/contracts/marketplace
 import codex/contracts/deployment
+import codex/periods
 import ../contracts/time
 import ../codex/helpers/eventually
 import ./twonodes
 
 twonodessuite "Proving integration test", debug1=false, debug2=false:
 
+  let validatorDir = getTempDir() / "CodexValidator"
+
   var marketplace: Marketplace
-  var config: MarketplaceConfig
+  var period: uint64
 
   setup:
     let deployment = Deployment.init()
     marketplace = Marketplace.new(!deployment.address(Marketplace), provider)
-    config = await marketplace.config()
+    period = (await marketplace.config()).proofs.period.truncate(uint64)
 
     # Our Hardhat configuration does use automine, which means that time tracked by `provider.currentTime()` is not
     # advanced until blocks are mined and that happens only when transaction is submitted.
     # As we use in tests provider.currentTime() which uses block timestamp this can lead to synchronization issues.
     await provider.advanceTime(1.u256)
 
-  proc waitUntilPurchaseIsStarted {.async.} =
+  proc waitUntilPurchaseIsStarted(proofProbability: uint64 = 3,
-    discard client2.postAvailability(size=0xFFFFF, duration=200, minPrice=300, maxCollateral=200)
+                                  duration: uint64 = 100 * period,
-    let expiry = (await provider.currentTime()) + 30
+                                  expiry: uint64 = 30) {.async.} =
+    discard client2.postAvailability(
+      size=0xFFFFF,
+      duration=duration,
+      minPrice=300,
+      maxCollateral=200
+    )
     let cid = client1.upload("some file contents")
-    let purchase = client1.requestStorage(cid, duration=100, reward=400, proofProbability=3, expiry=expiry, collateral=100)
+    let expiry = (await provider.currentTime()) + expiry.u256
+    let purchase = client1.requestStorage(
+      cid,
+      expiry=expiry,
+      duration=duration,
+      proofProbability=proofProbability,
+      collateral=100,
+      reward=400
+    )
     check eventually client1.getPurchase(purchase){"state"} == %"started"
 
-  test "hosts submit periodic proofs for slots they fill":
+  proc advanceToNextPeriod {.async.} =
-    await waitUntilPurchaseIsStarted()
+    let periodicity = Periodicity(seconds: period.u256)
+    let currentPeriod = periodicity.periodOf(await provider.currentTime())
+    let endOfPeriod = periodicity.periodEnd(currentPeriod)
+    await provider.advanceTimeTo(endOfPeriod + 1)
 
+  proc startValidator: NodeProcess =
+    startNode([
+      "--data-dir=" & validatorDir,
+      "--api-port=8089",
+      "--disc-port=8099",
+      "--validator",
+      "--eth-account=" & $accounts[2]
+    ], debug = false)
+
+  proc stopValidator(node: NodeProcess) =
+    node.stop()
+    removeDir(validatorDir)
+
+  test "hosts submit periodic proofs for slots they fill":
+    await waitUntilPurchaseIsStarted(proofProbability=1)
     var proofWasSubmitted = false
     proc onProofSubmitted(event: ProofSubmitted) =
       proofWasSubmitted = true
     let subscription = await marketplace.subscribe(ProofSubmitted, onProofSubmitted)
+    await provider.advanceTime(period.u256)
+    check eventually proofWasSubmitted
+    await subscription.unsubscribe()
+
+  test "validator will mark proofs as missing":
+    let validator = startValidator()
+    await waitUntilPurchaseIsStarted(proofProbability=1)
+
+    node2.stop()
+
+    var slotWasFreed = false
+    proc onSlotFreed(event: SlotFreed) =
+      slotWasFreed = true
+    let subscription = await marketplace.subscribe(SlotFreed, onSlotFreed)
 
     for _ in 0..<100:
-      if proofWasSubmitted:
+      if slotWasFreed:
         break
       else:
-        await provider.advanceTime(config.proofs.period)
+        await advanceToNextPeriod()
         await sleepAsync(1.seconds)
 
-    check proofWasSubmitted
+    check slotWasFreed
+
     await subscription.unsubscribe()
+    stopValidator(validator)

tests/testCodex.nim

@@ -13,5 +13,6 @@ import ./codex/testproving
 import ./codex/testutils
 import ./codex/testclock
 import ./codex/testsystemclock
+import ./codex/testvalidation
 
 {.warning[UnusedImport]: off.}

tests/testContracts.nim

@@ -1,8 +1,6 @@
 import ./contracts/testContracts
 import ./contracts/testMarket
-import ./contracts/testProofs
 import ./contracts/testDeployment
-import ./contracts/testInteractions
 import ./contracts/testClock
 
 {.warning[UnusedImport]:off.}

vendor/nim-ethers

@@ -1 +1 @@
-Subproject commit 3c12a65769b9a223028df25f78abb6dde06fef35
+Subproject commit 5a4f786757124c903ab46499689db8273ee5ac80