track column verification state per popSidecars (#8841)

* track column verification in column quarantine

* update some comments

* review

* redesign approach

* cleanup

* rework some parts

* remove unused imports

* items() isn't perfect, and it's not an iterator like sequtils items, but better than all()

---------

Co-authored-by: tersec <tersec@users.noreply.github.com>

Author: agnxsh

beacon_chain/el/el_getblobs_service.nim

@@ -182,7 +182,7 @@ proc attemptGetBlobs*(
       # skip the EL fetch and enqueue with the existing columns.
       if forkyBlck.root in self.columnFirstFetched:
         let sidecarsOpt =
-          self.fuluColumnQuarantine[].popSidecars(forkyBlck.root)
+          self.fuluColumnQuarantine[].popSidecarsForImport(forkyBlck.root)
         if sidecarsOpt.isSome():
           if not quarantine[].removeSidecarless(forkyBlck.root):
             return
@@ -318,7 +318,8 @@ proc attemptGetBlobs*(
       # column sidecars we just installed.
       self.partialColumnQuarantine[].pruneForBlock(forkyBlck.root)
 
-      let sidecarsOpt = self.fuluColumnQuarantine[].popSidecars(forkyBlck.root)
+      let sidecarsOpt =
+        self.fuluColumnQuarantine[].popSidecarsForImport(forkyBlck.root)
 
       self.blockProcessor.enqueueBlock(MsgSource.gossip, forkyBlck, sidecarsOpt)
     elif consensusFork == ConsensusFork.Heze:

beacon_chain/gossip_processing/block_processor.nim

@@ -14,7 +14,7 @@ import
   ../sszdump
 
 from std/deques import Deque, addLast, contains, initDeque, items, len, shrink
-from std/sequtils import anyIt, filterIt
+from std/sequtils import anyIt
 from ../consensus_object_pools/consensus_manager import
   ConsensusManager, to, updateHead, updateExecutionHead, checkExpectedEnvelope
 from ../consensus_object_pools/blockchain_dag import
@@ -130,11 +130,39 @@ type
 
   NoSidecars* = typeof(())
   SomeOptSidecars =
-    NoSidecars | Opt[BlobSidecars] | Opt[fulu.DataColumnSidecars] |
+    NoSidecars | Opt[BlobSidecars] | Opt[fulu.DataColumnSidecarsForImport] |
     Opt[gloas.DataColumnSidecars]
 
 const noSidecars* = default(NoSidecars)
 
+proc popSidecarsForImport*(
+    quarantine: var FuluColumnQuarantine,
+    blockRoot: Eth2Digest
+): Opt[fulu.DataColumnSidecarsForImport] =
+  ## Pop the columns of ``blockRoot``, split by whether their KZG proofs have
+  ## already been checked. The drain has to stay adjacent to the pop: read any
+  ## later, it could answer for another set of columns popped for the same root.
+  let sidecars = quarantine.popSidecars(blockRoot).valueOr:
+    return Opt.none(fulu.DataColumnSidecarsForImport)
+  let unverified = quarantine.popPendingVerify(blockRoot)
+
+  var res: fulu.DataColumnSidecarsForImport
+  for sidecar in sidecars:
+    if sidecar[].index in unverified:
+      res.untrusted.add(sidecar)
+    else:
+      res.trusted.add(sidecar)
+  Opt.some(res)
+
+proc popSidecarsForImport*(
+    quarantine: var GloasColumnQuarantine,
+    blockRoot: Eth2Digest
+): Opt[gloas.DataColumnSidecars] =
+  ## Gloas checks every column in `addPayload`; only the drain matters here.
+  let sidecars = quarantine.popSidecars(blockRoot)
+  discard quarantine.popPendingVerify(blockRoot)
+  sidecars
+
 # Initialization
 # ------------------------------------------------------------------------------
 
@@ -225,11 +253,12 @@ proc verifySidecars(
 
 proc verifySidecars(
     signedBlock: fulu.SignedBeaconBlock,
-    sidecarsOpt: Opt[fulu.DataColumnSidecars],
+    sidecarsOpt: Opt[fulu.DataColumnSidecarsForImport],
 ): Result[void, VerifierError] =
   sidecarsOpt.isErrOr:
-    if value.len > 0 and signedBlock.message.body.blob_kzg_commitments.len > 0:
-      verify_data_column_sidecar_kzg_proofs(value).isOkOr:
+    if value.untrusted.len > 0 and
+        signedBlock.message.body.blob_kzg_commitments.len > 0:
+      verify_data_column_sidecar_kzg_proofs(value.untrusted).isOkOr:
         debug "data column validation failed",
           blockRoot = shortLog(signedBlock.root),
           blck = shortLog(signedBlock.message),
@@ -240,17 +269,23 @@ proc verifySidecars(
 
 proc storeSidecars(
     self: BlockProcessor,
-    sidecarsOpt: Opt[fulu.DataColumnSidecars] | Opt[gloas.DataColumnSidecars]
+    sidecarsOpt: Opt[fulu.DataColumnSidecarsForImport] |
+                 Opt[gloas.DataColumnSidecars]
 ) =
   sidecarsOpt.isErrOr:
-    self.consensusManager.dag.db.putDataColumnSidecars(value)
-    if self.onDataColumnsStored != nil and value.len > 0:
+    let sidecars =
+      when value is gloas.DataColumnSidecars:
+        value
+      else:
+        value.items()
+    self.consensusManager.dag.db.putDataColumnSidecars(sidecars)
+    if self.onDataColumnsStored != nil and sidecars.len > 0:
       let slot =
         when value is gloas.DataColumnSidecars:
           # [Modified in Gloas:EIP7732] carries `slot` directly.
-          value[0][].slot
+          sidecars[0][].slot
         else:
-          value[0][].signed_block_header.message.slot
+          sidecars[0][].signed_block_header.message.slot
       self.onDataColumnsStored(slot)
 
 proc enqueuePayload*(self: ref BlockProcessor, blck: gloas.SignedBeaconBlock)
@@ -465,9 +500,9 @@ proc enqueueQuarantine(self: ref BlockProcessor, parent: BlockRef) =
       elif consensusFork == ConsensusFork.Fulu:
         let sidecarsOpt =
           if len(forkyBlck.message.body.blob_kzg_commitments) == 0:
-            Opt.some(default(fulu.DataColumnSidecars))
+            Opt.some(default(fulu.DataColumnSidecarsForImport))
           else:
-            self.fuluColumnQuarantine[].popSidecars(forkyBlck.root)
+            self.fuluColumnQuarantine[].popSidecarsForImport(forkyBlck.root)
       elif consensusFork in ConsensusFork.Phase0 .. ConsensusFork.Electra:
         const sidecarsOpt = noSidecars
       else:
@@ -630,7 +665,8 @@ proc enqueueFromDb(self: ref BlockProcessor, root: Eth2Digest) =
             sidecarsOk = false # Pruned, or inconsistent DB
             break
           data_column_sidecars.add data_column
-        Opt.some data_column_sidecars
+        # Columns reach the database only after their proofs were checked.
+        Opt.some data_column_sidecars.toTrustedImport()
       else:
         noSidecars
 
@@ -727,13 +763,7 @@ proc storeBlock(
   when consensusFork == ConsensusFork.Fulu:
     # Only request manager-sourced columns arrive unverified; getBlobsV2/V3/V4
     # and CL gossip are both either trusted or verified.
-    let pendingVerify =
-      self.fuluColumnQuarantine[].popPendingVerify(signedBlock.root)
-    if not pendingVerify.empty:
-      sidecarsOpt.isErrOr:
-        let toVerify = value.filterIt(it[].index in pendingVerify)
-        if toVerify.len > 0:
-          ?verifySidecars(signedBlock, Opt.some(toVerify))
+    ?verifySidecars(signedBlock, sidecarsOpt)
     debug "block_processor verifySidecars completed",
       verifySidecarsDur = Moment.now() - newPayloadTick,
       blck = shortLog(signedBlock.message),
@@ -933,10 +963,14 @@ proc addBlock*(
       # becomes canonical, it is vital to import it as quickly as possible.
       self.enqueueFromDb(blck.message.parent_root)
 
-      when sidecarsOpt is Opt[fulu.DataColumnSidecars]:
+      when sidecarsOpt is Opt[fulu.DataColumnSidecarsForImport]:
         if sidecarsOpt.isSome:
+          # Put them back as they came out, so trusted ones stay trusted.
+          let sidecars = sidecarsOpt.get
+          self.fuluColumnQuarantine[].put(
+            blockRoot, sidecars.untrusted, verified = false)
           self.fuluColumnQuarantine[].put(
-            blockRoot, sidecarsOpt.get, verified = false)
+            blockRoot, sidecars.trusted.asSeq, verified = true)
       elif sidecarsOpt is Opt[gloas.DataColumnSidecars]:
         # In Gloas, block is enqueued with NoSidecar so we need not to care
         # about quarantine.
@@ -1117,7 +1151,7 @@ proc enqueuePayload*(self: ref BlockProcessor, blck: gloas.SignedBeaconBlock) =
           if bid.message.blob_kzg_commitments.len() == 0:
             Opt.some(default(gloas.DataColumnSidecars))
           else:
-            self.gloasColumnQuarantine[].popSidecars(blck.root)
+            self.gloasColumnQuarantine[].popSidecarsForImport(blck.root)
         if sidecarsOpt.isNone():
           let dag = self.consensusManager.dag
           # As sidecars are missing, put envelope back to quarantine.

beacon_chain/gossip_processing/eth2_processor.nim

@@ -311,9 +311,9 @@ proc processSignedBeaconBlock*(
   elif consensusFork == ConsensusFork.Fulu:
     let sidecarsOpt =
       if len(signedBlock.message.body.blob_kzg_commitments) == 0:
-        Opt.some(default(fulu.DataColumnSidecars))
+        Opt.some(default(fulu.DataColumnSidecarsForImport))
       else:
-        self.fuluColumnQuarantine[].popSidecars(signedBlock.root)
+        self.fuluColumnQuarantine[].popSidecarsForImport(signedBlock.root)
     if sidecarsOpt.isNone():
       self.blockProcessor[].startExecutionValidity(signedBlock, wallTime)
       discard self.quarantine[].addSidecarless(self.dag.finalizedHead.slot, signedBlock)
@@ -426,7 +426,7 @@ proc processDataColumnSidecar*(
     block_root, dataColumnSidecar, verified = true)
 
   if block_root in self.quarantine[].sidecarless:
-    let cres = self.fuluColumnQuarantine[].popSidecars(block_root)
+    let cres = self.fuluColumnQuarantine[].popSidecarsForImport(block_root)
     if cres.isSome():
       let blck = self.quarantine[].popSidecarless(block_root).expect("checked above")
       withBlck(blck):

beacon_chain/nimbus_beacon_node.nim

@@ -664,7 +664,7 @@ proc initFullNode(
             # Disable sidecars processing at block time.
             const sidecarsOpt = noSidecars
           else:
-            let sidecarsOpt = Opt.none(fulu.DataColumnSidecars)
+            let sidecarsOpt = Opt.none(fulu.DataColumnSidecarsForImport)
         elif consensusFork in ConsensusFork.Phase0 .. ConsensusFork.Electra:
           const sidecarsOpt = noSidecars
         else:
@@ -688,9 +688,9 @@ proc initFullNode(
         elif consensusFork == ConsensusFork.Fulu:
           let sidecarsOpt =
             if len(forkyBlck.message.body.blob_kzg_commitments) == 0:
-              Opt.some(default(fulu.DataColumnSidecars))
+              Opt.some(default(fulu.DataColumnSidecarsForImport))
             else:
-              fuluColumnQuarantine[].popSidecars(forkyBlck.root)
+              fuluColumnQuarantine[].popSidecarsForImport(forkyBlck.root)
           if sidecarsOpt.isNone():
             # We don't have all the columns for this block, so we have
             # to put it in columnless quarantine.
@@ -756,7 +756,7 @@ proc initFullNode(
               if bid.message.blob_kzg_commitments.len() == 0:
                 Opt.some(default(gloas.DataColumnSidecars))
               else:
-                gloasColumnQuarantine[].popSidecars(blockRoot)
+                gloasColumnQuarantine[].popSidecarsForImport(blockRoot)
             if sidecarsOpt.isNone():
               # As sidecars are missing, put envelope back to quarantine.
               discard quarantine[].addSidecarless(dag.finalizedHead.slot, blck)

beacon_chain/spec/datatypes/fulu.nim

@@ -85,6 +85,15 @@ type
 
   DataColumnSidecars* = seq[ref DataColumnSidecar]
 
+  # Column sidecars whose KZG proofs have already been checked.
+  TrustedDataColumnSidecars* = distinct DataColumnSidecars
+
+  DataColumnSidecarsForImport* = object
+    ## All columns of one block, split by whether their proofs still need
+    ## checking.
+    trusted*: TrustedDataColumnSidecars
+    untrusted*: DataColumnSidecars
+
   # https://github.com/ethereum/consensus-specs/blob/v1.5.0-alpha.10/specs/fulu/p2p-interface.md#datacolumnidentifier
   DataColumnIdentifier* = object
     block_root*: Eth2Digest
@@ -544,6 +553,50 @@ template asTrusted*(
        SigVerifiedSignedBeaconBlock): TrustedSignedBeaconBlock =
   isomorphicCast[TrustedSignedBeaconBlock](x)
 
+template asTrusted*(x: DataColumnSidecars): TrustedDataColumnSidecars =
+  ## Only for columns whose KZG proofs have actually been checked.
+  TrustedDataColumnSidecars(x)
+
+template asSeq*(x: TrustedDataColumnSidecars): DataColumnSidecars =
+  DataColumnSidecars(x)
+
+func len*(x: TrustedDataColumnSidecars): int {.borrow.}
+
+# `borrow` cannot match `seq`'s generic `T` return, so these convert explicitly
+func `[]`*(x: TrustedDataColumnSidecars, i: int): ref DataColumnSidecar =
+  DataColumnSidecars(x)[i]
+
+func add*(x: var TrustedDataColumnSidecars, v: ref DataColumnSidecar) =
+  DataColumnSidecars(x).add(v)
+
+func len*(sidecars: DataColumnSidecarsForImport): int =
+  len(sidecars.trusted) + len(sidecars.untrusted)
+
+func items*(sidecars: DataColumnSidecarsForImport): DataColumnSidecars =
+  ## Every column, in ascending index order.
+  var
+    res = newSeqOfCap[ref DataColumnSidecar](len(sidecars))
+    i, j = 0
+  while i < len(sidecars.trusted) and j < len(sidecars.untrusted):
+    if sidecars.trusted[i][].index <= sidecars.untrusted[j][].index:
+      res.add(sidecars.trusted[i])
+      inc i
+    else:
+      res.add(sidecars.untrusted[j])
+      inc j
+  while i < len(sidecars.trusted):
+    res.add(sidecars.trusted[i])
+    inc i
+  while j < len(sidecars.untrusted):
+    res.add(sidecars.untrusted[j])
+    inc j
+  res
+
+func toTrustedImport*(
+    sidecars: DataColumnSidecars): DataColumnSidecarsForImport =
+  ## Only for sidecars built locally or read back from our own database.
+  DataColumnSidecarsForImport(trusted: sidecars.asTrusted)
+
 const
   KZG_COMMITMENTS_GINDEX* = get_generalized_index(
     BeaconBlockBody, "blob_kzg_commitments")

beacon_chain/validators/message_router.nim

@@ -69,7 +69,7 @@ type
     fulu.DataColumnSidecars
 
   SomeOptSidecars =
-    NoSidecars | Opt[BlobSidecars] | Opt[fulu.DataColumnSidecars]
+    NoSidecars | Opt[BlobSidecars] | Opt[fulu.DataColumnSidecarsForImport]
 
 func isGoodForSending(validationResult: ValidationRes): bool =
   # When routing messages from REST, it's possible that these have already
@@ -175,7 +175,8 @@ proc publishSidecars(
     router: ref MessageRouter,
     _: fulu.SignedBeaconBlock,
     cols: fulu.DataColumnSidecars
-): Future[Opt[fulu.DataColumnSidecars]] {.async: (raises: [CancelledError]).} =
+): Future[Opt[fulu.DataColumnSidecarsForImport]] {.
+    async: (raises: [CancelledError]).} =
   var workers = newSeq[Future[SendResult]](len(cols))
 
   for i, dc in cols:
@@ -194,8 +195,10 @@ proc publishSidecars(
       notice "Data column sent",
         data_column = shortLog(cols[i][])
 
+  # Assembled locally, so they need no KZG check on the way back in.
   Opt.some(cols.filterIt(
-    it[].index in router[].processor.fuluColumnQuarantine[].custodyMap))
+    it[].index in router[].processor.fuluColumnQuarantine[].custodyMap
+  ).toTrustedImport())
 
 proc publishSidecars(
     router: ref MessageRouter,

tests/test_block_processor.nim

@@ -339,7 +339,7 @@ suite "Block processor" & preset():
       withState(state[]):
         let b0 = addTestEngineBlock(cfg, consensusFork, forkyState, cache)
         when consensusFork == ConsensusFork.Fulu:
-          let sidecarsOpt = Opt.none(fulu.DataColumnSidecars)
+          let sidecarsOpt = Opt.none(fulu.DataColumnSidecarsForImport)
         else:
           let sidecarsOpt = noSidecars
         check (await processor.addBlock(
@@ -427,7 +427,9 @@ suite "Block processor" & preset():
         let res = await processor.addBlock(
           MsgSource.gossip,
           engineBlock.blck,
-          Opt.some(dataColumnSidecars)
+          # Untrusted, so this also exercises the KZG check in `storeBlock`
+          Opt.some(fulu.DataColumnSidecarsForImport(
+            untrusted: dataColumnSidecars))
         )
 
         check:
@@ -460,7 +462,7 @@ suite "Block processor" & preset():
         let res = await processor.addBlock(
           MsgSource.gossip,
           engineBlock.blck,
-          Opt.none(fulu.DataColumnSidecars)
+          Opt.none(fulu.DataColumnSidecarsForImport)
         )
 
         check: