add: data column reconstruction logic

beacon_chain/gossip_processing/block_processor.nim

@@ -35,6 +35,7 @@ from ../consensus_object_pools/data_column_quarantine import
 from ../validators/validator_monitor import
   MsgSource, ValidatorMonitor, registerAttestationInBlock, registerBeaconBlock,
   registerSyncAggregateInBlock
+from ../networking/eth2_network import Eth2Node
 from ../beacon_chain_db import getBlobSidecar, putBlobSidecar,
   getDataColumnSidecar, putDataColumnSidecar
 from ../spec/state_transition_block import validate_blobs
@@ -439,6 +440,64 @@ proc enqueueBlock*(
   except AsyncQueueFullError:
     raiseAssert "unbounded queue"
 
+proc reconstructDataColumns(
+    self: ref BlockProcessor,
+    node: Eth2Node,
+    signed_block: deneb.SignedBeaconBlock |
+    electra.SignedBeaconBlock,
+    data_column: DataColumnSidecar):
+    Result[bool, cstring] =
+  
+  let
+    dag = self.consensusManager.dag
+    root = signed_block.root
+    custodiedColumnIndices = get_custody_columns(
+        node.nodeId,
+        CUSTODY_REQUIREMENT)
+  
+  var
+    data_column_sidecars: DataColumnSidecars
+    columnsOk = true
+    storedColumns = 0
+  
+  # Loading the data columns from the database
+  for i in 0 ..< custodiedColumnIndices.len:
+    let data_column = DataColumnSidecar.new()
+    if not dag.db.getDataColumnSidecar(root, custodiedColumnIndices[i], data_column[]):
+      columnsOk = false
+      break
+    data_column_sidecars.add data_column
+    storedColumns.add data_column.index
+
+    if columnsOk:
+      debug "Loaded data column for reconstruction"
+
+  # storedColumn number is less than the NUMBER_OF_COLUMNS
+  # then reconstruction is not possible, and if all the data columns
+  # are already stored then we do not need to reconstruct at all
+  if storedColumns.len < NUMBER_OF_COLUMNS or storedColumns.len == NUMBER_OF_COLUMNS:
+    return ok(false)
+  else:
+    return err ("DataColumnSidecar: Reconstruction error!")
+
+  # Recover blobs from saved data column sidecars
+  let recovered_blobs = recover_blobs(data_column_sidecars, storedColumns.len, signed_block)
+  if not recovered_blobs.isOk:
+    return err ("Error recovering blobs from data columns")
+
+  # Reconstruct data column sidecars from recovered blobs
+  let reconstructedDataColumns = get_data_column_sidecars(signed_block, recovered_blobs.get)
+  if not reconstructedDataColumns.isOk:
+    return err ("Error reconstructing data columns from recovered blobs")
+
+  for data_column in data_column_sidecars:
+    if data_column.index notin custodiedColumnIndices:
+      continue
+    
+    dag.db.putDataColumnSidecar(data_column[])
+
+  ok(true)
+
 proc storeBlock(
     self: ref BlockProcessor, src: MsgSource, wallTime: BeaconTime,
     signedBlock: ForkySignedBeaconBlock,

beacon_chain/spec/datatypes/eip7594.nim

@@ -7,7 +7,9 @@
 
 {.push raises: [].}
 
-import "."/[base, deneb], kzg4844
+import 
+  "."/[base, deneb], 
+  kzg4844
 
 from std/sequtils import mapIt
 from std/strutils import join
@@ -71,7 +73,8 @@ type
     kzg_proof*: KzgProof
     column_index*: ColumnIndex
     row_index*: RowIndex
-    
+
+
 func shortLog*(v: DataColumnSidecar): auto =
   (
     index: v.index,

beacon_chain/spec/eip7594_helpers.nim

@@ -22,6 +22,7 @@ import
   
   ./datatypes/[eip7594, deneb]
 
+
 proc sortedColumnIndices*(columnsPerSubnet: ColumnIndex, subnetIds: HashSet[uint64]): seq[ColumnIndex] =
   var res: seq[ColumnIndex] = @[]
   for i in 0 ..< columnsPerSubnet: