Add Era1 backfill to portal_bridge history mode (#2077)

fluffy/fluffy.nim

@@ -329,7 +329,13 @@ proc run(config: PortalConf) {.raises: [CatchableError].} =
   ## Starting the JSON-RPC APIs
   if config.rpcEnabled:
     let ta = initTAddress(config.rpcAddress, config.rpcPort)
-    var rpcHttpServerWithProxy = RpcProxy.new([ta], config.proxyUri)
+
+    let rpcHttpServer = RpcHttpServer.new()
+    # Note: Set maxRequestBodySize to 4MB instead of 1MB as there are blocks
+    # that reach that limit (in hex, for gossip method).
+    rpcHttpServer.addHttpServer(ta, maxRequestBodySize = 4 * 1_048_576)
+    var rpcHttpServerWithProxy = RpcProxy.new(rpcHttpServer, config.proxyUri)
+
     rpcHttpServerWithProxy.installDiscoveryApiHandlers(d)
     rpcHttpServerWithProxy.installWeb3ApiHandlers()
     if stateNetwork.isSome():

fluffy/rpc/rpc_calls/rpc_portal_debug_calls.nim

@@ -19,6 +19,7 @@ createRpcSigsFromNim(RpcClient):
     era1File: string, epochAccumulatorFile: Opt[string]
   ): bool
 
+  proc portal_historyGossipHeaders(era1File: string): bool
   proc portal_historyGossipBlockContent(era1File: string): bool
   proc portal_history_storeContent(dataFile: string): bool
   proc portal_history_propagate(dataFile: string): bool

fluffy/tools/portal_bridge/portal_bridge_conf.nim

@@ -7,10 +7,24 @@
 
 {.push raises: [].}
 
-import std/uri, confutils, confutils/std/net, nimcrypto/hash, ../../logging
+import std/[os, uri], confutils, confutils/std/net, nimcrypto/hash, ../../logging
 
 export net
 
+proc defaultEthDataDir*(): string =
+  let dataDir =
+    when defined(windows):
+      "AppData" / "Roaming" / "EthData"
+    elif defined(macosx):
+      "Library" / "Application Support" / "EthData"
+    else:
+      ".cache" / "eth-data"
+
+  getHomeDir() / dataDir
+
+proc defaultEra1DataDir*(): string =
+  defaultEthDataDir() / "era1"
+
 type
   TrustedDigest* = MDigest[32 * 8]
 
@@ -83,6 +97,27 @@ type
         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",
+        defaultValue: true,
+        name: "latest"
+      .}: bool
+
+      backfill* {.
+        desc:
+          "Randomly backfill block headers, bodies and receipts into the network from the era1 files",
+        defaultValue: false,
+        name: "backfill"
+      .}: bool
+
+      era1Dir* {.
+        desc: "The directory where all era1 files are stored",
+        defaultValue: defaultEra1DataDir(),
+        defaultValueDesc: defaultEra1DataDir(),
+        name: "era1-dir"
+      .}: InputDir
     of PortalBridgeCmd.state:
       discard
 

fluffy/tools/portal_bridge/portal_bridge_history.nim

@@ -14,10 +14,14 @@ import
   results,
   stew/byteutils,
   eth/common/[eth_types, eth_types_rlp],
+  eth/keys,
+  eth/p2p/discoveryv5/random2,
   ../../../nimbus/beacon/web3_eth_conv,
   ../../../hive_integration/nodocker/engine/engine_client,
-  ../../rpc/[portal_rpc_client],
+  ../../rpc/portal_rpc_client,
   ../../network/history/[history_content, history_network],
+  ../../network_metadata,
+  ../../eth_data/[era1, history_data_ssz_e2s, history_data_seeding],
   ./portal_bridge_conf
 
 from stew/objects import checkedEnumAssign
