feat: Assign validation bucket when running validator

codex/codex.nim

@@ -140,7 +140,25 @@ proc bootstrapInteractions(
       host = some HostInteractions.new(clock, sales)
 
     if config.validator:
-      let validation = Validation.new(clock, market, config.validatorMaxSlots)
+      without marketplaceConfig =? (await marketplace.config()).catch:
+        warn "failed to get marketplace config, cannot validate --validator-bucket setting"
+
+      let totalBuckets = marketplaceConfig.validation.validators
+
+      if assignedBucket =? config.validatorBucket and
+         assignedBucket >= totalBuckets:
+        fatal "--validator-bucket parameter out of bounds",
+          lowerBound = 0,
+          upperBound = totalBuckets - 1
+        quit QuitFailure
+
+      let bucket = ValidationBucket.init(config.validatorBucket,
+                                         totalBuckets)
+
+      let validation = Validation.new(clock,
+                                      market,
+                                      config.validatorMaxSlots,
+                                      bucket)
       validator = some ValidatorInteractions.new(clock, validation)
 
     s.codexNode.contracts = (client, host, validator)

codex/conf.nim

@@ -293,6 +293,12 @@ type
         name: "validator-max-slots"
       .}: int
 
+      validatorBucket* {.
+        desc: "Range of SlotIds to validate. If not declared, all SlotIds will be validated, up to `validatorMaxSlots`"
+        defaultValue: uint16.none
+        name: "validator-bucket"
+      .}: Option[uint16]
+
       case persistenceCmd* {.
         defaultValue: noCmd
         command }: PersistenceCmd

codex/contracts/config.nim

@@ -8,6 +8,7 @@ type
   MarketplaceConfig* = object
     collateral*: CollateralConfig
     proofs*: ProofConfig
+    validation*: ValidationConfig
   CollateralConfig* = object
     repairRewardPercentage*: uint8 # percentage of remaining collateral slot has after it has been freed
     maxNumberOfSlashes*: uint8 # frees slot when the number of slashes reaches this value
@@ -18,8 +19,21 @@ type
     timeout*: UInt256 # mark proofs as missing before the timeout (in seconds)
     downtime*: uint8 # ignore this much recent blocks for proof requirements
     zkeyHash*: string # hash of the zkey file which is linked to the verifier
