Bloom filter acceleration for deposit processing

AllTests-mainnet.md

@@ -898,6 +898,12 @@ OK: 6/6 Fail: 0/6 Skip: 0/6
 + Dynamic validator set: updateDynamicValidators() test                                      OK
 ```
 OK: 4/4 Fail: 0/4 Skip: 0/4
+## ValidatorPubKey Bloom filter
+```diff
++ incremental construction with no false positives/negatives                                 OK
++ one-shot construction with no false positives/negatives                                    OK
+```
+OK: 2/2 Fail: 0/2 Skip: 0/2
 ## Zero signature sanity checks
 ```diff
 + SSZ serialization roundtrip of SignedBeaconBlockHeader                                     OK
@@ -993,4 +999,4 @@ OK: 2/2 Fail: 0/2 Skip: 0/2
 OK: 9/9 Fail: 0/9 Skip: 0/9
 
 ---TOTAL---
-OK: 670/675 Fail: 0/675 Skip: 5/675
+OK: 672/677 Fail: 0/677 Skip: 5/677

beacon_chain/bloomfilter.nim

@@ -0,0 +1,49 @@
+# beacon_chain
+# Copyright (c) 2024 Status Research & Development GmbH
+# Licensed and distributed under either of
+#   * MIT license (license terms in the root directory or at https://opensource.org/licenses/MIT).
+#   * Apache v2 license (license terms in the root directory or at https://www.apache.org/licenses/LICENSE-2.0).
+# at your option. This file may not be copied, modified, or distributed except according to those terms.
+
+{.push raises: [].}
+
+import "."/spec/crypto
+
+from stew/bitops2 import getBit, setBit
+from "."/spec/datatypes/base import Validator, pubkey
+from "."/spec/helpers import bytes_to_uint32
+
+const
+  # https://hur.st/bloomfilter/?n=4M&p=&m=8MiB&k=
+  pubkeyBloomFilterScale = 23   # 21 too small, 22 borderline, 24 also ok
+
+type
+  PubkeyBloomFilter* = object
+    data: array[1 shl pubkeyBloomFilterScale, byte]
+
+iterator bloomFilterHashes(pubkey: ValidatorPubKey): auto =
+  const pubkeyBloomFilterMask = (1 shl pubkeyBloomFilterScale) - 1
+  for r in countup(0'u32, 20'u32, 4'u32):
+    # ValidatorPubKeys have fairly uniform entropy; using enough hash
+    # functions also reduces risk of low-entropy portions
+    yield pubkey.blob.toOpenArray(r, r+3).bytes_to_uint32 and
+      pubkeyBloomFilterMask
+
+template incl*(bloomFilter: var PubkeyBloomFilter, pubkey: ValidatorPubKey) =
+  for bloomFilterHash in bloomFilterHashes(pubkey):
+    setBit(bloomFilter.data, bloomFilterHash)
+
+func constructBloomFilter*(x: openArray[Validator]): auto =
+  let res = new PubkeyBloomFilter
+  for m in x:
+    incl(res[], m.pubkey)
+  res
+
+func mightContain*(
+    bloomFilter: PubkeyBloomFilter, pubkey: ValidatorPubKey): bool =
+  # Might return false positive, but never false negative
+  for bloomFilterHash in bloomFilterHashes(pubkey):
+    if not getBit(bloomFilter.data, bloomFilterHash):
+      return false
+
+  true

beacon_chain/spec/state_transition_block.nim

@@ -287,10 +287,13 @@ func findValidatorIndex*(state: ForkyBeaconState, pubkey: ValidatorPubKey):
     if state.validators.asSeq[vidx - 1].pubkey == pubkey:
       return Opt[ValidatorIndex].ok((vidx - 1).ValidatorIndex)
 
-proc process_deposit*(cfg: RuntimeConfig,
-                      state: var ForkyBeaconState,
-                      deposit: Deposit,
-                      flags: UpdateFlags): Result[void, cstring] =
+from ".."/bloomfilter import
+  PubkeyBloomFilter, constructBloomFilter, incl, mightContain
+
+proc process_deposit*(
+    cfg: RuntimeConfig, state: var ForkyBeaconState,
+    bloom_filter: var PubkeyBloomFilter, deposit: Deposit, flags: UpdateFlags):
+    Result[void, cstring] =
   ## Process an Eth1 deposit, registering a validator or increasing its balance.
 
   # Verify the Merkle branch
@@ -309,7 +312,11 @@ proc process_deposit*(cfg: RuntimeConfig,
   let
     pubkey = deposit.data.pubkey
     amount = deposit.data.amount
-    index = findValidatorIndex(state, pubkey)
+    index =
+      if bloom_filter.mightContain(pubkey):
+        findValidatorIndex(state, pubkey)
+      else:
+        Opt.none(ValidatorIndex)
 
   if index.isSome():
     # Increase balance by deposit amount
@@ -335,6 +342,7 @@ proc process_deposit*(cfg: RuntimeConfig,
           return err("process_deposit: too many validators (inactivity_scores)")
 
       doAssert state.validators.len == state.balances.len
+      bloom_filter.incl pubkey
     else:
       # Deposits may come with invalid signatures - in that case, they are not
       # turned into a validator but still get processed to keep the deposit
@@ -465,8 +473,10 @@ proc process_operations(cfg: RuntimeConfig,
   for op in body.attestations:
     operations_rewards.attestations +=
       ? process_attestation(state, op, flags, base_reward_per_increment, cache)
-  for op in body.deposits:
-    ? process_deposit(cfg, state, op, flags)
+  if body.deposits.len > 0:
+    let bloom_filter = constructBloomFilter(state.validators.asSeq)
+    for op in body.deposits:
+      ? process_deposit(cfg, state, bloom_filter[], op, flags)
   for op in body.voluntary_exits:
     ? process_voluntary_exit(cfg, state, op, flags, cache)
   when typeof(body).kind >= ConsensusFork.Capella:

nfuzz/libnfuzz.nim

@@ -8,6 +8,8 @@
 # Required for deserialisation of ValidatorSig in Attestation due to
 # https://github.com/nim-lang/Nim/issues/11225
 
+{.push raises: [].}
+
 import
   stew/ptrops, chronicles,
   ../beacon_chain/networking/network_metadata,
@@ -41,7 +43,7 @@ type
   FuzzCrashError = object of CatchableError
 
 # TODO: change ptr uint to ptr csize_t when available in newer Nim version.
-proc copyState(state: phase0.BeaconState, xoutput: ptr byte,
+func copyState(state: phase0.BeaconState, xoutput: ptr byte,
     xoutput_size: ptr uint): bool {.raises: [FuzzCrashError].} =
   var resultState =
     try:
@@ -132,15 +134,20 @@ proc nfuzz_block(input: openArray[byte], xoutput: ptr byte,
     state_transition(
       getRuntimeConfig(some "mainnet"), data, data.beaconBlock, flags, noRollback).isOk
 
-proc nfuzz_block_header(input: openArray[byte], xoutput: ptr byte,
+func nfuzz_block_header(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
   decodeAndProcess(BlockHeaderInput):
     process_block_header(data.state, data.beaconBlock.message, flags, cache).isOk
 
+from ".."/beacon_chain/bloomfilter import constructBloomFilter
+
 proc nfuzz_deposit(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
   decodeAndProcess(DepositInput):
-    process_deposit(getRuntimeConfig(some "mainnet"), data.state, data.deposit, flags).isOk
+    process_deposit(
+      getRuntimeConfig(some "mainnet"), data.state,
+      constructBloomFilter(data.state.validators.asSeq)[], data.deposit,
+      flags).isOk
 
 proc nfuzz_proposer_slashing(input: openArray[byte], xoutput: ptr byte,
     xoutput_size: ptr uint, disable_bls: bool): bool {.exportc, raises: [FuzzCrashError].} =
@@ -156,7 +163,7 @@ proc nfuzz_voluntary_exit(input: openArray[byte], xoutput: ptr byte,
 # However, list_size needs to be known also outside this proc to allocate xoutput.
 # TODO: rework to copy immediatly in an uint8 openArray, considering we have to
 # go over the list anyhow?
-proc nfuzz_shuffle(input_seed: ptr byte, xoutput: var openArray[uint64]): bool
+func nfuzz_shuffle(input_seed: ptr byte, xoutput: var openArray[uint64]): bool
     {.exportc, raises: [].} =
   var seed: Eth2Digest
   # Should be OK as max 2 bytes are passed by the framework.

tests/all_tests.nim

@@ -21,6 +21,7 @@ import # Unit test
   ./test_block_dag,
   ./test_block_processor,
   ./test_block_quarantine,
+  ./test_bloom_filter,
   ./test_conf,
   ./test_datatypes,
   ./test_deposit_snapshots,

tests/consensus_spec/altair/test_fixture_operations.nim

@@ -114,11 +114,15 @@ suite baseDescription & "Block Header " & preset():
     runTest[altair.BeaconBlock, typeof applyBlockHeader](
       OpBlockHeaderDir, suiteName, "Block Header", "block", applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var altair.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/bellatrix/test_fixture_operations.nim

@@ -120,11 +120,15 @@ suite baseDescription & "Block Header " & preset():
       OpBlockHeaderDir, suiteName, "Block Header", "block",
       applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var bellatrix.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/capella/test_fixture_operations.nim

@@ -137,11 +137,15 @@ suite baseDescription & "BLS to execution change " & preset():
       OpBlsToExecutionChangeDir, suiteName, "BLS to execution change", "address_change",
       applyBlsToExecutionChange, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var capella.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/deneb/test_fixture_operations.nim

@@ -140,11 +140,15 @@ suite baseDescription & "BLS to execution change " & preset():
       OpBlsToExecutionChangeDir, suiteName, "BLS to execution change", "address_change",
       applyBlsToExecutionChange, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var deneb.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](

tests/consensus_spec/phase0/test_fixture_operations.nim

@@ -106,11 +106,15 @@ suite baseDescription & "Block Header " & preset():
       OpBlockHeaderDir, suiteName, "Block Header", "block",
       applyBlockHeader, path)
 
+from ".."/".."/".."/beacon_chain/bloomfilter import constructBloomFilter
+
 suite baseDescription & "Deposit " & preset():
   proc applyDeposit(
       preState: var phase0.BeaconState, deposit: Deposit):
       Result[void, cstring] =
-    process_deposit(defaultRuntimeConfig, preState, deposit, {})
+    process_deposit(
+      defaultRuntimeConfig, preState,
+      constructBloomFilter(preState.validators.asSeq)[], deposit, {})
 
   for path in walkTests(OpDepositsDir):
     runTest[Deposit, typeof applyDeposit](