Blob filestores

Author: mrzeszutko

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

@@ -882,6 +882,7 @@ export class Archiver extends (EventEmitter as new () => ArchiverEmitter) implem
         this.l1Addresses,
         this.instrumentation,
         this.log,
+        !this.initialSyncComplete, // isHistoricalSync
       );
 
       if (retrievedCheckpoints.length === 0) {

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

@@ -138,13 +138,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>,
@@ -160,6 +164,7 @@ export async function retrieveCheckpointsFromRollup(
   },
   instrumentation: ArchiverInstrumentation,
   logger: Logger = createLogger('archiver'),
+  isHistoricalSync: boolean = false,
 ): Promise<RetrievedCheckpoint[]> {
   const retrievedCheckpoints: RetrievedCheckpoint[] = [];
 
@@ -211,6 +216,7 @@ export async function retrieveCheckpointsFromRollup(
       contractAddresses,
       instrumentation,
       logger,
+      isHistoricalSync,
     );
     retrievedCheckpoints.push(...newCheckpoints);
     searchStartBlock = lastLog.blockNumber! + 1n;
@@ -222,11 +228,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>,
@@ -242,6 +254,7 @@ async function processCheckpointProposedLogs(
   },
   instrumentation: ArchiverInstrumentation,
   logger: Logger,
+  isHistoricalSync: boolean,
 ): Promise<RetrievedCheckpoint[]> {
   const retrievedCheckpoints: RetrievedCheckpoint[] = [];
   const calldataRetriever = new CalldataRetriever(
@@ -272,6 +285,7 @@ async function processCheckpointProposedLogs(
         blobHashes,
         checkpointNumber,
         logger,
+        isHistoricalSync,
       );
 
       const l1: L1PublishedData = {
@@ -309,8 +323,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

@@ -9,7 +9,8 @@
     "./server": "./dest/server/index.js",
     "./client": "./dest/client/index.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);
 }