+  ValidationConfig* = object
+    # Number of validators to cover the entire SlotId space, max 65,535
+    # (2^16-1). IMPORTANT: This value should be a power of 2 for even
+    # distribution, otherwise, the last validator will have a significantly less
+    # number of SlotIds to validate. The closest power of 2 without overflow is
+    # 2^15 = 32,768, giving each validator a maximum of 3.534e72 slots to
+    # validate.
+    validators*: uint16
 
 
+func fromTuple(_: type ValidationConfig, tupl: tuple): ValidationConfig =
+  ValidationConfig(
+    validators: tupl[0]
+  )
+
 func fromTuple(_: type ProofConfig, tupl: tuple): ProofConfig =
   ProofConfig(
     period: tupl[0],
@@ -51,6 +65,9 @@ func solidityType*(_: type CollateralConfig): string =
 func solidityType*(_: type MarketplaceConfig): string =
   solidityType(CollateralConfig.fieldTypes)
 
+func encode*(encoder: var AbiEncoder, slot: ValidationConfig) =
+  encoder.write(slot.fieldValues)
+
 func encode*(encoder: var AbiEncoder, slot: ProofConfig) =
   encoder.write(slot.fieldValues)
 
@@ -60,6 +77,10 @@ func encode*(encoder: var AbiEncoder, slot: CollateralConfig) =
 func encode*(encoder: var AbiEncoder, slot: MarketplaceConfig) =
   encoder.write(slot.fieldValues)
 
+func decode*(decoder: var AbiDecoder, T: type ValidationConfig): ?!T =
+  let tupl = ?decoder.read(ValidationConfig.fieldTypes)
+  success ValidationConfig.fromTuple(tupl)
+
 func decode*(decoder: var AbiDecoder, T: type ProofConfig): ?!T =
   let tupl = ?decoder.read(ProofConfig.fieldTypes)
   success ProofConfig.fromTuple(tupl)

codex/contracts/market.nim

@@ -88,6 +88,13 @@ method mySlots*(market: OnChainMarket): Future[seq[SlotId]] {.async.} =
 
     return slots
 
+method myValidationSlots*(market: OnChainMarket, bucketIdx: uint16): Future[seq[SlotId]] {.async.} =
+  convertEthersError:
+    let slots = await market.contract.myValidationSlots(bucketIdx)
+    debug "Fetched my slots for validation", numSlots=len(slots)
+
+    return slots
+
 method requestStorage(market: OnChainMarket, request: StorageRequest){.async.} =
   convertEthersError:
     debug "Requesting storage"

codex/contracts/marketplace.nim

@@ -52,6 +52,7 @@ proc getActiveSlot*(marketplace: Marketplace, id: SlotId): Slot {.contract, view
 
 proc myRequests*(marketplace: Marketplace): seq[RequestId] {.contract, view.}
 proc mySlots*(marketplace: Marketplace): seq[SlotId] {.contract, view.}
+proc myValidationSlots*(marketplace: Marketplace, bucketIdx: uint16): seq[SlotId] {.contract, view.}
 proc requestState*(marketplace: Marketplace, requestId: RequestId): RequestState {.contract, view.}
 proc slotState*(marketplace: Marketplace, slotId: SlotId): SlotState {.contract, view.}
 proc requestEnd*(marketplace: Marketplace, requestId: RequestId): SecondsSince1970 {.contract, view.}

codex/market.nim

@@ -67,6 +67,9 @@ method myRequests*(market: Market): Future[seq[RequestId]] {.base, async.} =
 method mySlots*(market: Market): Future[seq[SlotId]] {.base, async.} =
   raiseAssert("not implemented")
 
+method myValidationSlots*(market: Market, bucketIdx: uint16): Future[seq[SlotId]] {.base, async.} =
+  raiseAssert("not implemented")
+
 method getRequest*(market: Market,
                    id: RequestId):
                   Future[?StorageRequest] {.base, async.} =

codex/validation.nim

@@ -18,6 +18,10 @@ type
     running: Future[void]
     periodicity: Periodicity
     proofTimeout: UInt256
+    slotIdBucket: ValidationBucket
+  ValidationBucket* = object
+    assignedBucket: ?uint16
+    totalBuckets: uint16
 
 logScope:
   topics = "codex validator"
@@ -26,11 +30,19 @@ proc new*(
     _: type Validation,
     clock: Clock,
     market: Market,
-    maxSlots: int
+    maxSlots: int,
+    slotIdBucket: ValidationBucket
 ): Validation =
   ## Create a new Validation instance
   Validation(clock: clock, market: market, maxSlots: maxSlots)
 
+proc init*(
+  _: type ValidationBucket,
+  assignedBucket: ?uint16,
+  totalBuckets: uint16): ValidationBucket =
+
+  ValidationBucket(assignedBucket: assignedBucket, totalBuckets: totalBuckets)
+
 proc slots*(validation: Validation): seq[SlotId] =
   validation.slots.toSeq
 
@@ -43,6 +55,22 @@ proc waitUntilNextPeriod(validation: Validation) {.async.} =
   trace "Waiting until next period", currentPeriod = period
   await validation.clock.waitUntil(periodEnd.truncate(int64) + 1)
 
+proc fetchActiveSlots(validation: Validation): Future[seq[SlotId]] {.async.} =
+  var slots: seq[SlotId]
+
+  if assignedBucket =? validation.slotIdBucket.assignedBucket:
+    slots = await validation.market.myValidationSlots(assignedBucket)
+  else:
+    # no --validator-bucket was set, validate all slots
+    for bucketIdx in 0'u16..validation.slotIdBucket.totalBuckets:
+      slots.add (await validation.market.myValidationSlots(bucketIdx))
+
+  return slots
+
+proc loadActiveSlots(validation: Validation) {.async.} =
+  let slots = await validation.fetchActiveSlots()
+  validation.slots.incl(slots.toHashSet)
+
 proc subscribeSlotFilled(validation: Validation) {.async.} =
   proc onSlotFilled(requestId: RequestId, slotIndex: UInt256) =
     let slotId = slotId(requestId, slotIndex)
@@ -59,7 +87,7 @@ proc removeSlotsThatHaveEnded(validation: Validation) {.async.} =
   for slotId in slots:
     let state = await validation.market.slotState(slotId)
     if state != SlotState.Filled:
-      trace "Removing slot", slotId
+      trace "Slot no longer in filled state, removing", slotId
       ended.incl(slotId)
   validation.slots.excl(ended)
 
@@ -103,6 +131,7 @@ proc run(validation: Validation) {.async.} =
 proc start*(validation: Validation) {.async.} =
   validation.periodicity = await validation.market.periodicity()
   validation.proofTimeout = await validation.market.proofTimeout()
+  await validation.loadActiveSlots()
   await validation.subscribeSlotFilled()
   validation.running = validation.run()
 

vendor/codex-contracts-eth

@@ -1 +1 @@
-Subproject commit 7ad26688a3b75b914d626e2623174a36f4425f51
+Subproject commit c2a17a3a28350b7b9b4e44bfa7d5ec7b91d47359