portal bridge: Add sync backfill mode + a bit of bridge rework (#3770)

Author: kdeme

portal/bridge/history/portal_history_bridge.nim

@@ -26,14 +26,15 @@ import
   ../../database/era1_db,
   ../../../execution_chain/common/[hardforks, chain_config],
   ../common/rpc_helpers,
-  ../nimbus_portal_bridge_conf
+  ../nimbus_portal_bridge_conf,
+  beacon_chain/process_state
 
 from ../../network/network_metadata import loadAccumulator
 
 const newHeadPollInterval = 6.seconds # Slot with potential block is every 12s
 
 type PortalHistoryBridge = ref object
-  portalClient: RpcClient
+  portalClient: PortalRpcClient
   web3Client: RpcClient
   gossipQueue: AsyncQueue[(seq[byte], seq[byte])]
   cfg*: ChainConfig
@@ -128,18 +129,86 @@ proc gossipBlockContent(
   ok()
 
 proc runBackfillLoop(
-    bridge: PortalHistoryBridge, era1Dir: string, startEra: uint64, endEra: uint64
+    bridge: PortalHistoryBridge,
+    era1Dir: string,
+    startEra: uint64,
+    endEra: uint64,
+    loop: bool,
 ) {.async: (raises: [CancelledError]).} =
   let accumulator = loadAccumulator()
 
-  for era in startEra .. endEra:
-    let
-      root = accumulator.historicalEpochs[era]
-      era1File = era1Dir / era1FileName("mainnet", Era1(era), Digest(data: root))
+  while true:
+    for era in startEra .. endEra:
+      let
+        root = accumulator.historicalEpochs[era]
+        era1File = era1Dir / era1FileName("mainnet", Era1(era), Digest(data: root))
 
-    (await bridge.gossipBlockContent(era1File)).isOkOr:
-      error "Failed to gossip block content from era1 file", error, era1File
-      continue
+      (await bridge.gossipBlockContent(era1File)).isOkOr:
+        error "Failed to gossip block content from era1 file", error, era1File
+        continue
+
+    if not loop:
+      ProcessState.scheduleStop("backfill_complete")
+      break
+
+    info "Completed backfill loop, starting over"
+
+proc runBackfillLoopSyncMode(
+    bridge: PortalHistoryBridge,
+    era1Dir: string,
+    startEra: uint64,
+    endEra: uint64,
+    loop: bool,
+) {.async: (raises: [CancelledError]).} =
+  let
+    rng = newRng()
+    db = Era1DB.new(era1Dir, "mainnet", loadAccumulator(), bridge.cfg.posBlock.get())
+    blockLowerBound = startEra * EPOCH_SIZE # inclusive
+    blockUpperBound = ((endEra + 1) * EPOCH_SIZE) - 1 # inclusive
+    blockNumberQueue = newAsyncQueue[uint64](50)
+
+  proc blockWorker() {.async: (raises: [CancelledError]).} =
+    while true:
+      let blockNumber = await blockNumberQueue.popFirst()
+
+      logScope:
+        blockNumber = blockNumber
+
+      var blockTuple: BlockTuple
+      db.getBlockTuple(blockNumber, blockTuple).isOkOr:
+        error "Failed to get block tuple", error
+        continue
+
+      block bodyBlock:
+        let _ = (await historyGetBlockBody(bridge.portalClient, blockTuple.header)).valueOr:
+          info "Failed to find block body content, gossiping..", error = $error
+          await bridge.gossipBlockBody(blockNumber, blockTuple.body)
+          break bodyBlock
+
+      block receiptsBlock:
+        let _ = (await historyGetReceipts(bridge.portalClient, blockTuple.header)).valueOr:
+          info "Failed to find block receipts content, gossiping..", error = $error
+          await bridge.gossipReceipts(
+            blockNumber, blockTuple.receipts.to(StoredReceipts)
+          )
+          break receiptsBlock
+
+  var workers: seq[Future[void]] = @[]
+  for i in 0 ..< 50:
+    workers.add blockWorker()
+
+  while true:
+    for blockNumber in blockLowerBound .. blockUpperBound:
+      if blockNumber mod EPOCH_SIZE == 0:
+        info "Backfilling task at block", blockNumber, era = blockNumber div EPOCH_SIZE
+
+      await blockNumberQueue.addLast(blockNumber)
+
+    if not loop:
+      ProcessState.scheduleStop("backfill_complete")
+      break
+
+    info "Completed backfill loop, starting over"
 
 proc runBackfillLoopAuditMode(
     bridge: PortalHistoryBridge, era1Dir: string, startEra: uint64, endEra: uint64
@@ -163,51 +232,23 @@ proc runBackfillLoopAuditMode(
       error "Failed to get block tuple", error
       continue
 
-    var bodySuccess, receiptsSuccess = false
-
-    # body
     block bodyBlock:
-      let _ =
-        try:
-          (
-            await bridge.portalClient.portal_historyGetBlockBody(
-              rlp.encode(blockTuple.header).to0xHex()
-            )
-          )
-        except CatchableError as e:
-          error "Failed to find block body content", error = e.msg
-          break bodyBlock
-
-      info "Retrieved block body from Portal network"
-      bodySuccess = true
+      let _ = (await historyGetBlockBody(bridge.portalClient, blockTuple.header)).valueOr:
+        info "Failed to find block body content, gossiping..", error = $error
+        await bridge.gossipBlockBody(blockNumber, blockTuple.body)
+        break bodyBlock
 
-    # receipts
     block receiptsBlock:
-      let _ =
-        try:
-          (
-            await bridge.portalClient.portal_historyGetReceipts(
-              rlp.encode(blockTuple.header).to0xHex()
-            )
-          )
-        except CatchableError as e:
-          error "Failed to find block receipts content", error = e.msg
-          break receiptsBlock
-
-      info "Retrieved block receipts from Portal network"
-      receiptsSuccess = true
-
-    # Gossip missing content
-    if not bodySuccess:
-      await bridge.gossipBlockBody(blockNumber, blockTuple.body)
-    if not receiptsSuccess:
-      await bridge.gossipReceipts(blockNumber, blockTuple.receipts.to(StoredReceipts))
+      let _ = (await historyGetReceipts(bridge.portalClient, blockTuple.header)).valueOr:
+        info "Failed to find block receipts content, gossiping..", error = $error
+        await bridge.gossipReceipts(blockNumber, blockTuple.receipts.to(StoredReceipts))
+        break receiptsBlock
 
-    await sleepAsync(2.seconds)
+    await sleepAsync(1.seconds)
 
 proc runHistory*(config: PortalBridgeConf) =
   let bridge = PortalHistoryBridge(
-    portalClient: newRpcClientConnect(config.portalRpcUrl),
+    portalClient: PortalRpcClient.init(newRpcClientConnect(config.portalRpcUrl)),
     web3Client: newRpcClientConnect(config.web3Url),
     gossipQueue: newAsyncQueue[(seq[byte], seq[byte])](config.gossipConcurrency),
     cfg: chainConfigForNetwork(MainNet),
@@ -223,9 +264,8 @@ proc runHistory*(config: PortalBridgeConf) =
 
         while true:
           try:
-            let putContentResult = await bridge.portalClient.portal_historyPutContent(
-              contentKeyHex, contentValueHex
-            )
+            let putContentResult = await RpcClient(bridge.portalClient)
+            .portal_historyPutContent(contentKeyHex, contentValueHex)
             let
               peers = putContentResult.peerCount
               accepted = putContentResult.acceptMetadata.acceptedCount
@@ -287,12 +327,19 @@ proc runHistory*(config: PortalBridgeConf) =
   if config.latest:
     asyncSpawn bridge.runLatestLoop(config.blockVerify)
 
-  if config.backfill:
-    if config.audit:
-      asyncSpawn bridge.runBackfillLoopAuditMode(
-        config.era1Dir.string, config.startEra, config.endEra
-      )
-    else:
-      asyncSpawn bridge.runBackfillLoop(
-        config.era1Dir.string, config.startEra, config.endEra
-      )
+  case config.backfillMode
+  of BackfillMode.none:
+    if not config.latest:
+      ProcessState.scheduleStop("no_backfill_no_latest")
+  of BackfillMode.regular:
+    asyncSpawn bridge.runBackfillLoop(
+      config.era1Dir.string, config.startEra, config.endEra, config.backfillLoop
+    )
+  of BackfillMode.sync:
+    asyncSpawn bridge.runBackfillLoopSyncMode(
+      config.era1Dir.string, config.startEra, config.endEra, config.backfillLoop
+    )
+  of BackfillMode.audit:
+    asyncSpawn bridge.runBackfillLoopAuditMode(
+      config.era1Dir.string, config.startEra, config.endEra
+    )

portal/bridge/nimbus_portal_bridge_conf.nim

@@ -36,6 +36,12 @@ proc defaultEra1DataDir*(): string =
 const defaultEndEra* = uint64(era(network_metadata.mergeBlockNumber - 1))
 
 type
+  BackfillMode* = enum
+    none
+    regular
+    sync
+    audit
+
   TrustedDigest* = MDigest[32 * 8]
 
   PortalBridgeCmd* = enum
@@ -91,23 +97,36 @@ type
       .}: JsonRpcUrl
 
       blockVerify* {.
-        desc: "Verify the block header, body and receipts",
+        desc:
+          "Verify the block body and receipts against the received header. Does not verify against the chain.",
         defaultValue: false,
         name: "block-verify"
       .}: bool
 
       latest* {.
         desc:
-          "Follow the head of the chain and gossip latest block header, body and receipts into the network",
+          "Follow the head of the chain and gossip latest block body and receipts into the network. Requires web3-url to be set.",
         defaultValue: false,
         name: "latest"
       .}: bool
 
-      backfill* {.
+      backfillMode* {.
         desc:
-          "Randomly backfill pre-merge block headers, bodies and receipts into the network from the era1 files",
-        defaultValue: false,
-        name: "backfill"
+          "Backfill mode to use for the history bridge. Requires access to era1 files.",
+        longDesc:
+          "Valid values:\n" & "  none    — backfill disabled\n" &
+          "  regular — Gossip all block bodies and receipts from era1\n" &
+          "  sync    — Download all block bodies and receipts and gossip missing content\n" &
+          "  audit   — Download randomly sampled block bodies and receipts and gossip missing content",
+        defaultValue: BackfillMode.regular,
+        name: "backfill-mode"
+      .}: BackfillMode
+
+      backfillLoop* {.
+        desc:
+          "Restart the backfill loop when it finishes (only applies to sync and regular modes)",
+        defaultValue: true,
+        name: "backfill-loop"
       .}: bool
 
       startEra* {.desc: "The era to start from", defaultValue: 0, name: "start-era".}:
@@ -117,13 +136,6 @@ type
         desc: "The era to stop at", defaultValue: defaultEndEra, name: "end-era"
       .}: uint64
 
-      audit* {.
-        desc:
-          "Run pre-merge backfill in audit mode, which will only gossip content that if failed to fetch from the network",
-        defaultValue: true,
-        name: "audit"
-      .}: bool
-
       era1Dir* {.
         desc: "The directory where all era1 files are stored",
         defaultValue: defaultEra1DataDir(),

portal/docs/the_fluffy_book/docs/history-content-bridging.md

@@ -17,10 +17,9 @@ Run a Portal client with the Portal JSON-RPC API enabled, e.g. Nimbus Portal cli
 > Note: The `--storage-capacity:0` option is not required, but it is added here
 for the use case where the node's only focus is on gossiping content from the portal bridge.
 
-### Step 2: Run an EL client
+### Step 2: Run an EL client (Optional)
 
-The Nimbus Portal bridge needs access to the EL JSON-RPC API, either through a local
-Ethereum client or via a web3 provider.
+To seed the latest blocks, the Nimbus Portal bridge needs access to the EL JSON-RPC API, either through a local Ethereum client or via a web3 provider.
 
 ### Step 3: Run the Portal bridge in history mode
 
@@ -29,19 +28,24 @@ Build & run the Nimbus Portal bridge:
 make nimbus_portal_bridge
 
 WEB3_URL="http://127.0.0.1:8548" # Replace with your provider.
-./build/nimbus_portal_bridge history --web3-url:${WEB3_URL}
+./build/nimbus_portal_bridge history --era1-dir:/somedir/era1/ --web3-url:${WEB3_URL}
 ```
 
-By default the Nimbus Portal bridge will run in `--latest` mode, which means that only the
-latest block content will be gossiped into the network.
+By default the Nimbus Portal bridge runs in `--backfill-mode:regular`. This means the bridge will access era1 files to gossip block bodies and receipts into the network. There are other backfill modes available which first try to download the data from the network before gossiping it into the network, e.g. sync and audit mode.
 
-It also has a `--backfill` mode which will gossip pre-merge blocks
-from `era1` files into the network. By default the bridge will audit first whether
-the content is available on the network and if not it will gossip it into the
-network.
+Example: regular backfill (gossip everything):
+```bash
+./build/nimbus_portal_bridge history --backfill-mode:regular --era1-dir:/somedir/era1/
+```
+
+Example: audit-mode backfill (random sample & gossip missing):
+```bash
+./build/nimbus_portal_bridge history --backfill-mode:audit --era1-dir:/somedir/era1/
+```
+
+It is also possible to enable `--latest` mode, which means that the latest block content will be gossiped into the network.
 
-E.g. run latest + backfill with audit mode:
 ```bash
 WEB3_URL="http://127.0.0.1:8548" # Replace with your provider.
-./build/nimbus_portal_bridge history --latest:true --backfill:true --audit:true --era1-dir:/somedir/era1/ --web3-url:${WEB3_URL}
+./build/nimbus_portal_bridge history --latest:true --backfill-mode:none --era1-dir:/somedir/era1/ --web3-url:${WEB3_URL}
 ```