feat: blob sink in sandbox without extra process (#11032)

yarn-project/archiver/package.json

@@ -64,6 +64,7 @@
     ]
   },
   "dependencies": {
+    "@aztec/blob-sink": "workspace:^",
     "@aztec/circuit-types": "workspace:^",
     "@aztec/circuits.js": "workspace:^",
     "@aztec/ethereum": "workspace:^",

yarn-project/archiver/src/archiver/archiver.test.ts

@@ -1,3 +1,4 @@
+import { type BlobSinkClientInterface } from '@aztec/blob-sink/client';
 import { InboxLeaf, type L1RollupConstants, L2Block } from '@aztec/circuit-types';
 import { GENESIS_ARCHIVE_ROOT, PrivateLog } from '@aztec/circuits.js';
 import { DefaultL1ContractsConfig } from '@aztec/ethereum';
@@ -53,6 +54,7 @@ describe('Archiver', () => {
 
   let publicClient: MockProxy<PublicClient<HttpTransport, Chain>>;
   let instrumentation: MockProxy<ArchiverInstrumentation>;
+  let blobSinkClient: MockProxy<BlobSinkClientInterface>;
   let archiverStore: ArchiverDataStore;
   let now: number;
   let l1Constants: L1RollupConstants;
@@ -92,6 +94,7 @@ describe('Archiver', () => {
         );
       }) as any,
     });
+    blobSinkClient = mock<BlobSinkClientInterface>();
 
     const tracer = new NoopTelemetryClient().getTracer();
     instrumentation = mock<ArchiverInstrumentation>({ isEnabled: () => true, tracer });
@@ -109,6 +112,7 @@ describe('Archiver', () => {
       { rollupAddress, inboxAddress, registryAddress },
       archiverStore,
       { pollingIntervalMs: 1000, batchSize: 1000 },
+      blobSinkClient,
       instrumentation,
       l1Constants,
     );

yarn-project/archiver/src/archiver/archiver.ts

@@ -1,3 +1,4 @@
+import { type BlobSinkClientInterface } from '@aztec/blob-sink/client';
 import {
   type GetUnencryptedLogsResponse,
   type InBlock,
@@ -115,6 +116,7 @@ export class Archiver implements ArchiveSource, Traceable {
     private readonly l1Addresses: { rollupAddress: EthAddress; inboxAddress: EthAddress; registryAddress: EthAddress },
     readonly dataStore: ArchiverDataStore,
     private readonly config: { pollingIntervalMs: number; batchSize: number },
+    private readonly _blobSinkClient: BlobSinkClientInterface,
     private readonly instrumentation: ArchiverInstrumentation,
     private readonly l1constants: L1RollupConstants,
     private readonly log: Logger = createLogger('archiver'),
@@ -145,7 +147,7 @@ export class Archiver implements ArchiveSource, Traceable {
   public static async createAndSync(
     config: ArchiverConfig,
     archiverStore: ArchiverDataStore,
-    telemetry: TelemetryClient,
+    deps: { telemetry: TelemetryClient; blobSinkClient: BlobSinkClientInterface },
     blockUntilSynced = true,
   ): Promise<Archiver> {
     const chain = createEthereumChain(config.l1RpcUrl, config.l1ChainId);
@@ -176,7 +178,8 @@ export class Archiver implements ArchiveSource, Traceable {
         pollingIntervalMs: config.archiverPollingIntervalMS ?? 10_000,
         batchSize: config.archiverBatchSize ?? 100,
       },
-      await ArchiverInstrumentation.new(telemetry, () => archiverStore.estimateSize()),
+      deps.blobSinkClient,
+      await ArchiverInstrumentation.new(deps.telemetry, () => archiverStore.estimateSize()),
       { l1StartBlock, l1GenesisTime, epochDuration, slotDuration, ethereumSlotDuration },
     );
     await archiver.start(blockUntilSynced);

yarn-project/archiver/src/factory.ts

@@ -1,3 +1,4 @@
+import { type BlobSinkClientInterface } from '@aztec/blob-sink/client';
 import { type ArchiverApi, type Service } from '@aztec/circuit-types';
 import {
   type ContractClassPublic,
@@ -23,6 +24,7 @@ import { createArchiverClient } from './rpc/index.js';
 
 export async function createArchiver(
   config: ArchiverConfig & DataStoreConfig,
+  blobSinkClient: BlobSinkClientInterface,
   telemetry: TelemetryClient = new NoopTelemetryClient(),
   opts: { blockUntilSync: boolean } = { blockUntilSync: true },
 ): Promise<ArchiverApi & Maybe<Service>> {
@@ -31,7 +33,7 @@ export async function createArchiver(
     const archiverStore = new KVArchiverDataStore(store, config.maxLogs);
     await registerProtocolContracts(archiverStore);
     await registerCommonContracts(archiverStore);
-    return Archiver.createAndSync(config, archiverStore, telemetry, opts.blockUntilSync);
+    return Archiver.createAndSync(config, archiverStore, { telemetry, blobSinkClient }, opts.blockUntilSync);
   } else {
     return createArchiverClient(config.archiverUrl);
   }

yarn-project/archiver/tsconfig.json

@@ -6,6 +6,9 @@
     "tsBuildInfoFile": ".tsbuildinfo"
   },
   "references": [
+    {
+      "path": "../blob-sink"
+    },
     {
       "path": "../circuit-types"
     },

yarn-project/aztec-node/package.json

@@ -64,6 +64,7 @@
   "dependencies": {
     "@aztec/archiver": "workspace:^",
     "@aztec/bb-prover": "workspace:^",
+    "@aztec/blob-sink": "workspace:^",
     "@aztec/circuit-types": "workspace:^",
     "@aztec/circuits.js": "workspace:^",
     "@aztec/epoch-cache": "workspace:^",

yarn-project/aztec-node/src/aztec-node/server.ts

@@ -1,5 +1,6 @@
 import { createArchiver } from '@aztec/archiver';
 import { BBCircuitVerifier, TestCircuitVerifier } from '@aztec/bb-prover';
+import { type BlobSinkClientInterface, createBlobSinkClient } from '@aztec/blob-sink/client';
 import {
   type AztecNode,
   type ClientProtocolCircuitVerifier,
@@ -139,11 +140,13 @@ export class AztecNodeService implements AztecNode, Traceable {
       logger?: Logger;
       publisher?: L1Publisher;
       dateProvider?: DateProvider;
+      blobSinkClient?: BlobSinkClientInterface;
     } = {},
   ): Promise<AztecNodeService> {
     const telemetry = deps.telemetry ?? new NoopTelemetryClient();
     const log = deps.logger ?? createLogger('node');
     const dateProvider = deps.dateProvider ?? new DateProvider();
+    const blobSinkClient = deps.blobSinkClient ?? createBlobSinkClient(config.blobSinkUrl);
     const ethereumChain = createEthereumChain(config.l1RpcUrl, config.l1ChainId);
     //validate that the actual chain id matches that specified in configuration
     if (config.l1ChainId !== ethereumChain.chainInfo.id) {
@@ -152,7 +155,7 @@ export class AztecNodeService implements AztecNode, Traceable {
       );
     }
 
-    const archiver = await createArchiver(config, telemetry, { blockUntilSync: true });
+    const archiver = await createArchiver(config, blobSinkClient, telemetry, { blockUntilSync: true });
 
     // we identify the P2P transaction protocol by using the rollup contract address.
     // this may well change in future
@@ -190,6 +193,7 @@ export class AztecNodeService implements AztecNode, Traceable {
     const sequencer = config.disableValidator
       ? undefined
       : await SequencerClient.new(config, {
+          ...deps,
           validatorClient,
           p2pClient,
           worldStateSynchronizer,
@@ -199,7 +203,7 @@ export class AztecNodeService implements AztecNode, Traceable {
           l1ToL2MessageSource: archiver,
           telemetry,
           dateProvider,
-          ...deps,
+          blobSinkClient,
         });
 
     return new AztecNodeService(

yarn-project/aztec-node/tsconfig.json

@@ -12,6 +12,9 @@
     {
       "path": "../bb-prover"
     },
+    {
+      "path": "../blob-sink"
+    },
     {
       "path": "../circuit-types"
     },

yarn-project/aztec/package.json

@@ -35,6 +35,7 @@
     "@aztec/aztec-node": "workspace:^",
     "@aztec/aztec.js": "workspace:^",
     "@aztec/bb-prover": "workspace:^",
+    "@aztec/blob-sink": "workspace:^",
     "@aztec/bot": "workspace:^",
     "@aztec/builder": "workspace:^",
     "@aztec/circuit-types": "workspace:^",

yarn-project/aztec/src/cli/cmds/start_archiver.ts

@@ -1,5 +1,6 @@
 import { Archiver, type ArchiverConfig, KVArchiverDataStore, archiverConfigMappings } from '@aztec/archiver';
 import { createLogger } from '@aztec/aztec.js';
+import { createBlobSinkClient } from '@aztec/blob-sink/client';
 import { ArchiverApiSchema } from '@aztec/circuit-types';
 import { type NamespacedApiHandlers } from '@aztec/foundation/json-rpc/server';
 import { type DataStoreConfig, dataConfigMappings } from '@aztec/kv-store/config';
@@ -31,7 +32,9 @@ export async function startArchiver(
   const archiverStore = new KVArchiverDataStore(store, archiverConfig.maxLogs);
 
   const telemetry = await createAndStartTelemetryClient(getTelemetryClientConfig());
-  const archiver = await Archiver.createAndSync(archiverConfig, archiverStore, telemetry, true);
+  // TODO(https://github.com/AztecProtocol/aztec-packages/issues/10056): place CL url in config here
+  const blobSinkClient = createBlobSinkClient();
+  const archiver = await Archiver.createAndSync(archiverConfig, archiverStore, { telemetry, blobSinkClient }, true);
   services.archiver = [archiver, ArchiverApiSchema];
   signalHandlers.push(archiver.stop);
   return services;

yarn-project/aztec/src/cli/cmds/start_node.ts

@@ -91,7 +91,7 @@ export async function startNode(
   const telemetry = await createAndStartTelemetryClient(telemetryConfig);
 
   // Create and start Aztec Node
-  const node = await createAztecNode(nodeConfig, telemetry);
+  const node = await createAztecNode(nodeConfig, { telemetry });
 
   // Add node and p2p to services list
   services.node = [node, AztecNodeApiSchema];

yarn-project/aztec/src/sandbox.ts

@@ -2,6 +2,7 @@
 import { type AztecNodeConfig, AztecNodeService, getConfigEnvVars } from '@aztec/aztec-node';
 import { AnvilTestWatcher, EthCheatCodes, SignerlessWallet, retryUntil } from '@aztec/aztec.js';
 import { DefaultMultiCallEntrypoint } from '@aztec/aztec.js/entrypoint';
+import { type BlobSinkClientInterface, createBlobSinkClient } from '@aztec/blob-sink/client';
 import { type AztecNode } from '@aztec/circuit-types';
 import { setupCanonicalL2FeeJuice } from '@aztec/cli/setup-contracts';
 import {
@@ -143,8 +144,10 @@ export async function createSandbox(config: Partial<SandboxConfig> = {}) {
     await watcher.start();
   }
 
-  const client = await createAndStartTelemetryClient(getTelemetryClientConfig());
-  const node = await createAztecNode(aztecNodeConfig, client);
+  const telemetry = await createAndStartTelemetryClient(getTelemetryClientConfig());
+  // Create a local blob sink client inside the sandbox, no http connectivity
+  const blobSinkClient = createBlobSinkClient();
+  const node = await createAztecNode(aztecNodeConfig, { telemetry, blobSinkClient });
   const pxe = await createAztecPXE(node);
 
   if (config.enableGas) {
@@ -168,9 +171,12 @@ export async function createSandbox(config: Partial<SandboxConfig> = {}) {
  * Create and start a new Aztec RPC HTTP Server
  * @param config - Optional Aztec node settings.
  */
-export async function createAztecNode(config: Partial<AztecNodeConfig> = {}, telemetryClient?: TelemetryClient) {
+export async function createAztecNode(
+  config: Partial<AztecNodeConfig> = {},
+  deps: { telemetry?: TelemetryClient; blobSinkClient?: BlobSinkClientInterface } = {},
+) {
   const aztecNodeConfig: AztecNodeConfig = { ...getConfigEnvVars(), ...config };
-  const node = await AztecNodeService.createAndSync(aztecNodeConfig, { telemetry: telemetryClient });
+  const node = await AztecNodeService.createAndSync(aztecNodeConfig, deps);
   return node;
 }
 

yarn-project/aztec/tsconfig.json

@@ -24,6 +24,9 @@
     {
       "path": "../bb-prover"
     },
+    {
+      "path": "../blob-sink"
+    },
     {
       "path": "../bot"
     },

yarn-project/blob-sink/package.json

@@ -3,7 +3,8 @@
   "version": "0.1.0",
   "type": "module",
   "exports": {
-    ".": "./dest/index.js"
+    "./server": "./dest/server/index.js",
+    "./client": "./dest/client/index.js"
   },
   "inherits": [
     "../package.common.json"

yarn-project/blob-sink/src/client/blob-sink-client-tests.ts

@@ -0,0 +1,72 @@
+import { Blob } from '@aztec/foundation/blob';
+import { Fr } from '@aztec/foundation/fields';
+
+import { type BlobSinkClientInterface } from './interface.js';
+
+/**
+ * Shared test suite for blob sink clients
+ * @param createClient - Function that creates a client instance for testing
+ * @param cleanup - Optional cleanup function to run after each test
+ */
+export function runBlobSinkClientTests(
+  createClient: () => Promise<{ client: BlobSinkClientInterface; cleanup: () => Promise<void> }>,
+) {
+  let client: BlobSinkClientInterface;
+  let cleanup: () => Promise<void>;
+
+  beforeEach(async () => {
+    const setup = await createClient();
+    client = setup.client;
+    cleanup = setup.cleanup;
+  });
+
+  afterEach(async () => {
+    await cleanup();
+  });
+
+  it('should send and retrieve blobs', async () => {
+    const testFields = [Fr.random(), Fr.random(), Fr.random()];
+    const blob = Blob.fromFields(testFields);
+    const blockId = '0x1234';
+
+    const success = await client.sendBlobsToBlobSink(blockId, [blob]);
+    expect(success).toBe(true);
+
+    const retrievedBlobs = await client.getBlobSidecar(blockId);
+    expect(retrievedBlobs).toHaveLength(1);
+    expect(retrievedBlobs[0].fieldsHash.toString()).toBe(blob.fieldsHash.toString());
+    expect(retrievedBlobs[0].commitment.toString('hex')).toBe(blob.commitment.toString('hex'));
+  });
+
+  it('should handle multiple blobs', async () => {
+    const blobs = [
+      Blob.fromFields([Fr.random(), Fr.random()]),
+      Blob.fromFields([Fr.random(), Fr.random()]),
+      Blob.fromFields([Fr.random(), Fr.random()]),
+    ];
+    const blockId = '0x5678';
+
+    const success = await client.sendBlobsToBlobSink(blockId, blobs);
+    expect(success).toBe(true);
+
+    const retrievedBlobs = await client.getBlobSidecar(blockId);
+    expect(retrievedBlobs).toHaveLength(3);
+
+    for (let i = 0; i < blobs.length; i++) {
+      expect(retrievedBlobs[i].fieldsHash.toString()).toBe(blobs[i].fieldsHash.toString());
+      expect(retrievedBlobs[i].commitment.toString('hex')).toBe(blobs[i].commitment.toString('hex'));
+    }
+
+    // Can request blobs by index
+    const retrievedBlobsByIndex = await client.getBlobSidecar(blockId, [0, 2]);
+    expect(retrievedBlobsByIndex).toHaveLength(2);
+    expect(retrievedBlobsByIndex[0].fieldsHash.toString()).toBe(blobs[0].fieldsHash.toString());
+    expect(retrievedBlobsByIndex[1].fieldsHash.toString()).toBe(blobs[2].fieldsHash.toString());
+  });
+
+  it('should return empty array for non-existent block', async () => {
+    const blockId = '0xnonexistent';
+    const retrievedBlobs = await client.getBlobSidecar(blockId);
+    expect(retrievedBlobs).toEqual([]);
+  });
+}

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

@@ -0,0 +1,13 @@
+import { MemoryBlobStore } from '../blobstore/memory_blob_store.js';
+import { HttpBlobSinkClient } from './http.js';
+import { type BlobSinkClientInterface } from './interface.js';
+import { LocalBlobSinkClient } from './local.js';
+
+export function createBlobSinkClient(blobSinkUrl?: string): BlobSinkClientInterface {
+  if (!blobSinkUrl) {
+    const blobStore = new MemoryBlobStore();
+    return new LocalBlobSinkClient(blobStore);
+  }
+
+  return new HttpBlobSinkClient(blobSinkUrl);
+}

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

@@ -0,0 +1,35 @@
+import { Blob } from '@aztec/foundation/blob';
+import { Fr } from '@aztec/foundation/fields';
+
+import { BlobSinkServer } from '../server/server.js';
+import { runBlobSinkClientTests } from './blob-sink-client-tests.js';
+import { HttpBlobSinkClient } from './http.js';
+
+describe('HttpBlobSinkClient', () => {
+  runBlobSinkClientTests(async () => {
+    const server = new BlobSinkServer({
+      port: 0,
+    });
+    await server.start();
+
+    const client = new HttpBlobSinkClient(`http://localhost:${server.port}`);
+
+    return {
+      client,
+      cleanup: async () => {
+        await server.stop();
+      },
+    };
+  });
+
+  it('should handle server connection errors gracefully', async () => {
+    const client = new HttpBlobSinkClient('http://localhost:12345'); // Invalid port
+    const blob = Blob.fromFields([Fr.random()]);
+
+    const success = await client.sendBlobsToBlobSink('0x1234', [blob]);
+    expect(success).toBe(false);
+
+    const retrievedBlobs = await client.getBlobSidecar('0x1234');
+    expect(retrievedBlobs).toEqual([]);
+  });
+});