add: columns to Eth2Processor and BlockProcessor

beacon_chain/beacon_chain_file.nim

@@ -79,6 +79,8 @@ const
     int(ConsensusFork.Phase0) .. int(high(ConsensusFork))
   BlobForkCodeRange =
     MaxForksCount .. (MaxForksCount + int(high(ConsensusFork)) - int(ConsensusFork.Deneb))
+  DataColumnForkCodeRange =
+    MaxForksCount .. (MaxForksCount + int(high(ConsensusFork)) - int(ConsensusFork.Fulu))
 
 func getBlockForkCode(fork: ConsensusFork): uint64 =
   uint64(fork)
@@ -94,6 +96,13 @@ func getBlobForkCode(fork: ConsensusFork): uint64 =
   of ConsensusFork.Phase0 .. ConsensusFork.Capella:
     raiseAssert "Blobs are not supported for the fork"
 
+func getDataColumnForkCode(fork: ConsensusFork): uint64 =
+  case fork
+  of ConsensusFork.Fulu:
+    uint64(MaxForksCount)
+  of ConsensusFork.Phase0 .. ConsensusFork.Electra:
+    raiseAssert "Data columns are not supported for the fork"
+
 proc init(t: typedesc[ChainFileError], k: ChainFileErrorType,
               m: string): ChainFileError =
   ChainFileError(kind: k, message: m)
@@ -134,7 +143,8 @@ proc checkKind(kind: uint64): Result[void, string] =
       if res > uint64(high(int)):
         return err("Unsuppoted chunk kind value")
       int(res)
-  if (hkind in BlockForkCodeRange) or (hkind in BlobForkCodeRange):
+  if (hkind in BlockForkCodeRange) or (hkind in BlobForkCodeRange) or
+     (hkind in DataColumnForkCodeRange):
     ok()
   else:
     err("Unsuppoted chunk kind value")
@@ -260,6 +270,12 @@ template getBlobChunkKind(kind: ConsensusFork, last: bool): uint64 =
   else:
     getBlobForkCode(kind)
 
+template getDataColumnChunkKind(kind: ConsensusFork,last: bool): uint64 =
+  if last:
+    maskKind(getDataColumnForkCode(kind))
+  else:
+    getDataColumnForkCode(kind)
+
 proc getBlockConsensusFork(header: ChainFileHeader): ConsensusFork =
   let hkind = unmaskKind(header.kind)
   if int(hkind) in BlockForkCodeRange:
@@ -275,6 +291,10 @@ template isBlob(h: ChainFileHeader | ChainFileFooter): bool =
   let hkind = unmaskKind(h.kind)
   int(hkind) in BlobForkCodeRange
 
+template isDataColumn(h: ChainFileHeader | ChainFileFooter): bool =
+  let hkind = unmaskKind(h.kind)
+  int(hkind) in DataColumnForkCodeRange
+
 template isLast(h: ChainFileHeader | ChainFileFooter): bool =
   h.kind.isLast()
 
