serve finalized blobs within the blob prune window

ChainSafe · Mar 24, 2023 · 05539b7 · 05539b7
1 parent b5b0822
commit 05539b7
Show file tree

Hide file tree

Showing 3 changed files with 53 additions and 27 deletions.
diff --git a/packages/beacon-node/src/db/repositories/blobSidecarsArchive.ts b/packages/beacon-node/src/db/repositories/blobSidecarsArchive.ts
@@ -13,7 +13,7 @@ export class BlobSidecarsArchiveRepository extends Repository<Slot, deneb.BlobSi
     super(config, db, Bucket.allForks_blobSidecarsArchive, ssz.deneb.BlobSidecarsWrapper);
   }
 
-    // Handle key as slot
+  // Handle key as slot
 
   getId(value: deneb.BlobSidecarsWrapper): Slot {
     return value.slot;

diff --git a/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRange.ts b/packages/beacon-node/src/network/reqresp/handlers/blobSidecarsByRange.ts
@@ -1,6 +1,6 @@
 import {GENESIS_SLOT, MAX_REQUEST_BLOB_SIDECARS} from "@lodestar/params";
 import {ContextBytesType, EncodedPayloadBytes, EncodedPayloadType, ResponseError, RespStatus} from "@lodestar/reqresp";
-import {deneb} from "@lodestar/types";
+import {deneb, Slot} from "@lodestar/types";
 import {fromHex} from "@lodestar/utils";
 import {IBeaconChain} from "../../../chain/index.js";
 import {IBeaconDb} from "../../../db/index.js";
@@ -16,6 +16,18 @@ export async function* onBlobSidecarsByRange(
   const endSlot = startSlot + count;
   const finalizedSlot = chain.forkChoice.getFinalizedBlock().slot;
 
+  // Finalized range of blobs
+
+  if (startSlot <= finalizedSlot) {
+    // Chain of blobs won't change
+    for await (const {key, value: blobSideCarsBytesWrapped} of db.blobSidecarsArchive.binaryEntriesStream({
+      gte: startSlot,
+      lt: endSlot,
+    })) {
+      yield* iterateBlobBytesFromWrapper(blobSideCarsBytesWrapped, db.blobSidecarsArchive.decodeKey(key));
+    }
+  }
+
   if (endSlot > finalizedSlot) {
     const headRoot = chain.forkChoice.getHeadRoot();
     // TODO DENEB: forkChoice should mantain an array of canonical blocks, and change only on reorg
@@ -37,30 +49,7 @@ export async function* onBlobSidecarsByRange(
           // Handle the same to onBeaconBlocksByRange
           throw new ResponseError(RespStatus.SERVER_ERROR, `No item for root ${block.blockRoot} slot ${block.slot}`);
         }
-
-        const blobSideCarsBytes = blobSideCarsBytesWrapped.slice(BLOB_SIDECARS_IN_WRAPPER_INDEX);
-        const blobsLen = blobSideCarsBytes.length / BLOBSIDECAR_FIXED_SIZE;
-
-        for (let index = 0; index < blobsLen; index++) {
-          const blobSideCarBytes = blobSideCarsBytes.slice(
-            index * BLOBSIDECAR_FIXED_SIZE,
-            (index + 1) * BLOBSIDECAR_FIXED_SIZE
-          );
-          if (blobSideCarBytes.length !== BLOBSIDECAR_FIXED_SIZE) {
-            throw new ResponseError(
-              RespStatus.SERVER_ERROR,
-              `Invalid blobSidecar index=${index} bytes length=${blobSideCarBytes.length} expected=${BLOBSIDECAR_FIXED_SIZE} for ${block.blockRoot} slot ${block.slot} blobsLen=${blobsLen}`
-            );
-          }
-          yield {
-            type: EncodedPayloadType.bytes,
-            bytes: blobSideCarBytes,
-            contextBytes: {
-              type: ContextBytesType.ForkDigest,
-              forkSlot: block.slot,
-            },
-          };
-        }
+        yield* iterateBlobBytesFromWrapper(blobSideCarsBytesWrapped, block.slot);
       }
 
       // If block is after endSlot, stop iterating
@@ -71,6 +60,35 @@ export async function* onBlobSidecarsByRange(
   }
 }
 
+export function* iterateBlobBytesFromWrapper(
+  blobSideCarsBytesWrapped: Uint8Array,
+  blockSlot: Slot
+): Iterable<EncodedPayloadBytes> {
+  const blobSideCarsBytes = blobSideCarsBytesWrapped.slice(BLOB_SIDECARS_IN_WRAPPER_INDEX);
+  const blobsLen = blobSideCarsBytes.length / BLOBSIDECAR_FIXED_SIZE;
+
+  for (let index = 0; index < blobsLen; index++) {
+    const blobSideCarBytes = blobSideCarsBytes.slice(
+      index * BLOBSIDECAR_FIXED_SIZE,
+      (index + 1) * BLOBSIDECAR_FIXED_SIZE
+    );
+    if (blobSideCarBytes.length !== BLOBSIDECAR_FIXED_SIZE) {
+      throw new ResponseError(
+        RespStatus.SERVER_ERROR,
+        `Invalid blobSidecar index=${index} bytes length=${blobSideCarBytes.length} expected=${BLOBSIDECAR_FIXED_SIZE} for slot ${blockSlot} blobsLen=${blobsLen}`
+      );
+    }
+    yield {
+      type: EncodedPayloadType.bytes,
+      bytes: blobSideCarBytes,
+      contextBytes: {
+        type: ContextBytesType.ForkDigest,
+        forkSlot: blockSlot,
+      },
+    };
+  }
+}
+
 export function validateBlobSidecarsByRangeRequest(
   request: deneb.BlobSidecarsByRangeRequest
 ): deneb.BlobSidecarsByRangeRequest {

diff --git a/packages/beacon-node/test/unit/chain/archive/blockArchiver.test.ts b/packages/beacon-node/test/unit/chain/archive/blockArchiver.test.ts
@@ -36,7 +36,15 @@ describe("block archiver task", function () {
     const nonCanonicalBlocks = [blocks[2]];
     forkChoiceStub.getAllAncestorBlocks.returns(canonicalBlocks);
     forkChoiceStub.getAllNonAncestorBlocks.returns(nonCanonicalBlocks);
-    await archiveBlocks(config, dbStub, forkChoiceStub, lightclientServer, logger, {epoch: 5, rootHex: ZERO_HASH_HEX});
+    await archiveBlocks(
+      config,
+      dbStub,
+      forkChoiceStub,
+      lightclientServer,
+      logger,
+      {epoch: 5, rootHex: ZERO_HASH_HEX},
+      currentEpoch
+    );
 
     expect(dbStub.blockArchive.batchPutBinary.getCall(0).args[0]).to.deep.equal(
       canonicalBlocks.map((summary) => ({