feat: Blob filestores (#18971)

Added `FileStoreBlobClient` that enables reading and writing blobs to
cloud storage (S3, GCS) or local filesystem. This provides an
alternative blob source alongside the existing blob sink server, L1
consensus layer, and archive (e.g., Blobscan).

## Blob Download Logic

The download strategy varies based on sync state:

- **Historical sync:** Blob sink → FileStore → L1 consensus → Archive
- **Near-tip sync:** Blob sink → FileStore → L1 consensus → FileStore
(with retries) → Archive

Near-tip sync includes retries with exponential backoff, assuming
filestores should already contain the data when blobs are queried.

## Blob Upload Logic

Blobs are uploaded only if the node has write access to a filestore.
Uploads occur in three scenarios:

1. `SequencerPublisher` publishes blobs when posting a checkpoint to L1
2. The archiver fetches blobs from any source
3. A validator receives a valid block proposal

## Configuration

The following environment variables are introduced:

| Variable | Description |
|----------|-------------|
| `BLOB_FILE_STORE_URLS` | Comma-separated URLs for reading blobs
(`s3://`, `gs://`, `file://`, `https://`) |
| `BLOB_FILE_STORE_UPLOAD_URL` | URL for uploading blobs (`s3://`,
`gs://`, `file://`) |

`BLOB_FILE_STORE_URLS` can also be defined in the network configuration,
enabling centralized management of these URLs across nodes.

Author: mrzeszutko

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

@@ -970,6 +970,7 @@ export class Archiver
         this.l1Addresses,
         this.instrumentation,
         this.log,
+        !this.initialSyncComplete, // isHistoricalSync
       );
 
       if (retrievedCheckpoints.length === 0) {

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

@@ -137,13 +137,17 @@ export async function retrievedToPublishedCheckpoint({
 
 /**
  * Fetches new checkpoints.
+ * @param rollup - The rollup contract instance.
  * @param publicClient - The viem public client to use for transaction retrieval.
  * @param debugClient - The viem debug client to use for trace/debug RPC methods (optional).
- * @param rollupAddress - The address of the rollup contract.
+ * @param blobSinkClient - The blob sink client for fetching blob data.
  * @param searchStartBlock - The block number to use for starting the search.
  * @param searchEndBlock - The highest block number that we should search up to.
- * @param expectedNextL2BlockNum - The next L2 block number that we expect to find.
- * @returns An array of block; as well as the next eth block to search from.
+ * @param contractAddresses - The contract addresses (governanceProposerAddress, slashFactoryAddress, slashingProposerAddress).
+ * @param instrumentation - The archiver instrumentation instance.
+ * @param logger - The logger instance.
+ * @param isHistoricalSync - Whether this is a historical sync.
+ * @returns An array of retrieved checkpoints.
  */
 export async function retrieveCheckpointsFromRollup(
   rollup: GetContractReturnType<typeof RollupAbi, ViemPublicClient>,
@@ -159,6 +163,7 @@ export async function retrieveCheckpointsFromRollup(
   },
   instrumentation: ArchiverInstrumentation,
   logger: Logger = createLogger('archiver'),
+  isHistoricalSync: boolean = false,
 ): Promise<RetrievedCheckpoint[]> {
   const retrievedCheckpoints: RetrievedCheckpoint[] = [];
 
@@ -210,6 +215,7 @@ export async function retrieveCheckpointsFromRollup(
       contractAddresses,
       instrumentation,
       logger,
+      isHistoricalSync,
     );
     retrievedCheckpoints.push(...newCheckpoints);
     searchStartBlock = lastLog.blockNumber! + 1n;
@@ -221,11 +227,17 @@ export async function retrieveCheckpointsFromRollup(
 
 /**
  * Processes newly received CheckpointProposed logs.
- * @param rollup - The rollup contract
+ * @param rollup - The rollup contract instance.
  * @param publicClient - The viem public client to use for transaction retrieval.
  * @param debugClient - The viem debug client to use for trace/debug RPC methods (optional).
+ * @param blobSinkClient - The blob sink client for fetching blob data.
  * @param logs - CheckpointProposed logs.
- * @returns - An array of checkpoints.
+ * @param rollupConstants - The rollup constants (chainId, version, targetCommitteeSize).
+ * @param contractAddresses - The contract addresses (governanceProposerAddress, slashFactoryAddress, slashingProposerAddress).
+ * @param instrumentation - The archiver instrumentation instance.
+ * @param logger - The logger instance.
+ * @param isHistoricalSync - Whether this is a historical sync.
+ * @returns An array of retrieved checkpoints.
  */
 async function processCheckpointProposedLogs(
   rollup: GetContractReturnType<typeof RollupAbi, ViemPublicClient>,
@@ -241,6 +253,7 @@ async function processCheckpointProposedLogs(
   },
   instrumentation: ArchiverInstrumentation,
   logger: Logger,
+  isHistoricalSync: boolean,
 ): Promise<RetrievedCheckpoint[]> {
   const retrievedCheckpoints: RetrievedCheckpoint[] = [];
   const calldataRetriever = new CalldataRetriever(
@@ -278,6 +291,7 @@ async function processCheckpointProposedLogs(
         blobHashes,
         checkpointNumber,
         logger,
+        isHistoricalSync,
       );
 
       const l1 = new L1PublishedData(
@@ -315,8 +329,9 @@ export async function getCheckpointBlobDataFromBlobs(
   blobHashes: Buffer<ArrayBufferLike>[],
   checkpointNumber: CheckpointNumber,
   logger: Logger,
+  isHistoricalSync: boolean,
 ): Promise<CheckpointBlobData> {
-  const blobBodies = await blobSinkClient.getBlobSidecar(blockHash, blobHashes);
+  const blobBodies = await blobSinkClient.getBlobSidecar(blockHash, blobHashes, undefined, { isHistoricalSync });
   if (blobBodies.length === 0) {
     throw new NoBlobBodiesFoundError(checkpointNumber);
   }

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

@@ -1,6 +1,11 @@
 import { Archiver, createArchiver } from '@aztec/archiver';
 import { BBCircuitVerifier, QueuedIVCVerifier, TestCircuitVerifier } from '@aztec/bb-prover';
 import { type BlobSinkClientInterface, createBlobSinkClient } from '@aztec/blob-sink/client';
+import {
+  type BlobFileStoreMetadata,
+  createReadOnlyFileStoreBlobClients,
+  createWritableFileStoreBlobClient,
+} from '@aztec/blob-sink/filestore';
 import {
   ARCHIVE_HEIGHT,
   INITIAL_L2_BLOCK_NUM,
@@ -197,8 +202,6 @@ export class AztecNodeService implements AztecNode, AztecNodeAdmin, Traceable {
     const packageVersion = getPackageVersion() ?? '';
     const telemetry = deps.telemetry ?? getTelemetryClient();
     const dateProvider = deps.dateProvider ?? new DateProvider();
-    const blobSinkClient =
-      deps.blobSinkClient ?? createBlobSinkClient(config, { logger: createLogger('node:blob-sink:client') });
     const ethereumChain = createEthereumChain(config.l1RpcUrls, config.l1ChainId);
 
     // Build a key store from file if given or from environment otherwise
@@ -266,6 +269,25 @@ export class AztecNodeService implements AztecNode, AztecNodeAdmin, Traceable {
       );
     }
 
+    const blobFileStoreMetadata: BlobFileStoreMetadata = {
+      l1ChainId: config.l1ChainId,
+      rollupVersion: config.rollupVersion,
+      rollupAddress: config.l1Contracts.rollupAddress.toString(),
+    };
+
+    const [fileStoreClients, fileStoreUploadClient] = await Promise.all([
+      createReadOnlyFileStoreBlobClients(config.blobFileStoreUrls, blobFileStoreMetadata, log),
+      createWritableFileStoreBlobClient(config.blobFileStoreUploadUrl, blobFileStoreMetadata, log),
+    ]);
+
+    const blobSinkClient =
+      deps.blobSinkClient ??
+      createBlobSinkClient(config, {
+        logger: createLogger('node:blob-sink:client'),
+        fileStoreClients,
+        fileStoreUploadClient,
+      });
+
     // attempt snapshot sync if possible
     await trySnapshotSync(config, log);
 
@@ -331,6 +353,7 @@ export class AztecNodeService implements AztecNode, AztecNodeAdmin, Traceable {
       blockSource: archiver,
       l1ToL2MessageSource: archiver,
       keyStoreManager,
+      fileStoreBlobUploadClient: fileStoreUploadClient,
     });
 
     // If we have a validator client, register it as a source of offenses for the slasher,

yarn-project/blob-sink/package.json

@@ -11,7 +11,8 @@
     "./client": "./dest/client/index.js",
     "./client/config": "./dest/client/config.js",
     "./encoding": "./dest/encoding/index.js",
-    "./types": "./dest/types/index.js"
+    "./types": "./dest/types/index.js",
+    "./filestore": "./dest/filestore/index.js"
   },
   "inherits": [
     "../package.common.json"

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

@@ -45,6 +45,16 @@ export interface BlobSinkConfig extends BlobSinkArchiveApiConfig {
    * Whether to allow having no blob sources configured during startup
    */
   blobAllowEmptySources?: boolean;
+
+  /**
+   * URLs for reading blobs from filestore (s3://, gs://, file://, https://). Tried in order until blobs are found.
+   */
+  blobFileStoreUrls?: string[];
+
+  /**
+   * URL for uploading blobs to filestore (s3://, gs://, file://)
+   */
+  blobFileStoreUploadUrl?: string;
 }
 
 export const blobSinkConfigMapping: ConfigMappingsType<BlobSinkConfig> = {
@@ -84,6 +94,19 @@ export const blobSinkConfigMapping: ConfigMappingsType<BlobSinkConfig> = {
     description: 'Whether to allow having no blob sources configured during startup',
     ...booleanConfigHelper(false),
   },
+  blobFileStoreUrls: {
+    env: 'BLOB_FILE_STORE_URLS',
+    description: 'URLs for filestore blob archive, comma-separated. Tried in order until blobs are found.',
+    parseEnv: (val: string) =>
+      val
+        .split(',')
+        .map(url => url.trim())
+        .filter(url => url.length > 0),
+  },
+  blobFileStoreUploadUrl: {
+    env: 'BLOB_FILE_STORE_UPLOAD_URL',
+    description: 'URL for uploading blobs to filestore (s3://, gs://, file://)',
+  },
   ...blobSinkArchiveApiConfigMappings,
 };
 
@@ -99,5 +122,10 @@ export function getBlobSinkConfigFromEnv(): BlobSinkConfig {
  * Returns whether the given blob sink config has any remote sources defined.
  */
 export function hasRemoteBlobSinkSources(config: BlobSinkConfig = {}): boolean {
-  return !!(config.blobSinkUrl || config.l1ConsensusHostUrls?.length || config.archiveApiUrl);
+  return !!(
+    config.blobSinkUrl ||
+    config.l1ConsensusHostUrls?.length ||
+    config.archiveApiUrl ||
+    config.blobFileStoreUrls?.length
+  );
 }

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

@@ -1,12 +1,24 @@
 import { type Logger, createLogger } from '@aztec/foundation/log';
 
 import { MemoryBlobStore } from '../blobstore/memory_blob_store.js';
+import type { FileStoreBlobClient } from '../filestore/filestore_blob_client.js';
 import { type BlobSinkConfig, hasRemoteBlobSinkSources } from './config.js';
 import { HttpBlobSinkClient } from './http.js';
 import type { BlobSinkClientInterface } from './interface.js';
 import { LocalBlobSinkClient } from './local.js';
 
-export function createBlobSinkClient(config?: BlobSinkConfig, deps?: { logger: Logger }): BlobSinkClientInterface {
+export interface CreateBlobSinkClientDeps {
+  logger?: Logger;
+  /** FileStore clients for reading blobs */
+  fileStoreClients?: FileStoreBlobClient[];
+  /** FileStore client for uploading blobs */
+  fileStoreUploadClient?: FileStoreBlobClient;
+}
+
+export function createBlobSinkClient(
+  config?: BlobSinkConfig,
+  deps?: CreateBlobSinkClientDeps,
+): BlobSinkClientInterface {
   const log = deps?.logger ?? createLogger('blob-sink:client');
   if (!hasRemoteBlobSinkSources(config)) {
     log.info(`Creating local blob sink client.`);
@@ -18,6 +30,12 @@ export function createBlobSinkClient(config?: BlobSinkConfig, deps?: { logger: L
     blobSinkUrl: config?.blobSinkUrl,
     l1ConsensusHostUrls: config?.l1ConsensusHostUrls,
     archiveApiUrl: config?.archiveApiUrl,
+    fileStoreCount: deps?.fileStoreClients?.length ?? 0,
+    hasFileStoreUpload: !!deps?.fileStoreUploadClient,
+  });
+  return new HttpBlobSinkClient(config, {
+    logger: log,
+    fileStoreClients: deps?.fileStoreClients,
+    fileStoreUploadClient: deps?.fileStoreUploadClient,
   });
-  return new HttpBlobSinkClient(config, deps);
 }