fix: peerDAS - fix mismatch error when downloading columns at synced time (#7467)

**Motivation**

- fix this mismatch error at synced time
```
^[[39m: matchBlockWithDataColumns2 dataColumnIndexes=11 15, reque        stedColumnsPresent=false, slot=38584, peerClient=Lighthouse-cgc:128
2609158 Feb-12 18:58:44.857[sync]            ^[[34mdebug^[[39m: Missing or mismatching dataColumnSidecars from peerId=16U        iu2HAmJyaVGkRGR9ACqompkoge8T2x4KFH4KbzDkh7zz6uN2JX for blockSlot=38584 with numColumns=8 dataColumnSidecars=2 req        uestedColumnsPresent=false received dataColumnIndexes=11 15 requested=11 15 65 71 72 73 83 108 allBlocks=1, allDa        taColumnSidecars=2, peerId=16Uiu2HAmJyaVGkRGR9ACqompkoge8T2x4KFH4KbzDkh7zz6uN2JX, nodeId=0x70a752f736fd0254c715ae        31a3b64d9f3e011c1ca38e7f828d6d7f7bdf4391aa, blobKzgCommitmentsLen=6, peerClient=Lighthouse-cgc:128
```
- this is the logic inside `unavailableBeaconBlobsByRootPostFulu()`


**Description**

- at gossip time, `unavailableBeaconBlobsByRootPostFulu()` call range
sync's `matchBlockWithDataColumns()` which is a mistake because we
should not expect peer has enough columns for us at gossip time. Spec:
https://github.com/ethereum/consensus-specs/blob/dev/specs/fulu/p2p-interface.md#datacolumnsidecarsbyroot-v1
- instead of that, just get as much columns as we can and return data
promise at that time
- more details: bring `unavailableBeaconBlobsByRootPostFulu()` logic
without matching code

Co-authored-by: Tuyen Nguyen <[email protected]>

Author: twoeths

packages/beacon-node/src/network/reqresp/beaconBlocksMaybeBlobsByRoot.ts

@@ -19,6 +19,7 @@ import {
   NullBlockInput,
   getBlockInput,
   getBlockInputBlobs,
+  getBlockInputDataColumns,
 } from "../../chain/blocks/types.js";
 import {BlockInputAvailabilitySource} from "../../chain/seenCache/seenGossipBlockInput.js";
 import {IExecutionEngine} from "../../execution/index.js";
@@ -150,6 +151,9 @@ export async function beaconBlocksMaybeBlobsByRoot(
 
     // The last arg is to provide slot to which all blobs should be exausted in matching
     // and here it should be infinity since all bobs should match
+    // TODO: should not call matchBlockWithDataColumns() because it's supposed for range sync
+    // in that function, peers should return all requested data columns, this function runs at gossip time
+    // and it should not expect that
     const blockInputWithBlobs = matchBlockWithDataColumns(
       network,
       peerId,
@@ -434,12 +438,20 @@ export async function unavailableBeaconBlobsByRootPreFulu(
   return getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
 }
 
+
+/**
+ * Download more columns for a BlockInput
+ * - unavailableBlockInput should have block, but not enough blobs (deneb) or data columns (fulu)
+ *
+ * This function may return data promise, and consumer should continue with fetching more blobs or columns from other peers
+ * see UnknownBlockSync.fetchUnavailableBlockInput()
+ */
 export async function unavailableBeaconBlobsByRootPostFulu(
   config: ChainForkConfig,
   network: INetwork,
   peerId: PeerIdStr,
   peerClient: string,
-  unavailableBlockInput: BlockInput | NullBlockInput,
+  unavailableBlockInput: BlockInput,
   block: SignedBeaconBlock,
   cachedData: NullBlockInput["cachedData"],
   opts: {
@@ -451,7 +463,6 @@ export async function unavailableBeaconBlobsByRootPostFulu(
     return unavailableBlockInput;
   }
 
-  let availableBlockInput: BlockInput;
   if (cachedData.fork === ForkName.deneb || cachedData.fork === ForkName.electra) {
     const {blobsCache, resolveAvailability} = cachedData;
 
@@ -489,113 +500,108 @@ export async function unavailableBeaconBlobsByRootPostFulu(
     const blockData = {fork: cachedData.fork, ...allBlobs, blobsSource: BlobsSource.byRoot} as BlockInputBlobs;
     resolveAvailability(blockData);
     opts.metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.UNKNOWN_SYNC});
-    availableBlockInput = getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
-  } else if (cachedData.fork === ForkName.fulu) {
-    const {dataColumnsCache, resolveAvailability} = cachedData as CachedDataColumns;
+    return getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
+  }
 
-    // resolve missing blobs
-    const dataColumnIdentifiers: fulu.DataColumnIdentifier[] = [];
-    const slot = block.message.slot;
-    const blockRoot = config.getForkTypes(slot).BeaconBlock.hashTreeRoot(block.message);
+  // fulu fork
+  const {dataColumnsCache, resolveAvailability} = cachedData as CachedDataColumns;
 
-    const blobKzgCommitmentsLen = (block.message.body as deneb.BeaconBlockBody).blobKzgCommitments.length;
-    if (blobKzgCommitmentsLen === 0) {
-      const blockData = {
-        fork: cachedData.fork,
-        dataColumns: [],
-        dataColumnsBytes: [],
-        dataColumnsSource: DataColumnsSource.gossip,
-      } as BlockInputDataColumns;
+  // resolve missing blobs
+  const dataColumnIdentifiers: fulu.DataColumnIdentifier[] = [];
+  const slot = block.message.slot;
+  const blockRoot = config.getForkTypes(slot).BeaconBlock.hashTreeRoot(block.message);
 
-      resolveAvailability(blockData);
-      opts.metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.UNKNOWN_SYNC});
-      availableBlockInput = getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
-    } else {
-      const {custodyConfig} = network;
-      const neededColumns = custodyConfig.sampledColumns.reduce((acc, elem) => {
-        if (dataColumnsCache.get(elem) === undefined) {
-          acc.push(elem);
-        }
-        return acc;
-      }, [] as number[]);
+  const blobKzgCommitmentsLen = (block.message.body as deneb.BeaconBlockBody).blobKzgCommitments.length;
+  if (blobKzgCommitmentsLen === 0) {
+    const blockData = {
+      fork: cachedData.fork,
+      dataColumns: [],
+      dataColumnsBytes: [],
+      dataColumnsSource: DataColumnsSource.gossip,
+    } as BlockInputDataColumns;
 
-      const peerColumns = network.getConnectedPeerCustody(peerId);
+    resolveAvailability(blockData);
+    opts.metrics?.syncUnknownBlock.resolveAvailabilitySource.inc({source: BlockInputAvailabilitySource.UNKNOWN_SYNC});
+    return getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
+  } else {
+    const {custodyConfig} = network;
+    let neededColumns = custodyConfig.sampledColumns.reduce((acc, elem) => {
+      if (dataColumnsCache.get(elem) === undefined) {
+        acc.push(elem);
+      }
+      return acc;
+    }, [] as number[]);
 
-      // get match
-      const columns = peerColumns.reduce((acc, elem) => {
-        if (neededColumns.includes(elem)) {
-          acc.push(elem);
-        }
-        return acc;
-      }, [] as number[]);
+    const peerColumns = network.getConnectedPeerCustody(peerId);
 
-      // this peer can't help fetching columns for this block
-      if (unavailableBlockInput.block !== null && columns.length === 0 && neededColumns.length > 0) {
-        return unavailableBlockInput;
+    // get match
+    const columns = peerColumns.reduce((acc, elem) => {
+      if (neededColumns.includes(elem)) {
+        acc.push(elem);
       }
+      return acc;
+    }, [] as number[]);
 
-      for (const columnIndex of columns) {
-        dataColumnIdentifiers.push({blockRoot, index: columnIndex});
-      }
+    // this peer can't help fetching columns for this block
+    if (unavailableBlockInput.block !== null && columns.length === 0 && neededColumns.length > 0) {
+      return unavailableBlockInput;
+    }
 
-      // console.log("unavailableBlockInput fetching", {
-      //   neededColumns: neededColumns.length,
-      //   peerColumns: peerColumns.length,
-      //   intersectingColumns: columns.length,
-      //   dataColumnIdentifiers: dataColumnIdentifiers.length,
-      //   cacheId,
-      //   dataColumnsCache: dataColumnsCache.size,
-      //   blockRoot: toHexString(blockRoot),
-      // });
-
-      let allDataColumnSidecars: fulu.DataColumnSidecar[];
-      if (dataColumnIdentifiers.length > 0) {
-        allDataColumnSidecars = await network.sendDataColumnSidecarsByRoot(peerId, dataColumnIdentifiers);
-      } else {
-        allDataColumnSidecars = [];
+    for (const columnIndex of columns) {
+      dataColumnIdentifiers.push({blockRoot, index: columnIndex});
+    }
+
+    let allDataColumnSidecars: fulu.DataColumnSidecar[];
+    if (dataColumnIdentifiers.length > 0) {
+      allDataColumnSidecars = await network.sendDataColumnSidecarsByRoot(peerId, dataColumnIdentifiers);
+    } else {
+      allDataColumnSidecars = [];
+    }
+
+    const logCtx = {
+      slot: block.message.slot,
+      requestedColumns: columns.join(","),
+      respondedColumns: allDataColumnSidecars.map((dcs) => dcs.index).join(","),
+      peerClient,
+    };
+
+    // the same to matchBlockWithDataColumns() without expecting requested data columns = responded data columns
+    // because at gossip time peer may not have enough column to return
+    for (const dataColumnSidecar of allDataColumnSidecars) {
+      dataColumnsCache.set(dataColumnSidecar.index, {
+        dataColumn: dataColumnSidecar,
+        // TODO: req/resp should return bytes here
+        dataColumnBytes: null,
+      });
+    }
+
+    // reevaluate needeColumns and resolve availability if possible
+    neededColumns = custodyConfig.sampledColumns.reduce((acc, elem) => {
+      if (dataColumnsCache.get(elem) === undefined) {
+        acc.push(elem);
       }
+      return acc;
+    }, [] as number[]);
 
-      // console.log("unavailableBlockInput fetched", {
-      //   neededColumns: neededColumns.length,
-      //   peerColumns: peerColumns.length,
-      //   intersectingColumns: columns.length,
-      //   dataColumnIdentifiers: dataColumnIdentifiers.length,
-      //   allDataColumnSidecars: allDataColumnSidecars.length,
-      //   cacheId,
-      //   dataColumnsCache: dataColumnsCache.size,
-      //   blockRoot: toHexString(blockRoot),
-      // });
-
-      [availableBlockInput] = matchBlockWithDataColumns(
-        network,
-        peerId,
-        config,
-        custodyConfig,
-        columns,
-        [{data: block}],
-        allDataColumnSidecars,
-        block.message.slot,
-        BlockSource.byRoot,
-        DataColumnsSource.byRoot,
-        unavailableBlockInput.block !== null
-          ? {blocks: [unavailableBlockInput], pendingDataColumns: neededColumns}
-          : null,
-        peerClient,
-        opts.logger
+    if (neededColumns.length === 0) {
+      const {dataColumns, dataColumnsBytes} = getBlockInputDataColumns(
+        (cachedData as CachedDataColumns).dataColumnsCache,
+        custodyConfig.sampledColumns
       );
 
       // don't forget to resolve availability as the block may be stuck in availability wait
-      if (availableBlockInput !== undefined && availableBlockInput.type === BlockInputType.availableData) {
-        const {blockData} = availableBlockInput;
-        if (blockData.fork !== ForkName.fulu) {
-          throw Error(`unexpected blockData fork=${blockData.fork} returned by matchBlockWithDataColumns`);
-        }
-        resolveAvailability(blockData as BlockInputDataColumns);
-      }
+      const blockData = {
+        fork: config.getForkName(block.message.slot),
+        dataColumns,
+        dataColumnsBytes,
+        dataColumnsSource: DataColumnsSource.byRoot,
+      } as BlockInputDataColumns;
+      resolveAvailability(blockData);
+      opts.logger?.verbose("unavailableBeaconBlobsByRootPostFulu: Resolved availability for block with all data columns", logCtx);
+      return getBlockInput.availableData(config, block, BlockSource.byRoot, blockData);
+    } else {
+      opts.logger?.verbose("unavailableBeaconBlobsByRootPostFulu: Still missing data columns for block", logCtx);
+      return getBlockInput.dataPromise(config, block, BlockSource.byRoot, cachedData);
     }
-  } else {
-    throw Error("Invalid cachedData for unavailableBeaconBlobsByRoot");
   }
-
-  return availableBlockInput;
 }