feat(blobs): blob sink (#10079)

fixes: #10053

yarn-project/blob-sink/.eslintrc.cjs

@@ -0,0 +1 @@
+module.exports = require('@aztec/foundation/eslint');

yarn-project/blob-sink/README.md

@@ -0,0 +1,21 @@
+## Blob Sink
+
+A HTTP api that losely emulates the https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getBlobSidecars API.
+We do not support all of the possible values of block_id, namely `genesis`, `head`, `finalized`. As we are not using any of these values in our
+blobs integration.
+
+## When is this used?
+
+This service will run alongside end to end tests to capture the blob transactions that are sent alongside a `propose` transaction.
+
+### Why?
+
+Once we make the transition to blob transactions, we will need to be able to query for blobs. One way to do this is to run an entire L1 execution layer and consensus layer pair alongside all of our e2e tests and inside the sandbox. But this is a bit much, so instead the blob sink can be used to store and request blobs, without needing to run an entire consensus layer pair client.
+
+### Other Usecases
+
+Blobs are only held in the L1 consensus layer for a period of ~3 weeks, the blob sink can be used to store blobs for longer.
+
+### How?
+
+The blob sink is a simple HTTP server that can be run alongside the e2e tests. It will store the blobs in a local file system and provide an API to query for them.

yarn-project/blob-sink/package.json

@@ -0,0 +1,84 @@
+{
+  "name": "@aztec/blob-sink",
+  "version": "0.1.0",
+  "type": "module",
+  "exports": {
+    ".": "./dest/index.js"
+  },
+  "inherits": [
+    "../package.common.json"
+  ],
+  "scripts": {
+    "build": "yarn clean && tsc -b",
+    "build:dev": "tsc -b --watch",
+    "clean": "rm -rf ./dest .tsbuildinfo",
+    "formatting": "run -T prettier --check ./src && run -T eslint ./src",
+    "formatting:fix": "run -T eslint --fix ./src && run -T prettier -w ./src",
+    "test": "HARDWARE_CONCURRENCY=${HARDWARE_CONCURRENCY:-16} RAYON_NUM_THREADS=${RAYON_NUM_THREADS:-4} NODE_NO_WARNINGS=1 node --experimental-vm-modules ../node_modules/.bin/jest --passWithNoTests --maxWorkers=${JEST_MAX_WORKERS:-8}"
+  },
+  "jest": {
+    "moduleNameMapper": {
+      "^(\\.{1,2}/.*)\\.[cm]?js$": "$1"
+    },
+    "testRegex": "./src/.*\\.test\\.(js|mjs|ts)$",
+    "rootDir": "./src",
+    "transform": {
+      "^.+\\.tsx?$": [
+        "@swc/jest",
+        {
+          "jsc": {
+            "parser": {
+              "syntax": "typescript",
+              "decorators": true
+            },
+            "transform": {
+              "decoratorVersion": "2022-03"
+            }
+          }
+        }
+      ]
+    },
+    "extensionsToTreatAsEsm": [
+      ".ts"
+    ],
+    "reporters": [
+      "default"
+    ],
+    "testTimeout": 30000,
+    "setupFiles": [
+      "../../foundation/src/jest/setup.mjs"
+    ]
+  },
+  "dependencies": {
+    "@aztec/circuit-types": "workspace:^",
+    "@aztec/foundation": "workspace:^",
+    "@aztec/kv-store": "workspace:*",
+    "@aztec/telemetry-client": "workspace:*",
+    "express": "^4.21.1",
+    "source-map-support": "^0.5.21",
+    "tslib": "^2.4.0",
+    "zod": "^3.23.8"
+  },
+  "devDependencies": {
+    "@jest/globals": "^29.5.0",
+    "@types/jest": "^29.5.0",
+    "@types/memdown": "^3.0.0",
+    "@types/node": "^18.7.23",
+    "@types/source-map-support": "^0.5.10",
+    "@types/supertest": "^6.0.2",
+    "jest": "^29.5.0",
+    "jest-mock-extended": "^3.0.3",
+    "supertest": "^7.0.0",
+    "ts-node": "^10.9.1",
+    "typescript": "^5.0.4"
+  },
+  "files": [
+    "dest",
+    "src",
+    "!*.test.*"
+  ],
+  "types": "./dest/index.d.ts",
+  "engines": {
+    "node": ">=18"
+  }
+}

yarn-project/blob-sink/src/blob-sink.test.ts

@@ -0,0 +1,134 @@
+import { Blob } from '@aztec/foundation/blob';
+import { Fr } from '@aztec/foundation/fields';
+
+import request from 'supertest';
+
+import { BlobSinkServer } from './server.js';
+
+describe('BlobSinkService', () => {
+  let service: BlobSinkServer;
+
+  beforeEach(async () => {
+    service = new BlobSinkServer({
+      port: 0, // Using port 0 lets the OS assign a random available port
+    });
+    await service.start();
+  });
+
+  afterEach(async () => {
+    await service.stop();
+  });
+
+  describe('should store and retrieve a blob sidecar', () => {
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const testFields2 = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blob2 = Blob.fromFields(testFields2);
+    const blockId = '0x1234';
+
+    beforeEach(async () => {
+      // Post the blob
+      const postResponse = await request(service.getApp())
+        .post('/blob_sidecar')
+        .send({
+          // eslint-disable-next-line camelcase
+          block_id: blockId,
+          blobs: [
+            {
+              index: 0,
+              blob: blob.toBuffer(),
+            },
+            {
+              index: 1,
+              blob: blob2.toBuffer(),
+            },
+          ],
+        });
+
+      expect(postResponse.status).toBe(200);
+    });
+
+    it('should retrieve the blob', async () => {
+      // Retrieve the blob
+      const getResponse = await request(service.getApp()).get(`/eth/v1/beacon/blob_sidecars/${blockId}`);
+
+      expect(getResponse.status).toBe(200);
+
+      // Convert the response blob back to a Blob object and verify it matches
+      const retrievedBlobs = getResponse.body.data;
+
+      const retrievedBlob = Blob.fromBuffer(Buffer.from(retrievedBlobs[0].blob, 'hex'));
+      const retrievedBlob2 = Blob.fromBuffer(Buffer.from(retrievedBlobs[1].blob, 'hex'));
+      expect(retrievedBlob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+      expect(retrievedBlob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+      expect(retrievedBlob2.fieldsHash.toString()).toBe(blob2.fieldsHash.toString());
+      expect(retrievedBlob2.commitment.toString('hex')).toBe(blob2.commitment.toString('hex'));
+    });
+
+    it('should retrieve specific indicies', async () => {
+      // We can also request specific indicies
+      const getWithIndicies = await request(service.getApp()).get(
+        `/eth/v1/beacon/blob_sidecars/${blockId}?indices=0,1`,
+      );
+
+      expect(getWithIndicies.status).toBe(200);
+      expect(getWithIndicies.body.data.length).toBe(2);
+
+      const retrievedBlobs = getWithIndicies.body.data;
+      const retrievedBlob = Blob.fromBuffer(Buffer.from(retrievedBlobs[0].blob, 'hex'));
+      const retrievedBlob2 = Blob.fromBuffer(Buffer.from(retrievedBlobs[1].blob, 'hex'));
+      expect(retrievedBlob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+      expect(retrievedBlob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+      expect(retrievedBlob2.fieldsHash.toString()).toBe(blob2.fieldsHash.toString());
+      expect(retrievedBlob2.commitment.toString('hex')).toBe(blob2.commitment.toString('hex'));
+    });
+
+    it('should retreive a single index', async () => {
+      const getWithIndicies = await request(service.getApp()).get(`/eth/v1/beacon/blob_sidecars/${blockId}?indices=1`);
+
+      expect(getWithIndicies.status).toBe(200);
+      expect(getWithIndicies.body.data.length).toBe(1);
+
+      const retrievedBlobs = getWithIndicies.body.data;
+      const retrievedBlob = Blob.fromBuffer(Buffer.from(retrievedBlobs[0].blob, 'hex'));
+      expect(retrievedBlob.fieldsHash.toString()).toBe(blob2.fieldsHash.toString());
+      expect(retrievedBlob.commitment.toString('hex')).toBe(blob2.commitment.toString('hex'));
+    });
+  });
+
+  it('should return an error if invalid indicies are provided', async () => {
+    const blockId = '0x1234';
+
+    const response = await request(service.getApp()).get(`/eth/v1/beacon/blob_sidecars/${blockId}?indices=word`);
+    expect(response.status).toBe(400);
+    expect(response.body.error).toBe('Invalid indices parameter');
+  });
+
+  it('should return an error if the block ID is invalid (POST)', async () => {
+    const response = await request(service.getApp()).post('/blob_sidecar').send({
+      // eslint-disable-next-line camelcase
+      block_id: undefined,
+    });
+
+    expect(response.status).toBe(400);
+  });
+
+  it('should return an error if the block ID is invalid (GET)', async () => {
+    const response = await request(service.getApp()).get('/eth/v1/beacon/blob_sidecars/invalid-id');
+
+    expect(response.status).toBe(400);
+  });
+
+  it('should return 404 for non-existent blob', async () => {
+    const response = await request(service.getApp()).get('/eth/v1/beacon/blob_sidecars/0x999999');
+
+    expect(response.status).toBe(404);
+  });
+
+  it('should reject negative block IDs', async () => {
+    const response = await request(service.getApp()).get('/eth/v1/beacon/blob_sidecars/-123');
+
+    expect(response.status).toBe(400);
+    expect(response.body.error).toBe('Invalid block_id parameter');
+  });
+});

yarn-project/blob-sink/src/blobstore/blob_store_test_suite.ts

@@ -0,0 +1,142 @@
+import { Blob } from '@aztec/foundation/blob';
+import { Fr } from '@aztec/foundation/fields';
+
+import { BlobWithIndex } from '../types/index.js';
+import { type BlobStore } from './interface.js';
+
+export function describeBlobStore(getBlobStore: () => BlobStore) {
+  let blobStore: BlobStore;
+
+  beforeEach(() => {
+    blobStore = getBlobStore();
+  });
+
+  it('should store and retrieve a blob', async () => {
+    // Create a test blob with random fields
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blockId = '0x12345';
+    const blobWithIndex = new BlobWithIndex(blob, 0);
+
+    // Store the blob
+    await blobStore.addBlobSidecars(blockId, [blobWithIndex]);
+
+    // Retrieve the blob
+    const retrievedBlobs = await blobStore.getBlobSidecars(blockId);
+    const [retrievedBlob] = retrievedBlobs!;
+
+    // Verify the blob was retrieved and matches
+    expect(retrievedBlob).toBeDefined();
+    expect(retrievedBlob.blob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlob.blob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+  });
+
+  it('Should allow requesting a specific index of blob', async () => {
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blockId = '0x12345';
+    const blobWithIndex = new BlobWithIndex(blob, 0);
+    const blobWithIndex2 = new BlobWithIndex(blob, 1);
+
+    await blobStore.addBlobSidecars(blockId, [blobWithIndex, blobWithIndex2]);
+
+    const retrievedBlobs = await blobStore.getBlobSidecars(blockId, [0]);
+    const [retrievedBlob] = retrievedBlobs!;
+
+    expect(retrievedBlob.blob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlob.blob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+
+    const retrievedBlobs2 = await blobStore.getBlobSidecars(blockId, [1]);
+    const [retrievedBlob2] = retrievedBlobs2!;
+
+    expect(retrievedBlob2.blob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlob2.blob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+  });
+
+  it('Differentiate between blockHash and slot', async () => {
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const testFieldsSlot = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blobSlot = Blob.fromFields(testFieldsSlot);
+    const blockId = '0x12345';
+    const slot = '12345';
+    const blobWithIndex = new BlobWithIndex(blob, 0);
+    const blobWithIndexSlot = new BlobWithIndex(blobSlot, 0);
+
+    await blobStore.addBlobSidecars(blockId, [blobWithIndex]);
+    await blobStore.addBlobSidecars(slot, [blobWithIndexSlot]);
+
+    const retrievedBlobs = await blobStore.getBlobSidecars(blockId, [0]);
+    const [retrievedBlob] = retrievedBlobs!;
+
+    expect(retrievedBlob.blob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlob.blob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+
+    const retrievedBlobs2 = await blobStore.getBlobSidecars(slot, [0]);
+    const [retrievedBlob2] = retrievedBlobs2!;
+
+    expect(retrievedBlob2.blob.fieldsHash.toString()).toBe(blobSlot.fieldsHash.toString());
+    expect(retrievedBlob2.blob.commitment.toString('hex')).toBe(blobSlot.commitment.toString('hex'));
+  });
+
+  it('should return undefined for non-existent blob', async () => {
+    const nonExistentBlob = await blobStore.getBlobSidecars('999999');
+    expect(nonExistentBlob).toBeUndefined();
+  });
+
+  it('should handle multiple blobs with different block IDs', async () => {
+    // Create two different blobs
+    const blob1 = Blob.fromFields([Fr.random(), Fr.random()]);
+    const blob2 = Blob.fromFields([Fr.random(), Fr.random(), Fr.random()]);
+    const blobWithIndex1 = new BlobWithIndex(blob1, 0);
+    const blobWithIndex2 = new BlobWithIndex(blob2, 0);
+
+    // Store both blobs
+    await blobStore.addBlobSidecars('1', [blobWithIndex1]);
+    await blobStore.addBlobSidecars('2', [blobWithIndex2]);
+
+    // Retrieve and verify both blobs
+    const retrieved1 = await blobStore.getBlobSidecars('1');
+    const retrieved2 = await blobStore.getBlobSidecars('2');
+    const [retrievedBlob1] = retrieved1!;
+    const [retrievedBlob2] = retrieved2!;
+
+    expect(retrievedBlob1.blob.commitment.toString('hex')).toBe(blob1.commitment.toString('hex'));
+    expect(retrievedBlob2.blob.commitment.toString('hex')).toBe(blob2.commitment.toString('hex'));
+  });
+
+  it('should overwrite blob when using same block ID', async () => {
+    // Create two different blobs
+    const originalBlob = Blob.fromFields([Fr.random()]);
+    const newBlob = Blob.fromFields([Fr.random(), Fr.random()]);
+    const blockId = '1';
+    const originalBlobWithIndex = new BlobWithIndex(originalBlob, 0);
+    const newBlobWithIndex = new BlobWithIndex(newBlob, 0);
+
+    // Store original blob
+    await blobStore.addBlobSidecars(blockId, [originalBlobWithIndex]);
+
+    // Overwrite with new blob
+    await blobStore.addBlobSidecars(blockId, [newBlobWithIndex]);
+
+    // Retrieve and verify it's the new blob
+    const retrievedBlobs = await blobStore.getBlobSidecars(blockId);
+    const [retrievedBlob] = retrievedBlobs!;
+    expect(retrievedBlob.blob.commitment.toString('hex')).toBe(newBlob.commitment.toString('hex'));
+    expect(retrievedBlob.blob.commitment.toString('hex')).not.toBe(originalBlob.commitment.toString('hex'));
+  });
+
+  it('should handle multiple blobs with the same block ID', async () => {
+    const blob1 = Blob.fromFields([Fr.random()]);
+    const blob2 = Blob.fromFields([Fr.random()]);
+    const blobWithIndex1 = new BlobWithIndex(blob1, 0);
+    const blobWithIndex2 = new BlobWithIndex(blob2, 0);
+
+    await blobStore.addBlobSidecars('1', [blobWithIndex1, blobWithIndex2]);
+    const retrievedBlobs = await blobStore.getBlobSidecars('1');
+    const [retrievedBlob1, retrievedBlob2] = retrievedBlobs!;
+
+    expect(retrievedBlob1.blob.commitment.toString('hex')).toBe(blob1.commitment.toString('hex'));
+    expect(retrievedBlob2.blob.commitment.toString('hex')).toBe(blob2.commitment.toString('hex'));
+  });
+}

yarn-project/blob-sink/src/blobstore/disk_blob_store.test.ts

@@ -0,0 +1,8 @@
+import { openTmpStore } from '@aztec/kv-store/lmdb';
+
+import { describeBlobStore } from './blob_store_test_suite.js';
+import { DiskBlobStore } from './disk_blob_store.js';
+
+describe('DiskBlobStore', () => {
+  describeBlobStore(() => new DiskBlobStore(openTmpStore()));
+});