@@ -285,6 +289,124 @@ proc runLatestLoop(
     else:
       warn "Block gossip took longer than the poll interval"
 
+proc gossipHeadersWithProof(
+    portalClient: RpcClient,
+    era1File: string,
+    epochAccumulatorFile: Opt[string] = Opt.none(string),
+    verifyEra = false,
+): Future[Result[void, string]] {.async: (raises: []).} =
+  let f = ?Era1File.open(era1File)
+
+  if verifyEra:
+    let _ = ?f.verify()
+
+  # Note: building the accumulator takes about 150ms vs 10ms for reading it,
+  # so it is probably not really worth using the read version considering the
+  # UX hassle it adds to provide the accumulator ssz files.
+  let epochAccumulator =
+    if epochAccumulatorFile.isNone:
+      ?f.buildAccumulator()
+    else:
+      ?readEpochAccumulatorCached(epochAccumulatorFile.get())
+
+  for (contentKey, contentValue) in f.headersWithProof(epochAccumulator):
+    let peers =
+      try:
+        await portalClient.portal_historyGossip(
+          contentKey.asSeq.toHex(), contentValue.toHex()
+        )
+      except CatchableError as e:
+        return err("JSON-RPC error: " & $e.msg)
+    info "Block header gossiped", peers, contentKey
+
+  ok()
+
+proc gossipBlockContent(
+    portalClient: RpcClient, era1File: string, verifyEra = false
+): Future[Result[void, string]] {.async: (raises: []).} =
+  let f = ?Era1File.open(era1File)
+
+  if verifyEra:
+    let _ = ?f.verify()
+
+  for (contentKey, contentValue) in f.blockContent():
+    let peers =
+      try:
+        await portalClient.portal_historyGossip(
+          contentKey.asSeq.toHex(), contentValue.toHex()
+        )
+      except CatchableError as e:
+        return err("JSON-RPC error: " & $e.msg)
+    info "Block content gossiped", peers, contentKey
+
+  ok()
+
+proc runBackfillLoop(
+    portalClient: RpcClient, web3Client: RpcClient, era1Dir: string
+) {.async: (raises: [CancelledError]).} =
+  let
+    rng = newRng()
+    accumulator =
+      try:
+        SSZ.decode(finishedAccumulator, FinishedAccumulator)
+      except SerializationError as err:
+        raiseAssert "Invalid baked-in accumulator: " & err.msg
+
+  while true:
+    let
+      # Grab a random era1 to backfill
+      era = rng[].rand(int(era(network_metadata.mergeBlockNumber - 1)))
+      root = accumulator.historicalEpochs[era]
+      eraFile = era1FileName("mainnet", Era1(era), Digest(data: root))
+
+    # Note:
+    # There are two design options here:
+    # 1. Provide the Era1 file through the fluffy custom debug API and let
+    # fluffy process the Era1 file and gossip the content from there.
+    # 2. Process the Era1 files in the bridge and call the
+    # standardized gossip JSON-RPC method.
+    #
+    # Option 2. is more conceptually clean and compatible due to no usage of
+    # custom API, however it will involve invoking a lot of JSON-RPC calls
+    # to pass along block data (in hex).
+    # Option 2. is used here. Switch to Option 1. can be made in case efficiency
+    # turns out the be a problem. It is however a bit more tricky to know when a
+    # new era1 can be gossiped (might need another custom json-rpc that checks
+    # the offer queue)
+    when false:
+      info "Gossip headers from era1 file", eraFile
+      let headerRes =
+        try:
+          await portalClient.portal_historyGossipHeaders(eraFile)
+        except CatchableError as e:
+          error "JSON-RPC method failed", error = e.msg
+          false
+
+      if headerRes:
+        info "Gossip block content from era1 file", eraFile
+        let res =
+          try:
+            await portalClient.portal_historyGossipBlockContent(eraFile)
+          except CatchableError as e:
+            error "JSON-RPC method failed", error = e.msg
+            false
+        if res:
+          error "Failed to gossip block content from era1 file", eraFile
+      else:
+        error "Failed to gossip headers from era1 file", eraFile
+    else:
+      info "Gossip headers from era1 file", eraFile
+      (await portalClient.gossipHeadersWithProof(eraFile)).isOkOr:
+        error "Failed to gossip headers from era1 file", error, eraFile
+        continue
+
+      info "Gossip block content from era1 file", eraFile
+      (await portalClient.gossipBlockContent(eraFile)).isOkOr:
+        error "Failed to gossip block content from era1 file", error, eraFile
+        continue
+
+      info "Succesfully gossiped era1 file", eraFile
+
 proc runHistory*(config: PortalBridgeConf) =
   let
     portalClient = newRpcHttpClient()
@@ -306,7 +428,11 @@ proc runHistory*(config: PortalBridgeConf) =
     except CatchableError as e:
       error "Failed to connect to web3 RPC", error = $e.msg
 
-  asyncSpawn runLatestLoop(portalClient, web3Client, config.blockVerify)
+  if config.latest:
+    asyncSpawn runLatestLoop(portalClient, web3Client, config.blockVerify)
+
+  if config.backfill:
+    asyncSpawn runBackfillLoop(portalClient, web3Client, config.era1Dir.string)
 
   while true:
     poll()