Validator - support partitioning of the slot id space (#890)

* Adds validatorPartitionSize and validatorPartitionIndex config options

* adds partitioning options to the validation type

* adds partitioning logic to the validator

* ignores partitionIndex when partitionSize is either 0 or 1

* clips the partition index to <<partitionIndex mod partitionSize>>

* handles negative values for the validation partition index

* updates long description of the new validator cli options

* makes default partitionSize to be 0 for better backward compatibility

* Improving formatting on validator CLI

* reactors validation params into a separate type and simplifies validation of validation params

* removes suspected duplication

* fixes typo in validator CLI help

* updates README

* Applies review comments - using optionals and range types to handle validation params

* Adds initializer to the configFactory for validatorMaxSlots

* [Review] update validator CLI description and README

* [Review]: renaming validationParams to validationConfig (config)

* [Review]: move validationconfig.nim to a higher level (next to validation.nim)

* changes backing type of MaxSlots to be int and makes sure slots are validated without limit when maxSlots is set to 0

* adds more end-to-end test for the validator and the groups

* fixes typo in README and conf.nim

* makes `maxSlotsConstraintRespected` and `shouldValidateSlot` private + updates the tests

* fixes public address of the signer account in the marketplace tutorial

* applies review comments - removes two tests

README.md

@@ -1,6 +1,8 @@
 # Codex Decentralized Durability Engine
 
-> The Codex project aims to create a decentralized durability engine that allows persisting data in p2p networks. In other words, it allows storing files and data with predictable durability guarantees for later retrieval.
+> The Codex project aims to create a decentralized durability engine that
+> allows persisting data in p2p networks. In other words, it allows storing 
+> files and data with predictable durability guarantees for later retrieval.
 
 > WARNING: This project is under active development and is considered pre-alpha.
 
@@ -24,7 +26,8 @@ To build the project, clone it and run:
 make update && make
 ```
 
-The executable will be placed under the `build` directory under the project root.
+The executable will be placed under the `build` directory under the project
+root.
 
 Run the client with:
 
@@ -38,33 +41,42 @@ It is possible to configure a Codex node in several ways:
  2. Env. variable
  3. Config
 
-The order of priority is the same as above: Cli arguments > Env variables > Config file values.
+The order of priority is the same as above:
+Cli arguments > Env variables > Config file values.
 
 ### Environment variables
 
-In order to set a configuration option using environment variables, first find the desired CLI option
+In order to set a configuration option using environment variables,
-and then transform it in the following way:
+first find the desired CLI option and then transform it in the following way:
 
  1. prepend it with `CODEX_`
  2. make it uppercase
  3. replace `-` with `_`
 
-For example, to configure `--log-level`, use `CODEX_LOG_LEVEL` as the environment variable name.
+For example, to configure `--log-level`, use `CODEX_LOG_LEVEL` as the
+environment variable name.
 
 ### Configuration file
 
-A [TOML](https://toml.io/en/) configuration file can also be used to set configuration values. Configuration option names and corresponding values are placed in the file, separated by `=`. Configuration option names can be obtained from the `codex --help` command, and should not include the `--` prefix. For example, a node's log level (`--log-level`) can be configured using TOML as follows:
+A [TOML](https://toml.io/en/) configuration file can also be used to set
+configuration values. Configuration option names and corresponding values are
+placed in the file, separated by `=`. Configuration option names can be
+obtained from the `codex --help` command, and should not include
+the `--` prefix. For example, a node's log level (`--log-level`) can be
+configured using TOML as follows:
 
 ```toml
 log-level = "trace"
 ```
 
-The Codex node can then read the configuration from this file using the `--config-file` CLI parameter, like `codex --config-file=/path/to/your/config.toml`.
+The Codex node can then read the configuration from this file using
+the `--config-file` CLI parameter, like
+`codex --config-file=/path/to/your/config.toml`.
 
 ### CLI Options
 
 ```
-build/codex --help
+$ build/codex --help
 Usage:
 
 codex [OPTIONS]... command
@@ -87,8 +99,11 @@ The following options are available:
      --agent-string         Node agent string which is used as identifier in network [=Codex].
      --api-bindaddr         The REST API bind address [=127.0.0.1].
  -p, --api-port             The REST Api port [=8080].
-     --repo-kind            Backend for main repo store (fs, sqlite) [=fs].
+     --api-cors-origin      The REST Api CORS allowed origin for downloading data. '*' will allow all
- -q, --storage-quota        The size of the total storage quota dedicated to the node [=8589934592].
+                            origins, '' will allow none. [=Disallow all cross origin requests to download
+                            data].
+     --repo-kind            Backend for main repo store (fs, sqlite, leveldb) [=fs].
+ -q, --storage-quota        The size of the total storage quota dedicated to the node [=$DefaultQuotaBytes].
  -t, --block-ttl            Default block timeout in seconds - 0 disables the ttl [=$DefaultBlockTtl].
      --block-mi             Time interval in seconds - determines frequency of block maintenance cycle: how
                             often blocks are checked for expiration and cleanup
@@ -109,6 +124,18 @@ The following options are available:
      --marketplace-address  Address of deployed Marketplace contract.
      --validator            Enables validator, requires an Ethereum node [=false].
      --validator-max-slots  Maximum number of slots that the validator monitors [=1000].
+                            If set to 0, the validator will not limit the maximum number of slots it
+                            monitors.
+     --validator-groups     Slot validation groups [=ValidationGroups.none].
+                            A number indicating total number of groups into which the whole slot id space
+                            will be divided. The value must be in the range [2, 65535]. If not provided, the
+                            validator will observe the whole slot id space and the value of the
+                            --validator-group-index parameter will be ignored. Powers of twos are advised
+                            for even distribution.
+     --validator-group-index  Slot validation group index [=0].
+                            The value provided must be in the range [0, validatorGroups). Ignored when
+                            --validator-groups is not provided. Only slot ids satisfying condition [(slotId
+                            mod validationGroups) == groupIndex] will be observed by the validator.
 
 Available sub-commands:
 
@@ -129,19 +156,27 @@ The following options are available:
 
 #### Logging
 
-Codex uses [Chronicles](https://github.com/status-im/nim-chronicles) logging library, which allows great flexibility in working with logs.
+Codex uses [Chronicles](https://github.com/status-im/nim-chronicles) logging
-Chronicles has the concept of topics, which categorize log entries into semantic groups.
+library, which allows great flexibility in working with logs.
+Chronicles has the concept of topics, which categorize log entries into
+semantic groups.
 
-Using the `log-level` parameter, you can set the top-level log level like `--log-level="trace"`, but more importantly,
+Using the `log-level` parameter, you can set the top-level log level like
-you can set log levels for specific topics like `--log-level="info; trace: marketplace,node; error: blockexchange"`,
+`--log-level="trace"`, but more importantly, you can set log levels for
-which sets the top-level log level to `info` and then for topics `marketplace` and `node` sets the level to `trace` and so on.
+specific topics like `--log-level="info; trace: marketplace,node; error: blockexchange"`,
+which sets the top-level log level to `info` and then for topics
+`marketplace` and `node` sets the level to `trace` and so on.
 
 ### Guides
 
 To get acquainted with Codex, consider:
-* running the simple [Codex Two-Client Test](docs/TwoClientTest.md) for a start, and;
+* running the simple [Codex Two-Client Test](docs/TwoClientTest.md) for
-* if you are feeling more adventurous, try [Running a Local Codex Network with Marketplace Support](docs/Marketplace.md) using a local blockchain as well.
+  a start, and;
+* if you are feeling more adventurous, try
+  [Running a Local Codex Network with Marketplace Support](docs/Marketplace.md)
+  using a local blockchain as well.
 
 ## API
 
-The client exposes a REST API that can be used to interact with the clients. Overview of the API can be found on [api.codex.storage](https://api.codex.storage).
+The client exposes a REST API that can be used to interact with the clients.
+Overview of the API can be found on [api.codex.storage](https://api.codex.storage).

codex/codex.nim

@@ -122,7 +122,6 @@ proc bootstrapInteractions(
     else:
       s.codexNode.clock = SystemClock()
 
-    if config.persistence:
     # 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:
@@ -140,7 +139,13 @@ proc bootstrapInteractions(
     host = some HostInteractions.new(clock, sales)
 
     if config.validator:
-      let validation = Validation.new(clock, market, config.validatorMaxSlots)
+      without validationConfig =? ValidationConfig.init(
+        config.validatorMaxSlots,
+        config.validatorGroups,
+        config.validatorGroupIndex), err:
+          error "Invalid validation parameters", err = err.msg
+          quit QuitFailure
+      let validation = Validation.new(clock, market, validationConfig)
       validator = some ValidatorInteractions.new(clock, validation)
 
     s.codexNode.contracts = (client, host, validator)

codex/conf.nim

@@ -37,8 +37,10 @@ import ./logutils
 import ./stores
 import ./units
 import ./utils
+from ./validationconfig import MaxSlots, ValidationGroups
 
 export units, net, codextypes, logutils
+export ValidationGroups, MaxSlots
 
 export
   DefaultQuotaBytes,
@@ -99,7 +101,8 @@ type
 
     logFormat* {.
       hidden
-      desc: "Specifies what kind of logs should be written to stdout (auto, colors, nocolors, json)"
+      desc: "Specifies what kind of logs should be written to stdout (auto, " &
+        "colors, nocolors, json)"
       defaultValueDesc: "auto"
       defaultValue: LogKind.Auto
       name: "log-format" }: LogKind
@@ -164,7 +167,8 @@ type
       name: "net-privkey" }: string
 
     bootstrapNodes* {.
-      desc: "Specifies one or more bootstrap nodes to use when connecting to the network"
+      desc: "Specifies one or more bootstrap nodes to use when " &
+        "connecting to the network"
       abbr: "b"
       name: "bootstrap-node" }: seq[SignedPeerRecord]
 
@@ -192,7 +196,8 @@ type
       abbr: "p" }: Port
 
     apiCorsAllowedOrigin* {.
-      desc: "The REST Api CORS allowed origin for downloading data. '*' will allow all origins, '' will allow none.",
+      desc: "The REST Api CORS allowed origin for downloading data. " &
+        "'*' will allow all origins, '' will allow none.",
       defaultValue: string.none
       defaultValueDesc: "Disallow all cross origin requests to download data"
       name: "api-cors-origin" }: Option[string]
@@ -218,7 +223,9 @@ type
       abbr: "t" }: Duration
 
     blockMaintenanceInterval* {.
-      desc: "Time interval in seconds - determines frequency of block maintenance cycle: how often blocks are checked for expiration and cleanup"
+      desc: "Time interval in seconds - determines frequency of block " &
+        "maintenance cycle: how often blocks are checked " &
+        "for expiration and cleanup"
       defaultValue: DefaultBlockMaintenanceInterval
       defaultValueDesc: $DefaultBlockMaintenanceInterval
       name: "block-mi" }: Duration
@@ -230,7 +237,8 @@ type
       name: "block-mn" }: int
 
     cacheSize* {.
-      desc: "The size of the block cache, 0 disables the cache - might help on slow hardrives"
+      desc: "The size of the block cache, 0 disables the cache - " &
+        "might help on slow hardrives"
       defaultValue: 0
       defaultValueDesc: "0"
       name: "cache-size"
@@ -290,9 +298,35 @@ type
 
       validatorMaxSlots* {.
         desc: "Maximum number of slots that the validator monitors"
+        longDesc: "If set to 0, the validator will not limit " &
+          "the maximum number of slots it monitors"
         defaultValue: 1000
         name: "validator-max-slots"
-      .}: int
+      .}: MaxSlots
+
+      validatorGroups* {.
+        desc: "Slot validation groups"
+        longDesc: "A number indicating total number of groups into " &
+          "which the whole slot id space will be divided. " &
+          "The value must be in the range [2, 65535]. " &
+          "If not provided, the validator will observe " &
+          "the whole slot id space and the value of " &
+          "the --validator-group-index parameter will be ignored. " &
+          "Powers of twos are advised for even distribution"
+        defaultValue: ValidationGroups.none
+        name: "validator-groups"
+      .}: Option[ValidationGroups]
+      
+      validatorGroupIndex* {.
+        desc: "Slot validation group index"
+        longDesc: "The value provided must be in the range " &
+          "[0, validatorGroups). Ignored when --validator-groups " &
+          "is not provided. Only slot ids satisfying condition " &
+          "[(slotId mod validationGroups) == groupIndex] will be " &
+          "observed by the validator"
+        defaultValue: 0
+        name: "validator-group-index"
+      .}: uint16
 
       rewardRecipient* {.
         desc: "Address to send payouts to (eg rewards and refunds)"
@@ -546,7 +580,10 @@ proc updateLogLevel*(logLevel: string) {.upraises: [ValueError].} =
   try:
     setLogLevel(parseEnum[LogLevel](directives[0].toUpperAscii))
   except ValueError:
-    raise (ref ValueError)(msg: "Please specify one of: trace, debug, info, notice, warn, error or fatal")
+    raise (ref ValueError)(
+      msg: "Please specify one of: trace, debug, " &
+        "info, notice, warn, error or fatal"
+    )
 
   if directives.len > 1:
     for topicName, settings in parseTopicDirectives(directives[1..^1]):

codex/validation.nim

@@ -1,23 +1,27 @@
 import std/sets
 import std/sequtils
 import pkg/chronos
+import pkg/questionable/results
+
+import ./validationconfig
 import ./market
 import ./clock
 import ./logutils
 
 export market
 export sets
+export validationconfig
 
 type
   Validation* = ref object
     slots: HashSet[SlotId]
-    maxSlots: int
     clock: Clock
     market: Market
     subscriptions: seq[Subscription]
     running: Future[void]
     periodicity: Periodicity
     proofTimeout: UInt256
+    config: ValidationConfig
 
 logScope:
   topics = "codex validator"
@@ -26,10 +30,9 @@ proc new*(
   _: type Validation,
   clock: Clock,
   market: Market,
-    maxSlots: int
+  config: ValidationConfig
 ): Validation =
-  ## Create a new Validation instance
+  Validation(clock: clock, market: market, config: config)
-  Validation(clock: clock, market: market, maxSlots: maxSlots)
 
 proc slots*(validation: Validation): seq[SlotId] =
   validation.slots.toSeq
@@ -43,11 +46,27 @@ proc waitUntilNextPeriod(validation: Validation) {.async.} =
   trace "Waiting until next period", currentPeriod = period
   await validation.clock.waitUntil(periodEnd.truncate(int64) + 1)
 
+func groupIndexForSlotId*(slotId: SlotId,
+                          validationGroups: ValidationGroups): uint16 =
+  let slotIdUInt256 = UInt256.fromBytesBE(slotId.toArray)
+  (slotIdUInt256 mod validationGroups.u256).truncate(uint16)
+
+func maxSlotsConstraintRespected(validation: Validation): bool =
+  validation.config.maxSlots == 0 or
+    validation.slots.len < validation.config.maxSlots
+
+func shouldValidateSlot(validation: Validation, slotId: SlotId): bool =
+  if (validationGroups =? validation.config.groups):
+    (groupIndexForSlotId(slotId, validationGroups) ==
+    validation.config.groupIndex) and
+    validation.maxSlotsConstraintRespected
+  else:
+    validation.maxSlotsConstraintRespected
+
 proc subscribeSlotFilled(validation: Validation) {.async.} =
   proc onSlotFilled(requestId: RequestId, slotIndex: UInt256) =
     let slotId = slotId(requestId, slotIndex)
-    if slotId notin validation.slots:
+    if validation.shouldValidateSlot(slotId):
-      if validation.slots.len < validation.maxSlots:
       trace "Adding slot", slotId
       validation.slots.incl(slotId)
   let subscription = await validation.market.subscribeSlotFilled(onSlotFilled)

codex/validationconfig.nim

@@ -0,0 +1,36 @@
+import std/strformat
+import pkg/questionable
+import pkg/questionable/results
+
+type
+  ValidationGroups* = range[2..65535]
+  MaxSlots* = int
+  ValidationConfig* = object
+    maxSlots: MaxSlots
+    groups: ?ValidationGroups
+    groupIndex: uint16
+
+func init*(
+    _: type ValidationConfig,
+    maxSlots: MaxSlots,
+    groups: ?ValidationGroups,
+    groupIndex: uint16 = 0): ?!ValidationConfig =
+  if maxSlots < 0:
+    return failure "The value of maxSlots must be greater than " &
+        fmt"or equal to 0! (got: {maxSlots})"
+  if validationGroups =? groups and groupIndex >= uint16(validationGroups):
+    return failure "The value of the group index must be less than " &
+        fmt"validation groups! (got: {groupIndex = }, " &
+        fmt"groups = {validationGroups})"
+  
+  success ValidationConfig(
+      maxSlots: maxSlots, groups: groups, groupIndex: groupIndex)
+
+func maxSlots*(config: ValidationConfig): MaxSlots =
+  config.maxSlots
+
+func groups*(config: ValidationConfig): ?ValidationGroups =
+  config.groups
+
+func groupIndex*(config: ValidationConfig): uint16 =
+  config.groupIndex

docs/Marketplace.md

@@ -79,7 +79,7 @@ echo ${GETH_SIGNER_ADDR} > geth_signer_address.txt
 
 > Here make sure you replace `0x0000000000000000000000000000000000000000`
 > with your public address of the signer account
-> (`0x93976895c4939d99837C8e0E1779787718EF8368` in our example).
+> (`0x33A904Ad57D0E2CB8ffe347D3C0E83C2e875E7dB` in our example).
 
 ### 1.2. Configure The Network and Create the Genesis Block
 

tests/codex/testvalidation.nim

@@ -1,4 +1,6 @@
 import pkg/chronos
+import std/strformat
+import std/random
 
 import codex/validation
 import codex/periods
@@ -12,7 +14,8 @@ import ./helpers
 asyncchecksuite "validation":
   let period = 10
   let timeout = 5
-  let maxSlots = 100
+  let maxSlots = MaxSlots(100)
+  let validationGroups = ValidationGroups(8).some
   let slot = Slot.example
   let proof = Groth16Proof.example
   let collateral = slot.request.ask.collateral
@@ -20,11 +23,23 @@ asyncchecksuite "validation":
   var validation: Validation
   var market: MockMarket
   var clock: MockClock
+  var groupIndex: uint16
+
+  proc initValidationConfig(maxSlots: MaxSlots,
+                            validationGroups: ?ValidationGroups,
+                            groupIndex: uint16 = 0): ValidationConfig =
+    without validationConfig =? ValidationConfig.init(
+      maxSlots, groups=validationGroups, groupIndex), error:
+      raiseAssert fmt"Creating ValidationConfig failed! Error msg: {error.msg}"
+    validationConfig
 
   setup:
+    groupIndex = groupIndexForSlotId(slot.id, !validationGroups)
     market = MockMarket.new()
     clock = MockClock.new()
-    validation = Validation.new(clock, market, maxSlots)
+    let validationConfig = initValidationConfig(
+        maxSlots, validationGroups, groupIndex)
+    validation = Validation.new(clock, market, validationConfig)
     market.config.proofs.period = period.u256
     market.config.proofs.timeout = timeout.u256
     await validation.start()
@@ -41,12 +56,69 @@ asyncchecksuite "validation":
   test "the list of slots that it's monitoring is empty initially":
     check validation.slots.len == 0
 
+  for (validationGroups, groupIndex) in [(100, 100'u16), (100, 101'u16)]:
+    test "initializing ValidationConfig fails when groupIndex is " &
+        "greater than or equal to validationGroups " &
+        fmt"(testing for {groupIndex = }, {validationGroups = })":
+      let groups = ValidationGroups(validationGroups).some
+      let validationConfig = ValidationConfig.init(
+          maxSlots, groups = groups, groupIndex = groupIndex)
+      check validationConfig.isFailure == true
+      check validationConfig.error.msg == "The value of the group index " &
+          "must be less than validation groups! " &
+          fmt"(got: {groupIndex = }, groups = {!groups})"
+  
+  test "initializing ValidationConfig fails when maxSlots is negative":
+    let maxSlots = -1
+    let validationConfig = ValidationConfig.init(
+        maxSlots = maxSlots, groups = ValidationGroups.none)
+    check validationConfig.isFailure == true
+    check validationConfig.error.msg == "The value of maxSlots must " &
+        fmt"be greater than or equal to 0! (got: {maxSlots})"
+  
+  test "initializing ValidationConfig fails when maxSlots is negative " &
+      "(validationGroups set)":
+    let maxSlots = -1
+    let validationConfig = ValidationConfig.init(
+        maxSlots = maxSlots, groups = validationGroups, groupIndex)
+    check validationConfig.isFailure == true
+    check validationConfig.error.msg == "The value of maxSlots must " &
+        fmt"be greater than or equal to 0! (got: {maxSlots})"
+
+  test "slot is not observed if it is not in the validation group":
+    let validationConfig = initValidationConfig(maxSlots, validationGroups,
+        (groupIndex + 1) mod uint16(!validationGroups))
+    let validation = Validation.new(clock, market, validationConfig)
+    await validation.start()
+    await market.fillSlot(slot.request.id, slot.slotIndex, proof, collateral)
+    await validation.stop()
+    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, proof, collateral)
     check validation.slots == @[slot.id]
   
+  test "slot should be observed if maxSlots is set to 0":
+    let validationConfig = initValidationConfig(
+        maxSlots = 0, ValidationGroups.none)
+    let validation = Validation.new(clock, market, validationConfig)
+    await validation.start()
+    await market.fillSlot(slot.request.id, slot.slotIndex, proof, collateral)
+    await validation.stop()
+    check validation.slots == @[slot.id]
+
+  test "slot should be observed if validation group is not set (and " &
+      "maxSlots is not 0)":
+    let validationConfig = initValidationConfig(
+        maxSlots, ValidationGroups.none)
+    let validation = Validation.new(clock, market, validationConfig)
+    await validation.start()
+    await market.fillSlot(slot.request.id, slot.slotIndex, proof, collateral)
+    await validation.stop()
+    check validation.slots == @[slot.id]
+
   for state in [SlotState.Finished, SlotState.Failed]:
-    test "when slot state changes, it is removed from the list":
+    test fmt"when slot state changes to {state}, it is removed from the list":
       await market.fillSlot(slot.request.id, slot.slotIndex, proof, collateral)
       market.slotState[slot.id] = state
       advanceToNextPeriod()
@@ -67,7 +139,13 @@ asyncchecksuite "validation":
     check market.markedAsMissingProofs.len == 0
 
   test "it does not monitor more than the maximum number of slots":
+    let validationGroups = ValidationGroups.none
+    let validationConfig = initValidationConfig(
+        maxSlots, validationGroups)
+    let validation = Validation.new(clock, market, validationConfig)
+    await validation.start()
     for _ in 0..<maxSlots + 1:
       let slot = Slot.example
       await market.fillSlot(slot.request.id, slot.slotIndex, proof, collateral)
+    await validation.stop()
     check validation.slots.len == maxSlots