feat(blobs): blob sink

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

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

yarn-project/blob-sink/README.md

@@ -0,0 +1,7 @@
+## Blob Sink
+
+A HTTP api that emulated the https://ethereum.github.io/beacon-APIs/?urls.primaryName=dev#/Beacon/getBlobSidecars API.
+
+## 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.

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,91 @@
+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();
+  });
+
+  it('should store and retrieve a blob sidecar', async () => {
+    // Create a test blob
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blockId = '0x1234';
+
+    // 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(),
+          },
+        ],
+      });
+
+    expect(postResponse.status).toBe(200);
+
+    // 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'));
+    expect(retrievedBlob.fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlob.commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+  });
+
+  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 invalid block IDs', async () => {
+    const response = await request(service.getApp()).get('/eth/v1/beacon/blob_sidecars/invalid-id');
+
+    expect(response.status).toBe(400);
+    expect(response.body.error).toBe('Invalid block_id parameter');
+  });
+
+  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,94 @@
+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 = '12345';
+    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 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()));
+});

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

@@ -0,0 +1,25 @@
+import { type AztecKVStore, type AztecMap } from '@aztec/kv-store';
+
+import { type BlobWithIndex, BlobsWithIndexes } from '../types/index.js';
+import { type BlobStore } from './interface.js';
+
+export class DiskBlobStore implements BlobStore {
+  blobs: AztecMap<string, Buffer>;
+
+  constructor(store: AztecKVStore) {
+    this.blobs = store.openMap('blobs');
+  }
+
+  public getBlobSidecars(blockId: string): Promise<BlobWithIndex[] | undefined> {
+    const blobBuffer = this.blobs.get(`${blockId}`);
+    if (!blobBuffer) {
+      return Promise.resolve(undefined);
+    }
+    return Promise.resolve(BlobsWithIndexes.fromBuffer(blobBuffer).blobs);
+  }
+
+  public async addBlobSidecars(blockId: string, blobSidecars: BlobWithIndex[]): Promise<void> {
+    await this.blobs.set(blockId, new BlobsWithIndexes(blobSidecars).toBuffer());
+    return Promise.resolve();
+  }
+}

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

@@ -0,0 +1,3 @@
+export * from './memory_blob_store.js';
+export * from './disk_blob_store.js';
+export * from './interface.js';

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

@@ -0,0 +1,12 @@
+import { type BlobWithIndex } from '../types/index.js';
+
+export interface BlobStore {
+  /**
+   * Get a blob by block id
+   */
+  getBlobSidecars: (blockId: string) => Promise<BlobWithIndex[] | undefined>;
+  /**
+   * Add a blob to the store
+   */
+  addBlobSidecars: (blockId: string, blobSidecars: BlobWithIndex[]) => Promise<void>;
+}

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

@@ -0,0 +1,6 @@
+import { describeBlobStore } from './blob_store_test_suite.js';
+import { MemoryBlobStore } from './memory_blob_store.js';
+
+describe('MemoryBlobStore', () => {
+  describeBlobStore(() => new MemoryBlobStore());
+});

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

@@ -0,0 +1,19 @@
+import { type BlobWithIndex, BlobsWithIndexes } from '../types/index.js';
+import { type BlobStore } from './interface.js';
+
+export class MemoryBlobStore implements BlobStore {
+  private blobs: Map<string, Buffer> = new Map();
+
+  public getBlobSidecars(blockId: string): Promise<BlobWithIndex[] | undefined> {
+    const blobBuffer = this.blobs.get(blockId);
+    if (!blobBuffer) {
+      return Promise.resolve(undefined);
+    }
+    return Promise.resolve(BlobsWithIndexes.fromBuffer(blobBuffer).blobs);
+  }
+
+  public addBlobSidecars(blockId: string, blobSidecars: BlobWithIndex[]): Promise<void> {
+    this.blobs.set(blockId, new BlobsWithIndexes(blobSidecars).toBuffer());
+    return Promise.resolve();
+  }
+}

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

@@ -0,0 +1,7 @@
+import { type DataStoreConfig } from '@aztec/kv-store/config';
+
+export interface BlobSinkConfig {
+  port?: number;
+  dataStoreConfig?: DataStoreConfig;
+  otelMetricsCollectorUrl?: string;
+}

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

@@ -0,0 +1,27 @@
+import { type AztecKVStore } from '@aztec/kv-store';
+import { createStore } from '@aztec/kv-store/lmdb';
+import { type TelemetryClient } from '@aztec/telemetry-client';
+
+import { type BlobSinkConfig } from './config.js';
+import { BlobSinkServer } from './server.js';
+
+// If data store settings are provided, the store is created and returned.
+// Otherwise, undefined is returned and an in memory store will be used.
+async function getDataStoreConfig(config?: BlobSinkConfig): Promise<AztecKVStore | undefined> {
+  if (!config?.dataStoreConfig) {
+    return undefined;
+  }
+  return await createStore('blob-sink', config.dataStoreConfig);
+}
+
+/**
+ * Creates a blob sink service from the provided config.
+ */
+export async function createBlobSinkServer(
+  config?: BlobSinkConfig,
+  telemetry?: TelemetryClient,
+): Promise<BlobSinkServer> {
+  const store = await getDataStoreConfig(config);
+
+  return new BlobSinkServer(config, store, telemetry);
+}