@@ -291,7 +311,8 @@ proc setTail*(chandle: var ChainFileHandle, bdata: BlockData) =
   chandle.data.tail = Opt.some(bdata)
 
 proc store*(chandle: ChainFileHandle, signedBlock: ForkedSignedBeaconBlock,
-            blobs: Opt[BlobSidecars]): Result[void, string] =
+            blobs: Opt[BlobSidecars], dataColumns: Opt[DataColumnSidecars]):
+            Result[void, string] =
   let origOffset =
     updateFilePos(chandle.handle, 0'i64, SeekPosition.SeekEnd).valueOr:
       return err(ioErrorMsg(error))
@@ -342,6 +363,36 @@ proc store*(chandle: ChainFileHandle, signedBlock: ForkedSignedBeaconBlock,
         discard fsync(chandle.handle)
         return err(IncompleteWriteError)
 
+  if dataColumns.isSome():
+    let dataColumnSidecars =
+      dataColumns.get
+    for index, dataColumn in dataColumnSidecars.pairs():
+      let
+        kind =
+          getDataColumnChunkKind(signedBlock.kind, (index + 1) ==
+            len(dataColumnSidecars))
+        (data, plainSize) =
+          block:
+            let res = SSZ.encode(dataColumn[])
+            (snappy.encode(res), len(res))
+        slot = dataColumn[].signed_block_header.message.slot
+        buffer = Chunk.init(kind, uint64(slot), uint32(plainSize), data)
+
+      setFilePos(chandle.handle, 0'i64, SeekPosition.SeekEnd).isOkOr:
+        discard truncate(chandle.handle, origOffset)
+        discard fsync(chandle.handle)
+        return err(ioErrorMsg(error))
+
+      let
+        wrote = writeFile(chandle.handle, buffer).valueOr:
+          discard truncate(chandle.handle, origOffset)
+          discard fsync(chandle.handle)
+          return err(ioErrorMsg(error))
+      if wrote != uint(len(buffer)):
+        discard truncate(chandle.handle, origOffset)
+        discard fsync(chandle.handle)
+        return err(IncompleteWriteError)
+
   fsync(chandle.handle).isOkOr:
     discard truncate(chandle.handle, origOffset)
     return err(ioErrorMsg(error))
@@ -550,6 +601,22 @@ proc decodeBlob(
         return err("Incorrect blob format")
   ok(blob)
 
+proc decodeDataColumn(
+    header: ChainFileHeader,
+    data: openArray[byte],
+): Result[DataColumnSidecar, string] =
+  if header.plainSize > uint32(MaxChunkSize):
+    return err("Size of data column is enormously big")
+
+  let
+    decompressed = snappy.decode(data, uint32(header.plainSize))
+    dataColumn =
+      try:
+        SSZ.decode(decompressed, DataColumnSidecar)
+      except SerializationError:
+        return err("Incorrect data column format")
+  ok(dataColumn)
+
 proc getChainFileTail*(handle: IoHandle): Result[Opt[BlockData], string] =
   var sidecars: BlobSidecars
   while true:

beacon_chain/consensus_object_pools/block_pools_types.nim

@@ -288,6 +288,7 @@ type
   BlockData* = object
     blck*: ForkedSignedBeaconBlock
     blob*: Opt[BlobSidecars]
+    dataColumn*: Opt[DataColumnSidecars]
 
   OnBlockAdded*[T: ForkyTrustedSignedBeaconBlock] = proc(
     blckRef: BlockRef, blck: T, epochRef: EpochRef,

beacon_chain/consensus_object_pools/blockchain_list.nim

@@ -9,6 +9,7 @@
 
 import std/sequtils, stew/io2, chronicles, chronos, metrics,
        ../spec/forks,
+       ../spec/peerdas_helpers,
        ../[beacon_chain_file, beacon_clock],
        ../sszdump
 
@@ -128,16 +129,17 @@ proc setTail*(clist: ChainListRef, bdata: BlockData) =
   clist.handle = Opt.some(handle)
 
 proc store*(clist: ChainListRef, signedBlock: ForkedSignedBeaconBlock,
-            blobs: Opt[BlobSidecars]): Result[void, string] =
+            blobs: Opt[BlobSidecars], dataColumns: Opt[DataColumnSidecars]):
+            Result[void, string] =
   if clist.handle.isNone():
     let
       filename = clist.path.chainFilePath()
       flags = {ChainFileFlag.Repair, ChainFileFlag.OpenAlways}
       handle = ? ChainFileHandle.init(filename, flags)
     clist.handle = Opt.some(handle)
-    store(handle, signedBlock, blobs)
+    store(handle, signedBlock, blobs, dataColumns)
   else:
-    store(clist.handle.get(), signedBlock, blobs)
+    store(clist.handle.get(), signedBlock, blobs, dataColumns)
 
 proc checkBlobs(signedBlock: ForkedSignedBeaconBlock,
                 blobsOpt: Opt[BlobSidecars]): Result[void, VerifierError] =
@@ -167,9 +169,31 @@ proc checkBlobs(signedBlock: ForkedSignedBeaconBlock,
             return err(VerifierError.Invalid)
   ok()
 
+proc checkDataColumns*(signedBlock: ForkedSignedBeaconBlock,
+                       dataColumnsOpt: Opt[DataColumnSidecars]):
+                       Result[void, VerifierError] =
+  withBlck(signedBlock):
+    when consensusFork >= ConsensusFork.Fulu:
+      if dataColumnsOpt.isSome:
+        let dataColumns = dataColumnsOpt.get()
+        if dataColumns.len > 0:
+          for i in 0..<dataColumns.len:
+            let r =
+              verify_data_column_sidecar_kzg_proofs(dataColumns[i][])
+            if r.isErr:
+              debug "Data column validation failed",
+                    blockRoot = shortLog(forkyBlck.root),
+                    dataColumn = shortLog(dataColumns[i][]),
+                    blck = shortLog(forkyBlck.message),
+                    signature = shortLog(forkyBlck.signature),
+                    msg = r.error()
+              return err(VerifierError.Invalid)
+
+
 proc addBackfillBlockData*(
     clist: ChainListRef, signedBlock: ForkedSignedBeaconBlock,
-    blobsOpt: Opt[BlobSidecars]): Result[void, VerifierError] =
+    blobsOpt: Opt[BlobSidecars], dataColumnsOpt: Opt[DataColumnSidecars]):
+    Result[void, VerifierError] =
   doAssert(not(isNil(clist)))
 
   logScope:
@@ -182,15 +206,17 @@ proc addBackfillBlockData*(
 
   if clist.tail.isNone():
     ? checkBlobs(signedBlock, blobsOpt)
+    ? checkDataColumns(signedBlock, dataColumnsOpt)
 
     let storeBlockTick = Moment.now()
 
-    store(clist, signedBlock, blobsOpt).isOkOr:
+    store(clist, signedBlock, blobsOpt, dataColumnsOpt).isOkOr:
       fatal "Unexpected failure while trying to store data",
             filename = chainFilePath(clist.path), reason = error
       quit 1
 
-    let bdata = BlockData(blck: signedBlock, blob: blobsOpt)
+    let bdata = BlockData(blck: signedBlock, blob: blobsOpt,
+                          dataColumn: dataColumnsOpt)
     clist.setTail(bdata)
     if clist.head.isNone():
       clist.setHead(bdata)
@@ -219,10 +245,11 @@ proc addBackfillBlockData*(
     return err(VerifierError.MissingParent)
 
   ? checkBlobs(signedBlock, blobsOpt)
+  ? checkDataColumns(signedBlock, dataColumnsOpt)
 
   let storeBlockTick = Moment.now()
 
-  store(clist, signedBlock, blobsOpt).isOkOr:
+  store(clist, signedBlock, blobsOpt, dataColumnsOpt).isOkOr:
     fatal "Unexpected failure while trying to store data",
            filename = chainFilePath(clist.path), reason = error
     quit 1
@@ -231,17 +258,19 @@ proc addBackfillBlockData*(
         verify_block_duration = shortLog(storeBlockTick - verifyBlockTick),
         store_block_duration = shortLog(Moment.now() - storeBlockTick)
 
-  clist.setTail(BlockData(blck: signedBlock, blob: blobsOpt))
+  clist.setTail(BlockData(blck: signedBlock, blob: blobsOpt, dataColumn: dataColumnsOpt))
 
   ok()
 
 proc untrustedBackfillVerifier*(
     clist: ChainListRef,
     signedBlock: ForkedSignedBeaconBlock,
     blobs: Opt[BlobSidecars],
+    dataColumns: Opt[DataColumnSidecars],
     maybeFinalized: bool
 ): Future[Result[void, VerifierError]] {.
   async: (raises: [CancelledError], raw: true).} =
   let retFuture = newFuture[Result[void, VerifierError]]()
-  retFuture.complete(clist.addBackfillBlockData(signedBlock, blobs))
+  retFuture.complete(clist.addBackfillBlockData(signedBlock, blobs,
+                                                dataColumns))
   retFuture

beacon_chain/consensus_object_pools/data_column_quarantine.nim

@@ -139,13 +139,12 @@ func hasMissingDataColumns*(quarantine: DataColumnQuarantine,
         index: idx)
     if dc_identifier notin quarantine.data_columns:
       inc col_counter
-  if quarantine.supernode and col_counter != NUMBER_OF_COLUMNS:
-    return false
-  elif quarantine.supernode == false and
-      col_counter != max(SAMPLES_PER_SLOT, CUSTODY_REQUIREMENT):
-    return false
-  else:
+  if quarantine.supernode and col_counter == NUMBER_OF_COLUMNS:
+    return true
+  if quarantine.supernode == false and
+      col_counter == max(SAMPLES_PER_SLOT, CUSTODY_REQUIREMENT):
     return true
+  false
 
 func hasEnoughDataColumns*(quarantine: DataColumnQuarantine,
     blck: fulu.SignedBeaconBlock): bool =