fix: Use R2 for snapshots (#1943)

## Motivation

Upload / download snapshots from R2 instead of S3


## Merge Checklist

_Choose all relevant options below by adding an `x` now or at any time
before submitting for review_

- [X] PR title adheres to the [conventional
commits](https://www.conventionalcommits.org/en/v1.0.0/) standard
- [X] PR has a
[changeset](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#35-adding-changesets)
- [ ] PR has been tagged with a change label(s) (i.e. documentation,
feature, bugfix, or chore)
- [ ] PR includes
[documentation](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#32-writing-docs)
if necessary.
- [X] All [commits have been
signed](https://github.com/farcasterxyz/hub-monorepo/blob/main/CONTRIBUTING.md#22-signing-commits)

<!-- start pr-codex -->

---

## PR-Codex overview
This PR updates the Hubble app to use R2 for snapshots and introduces a
new `SNAPSHOT_S3_UPLOAD_BUCKET`.

### Detailed summary
- Updated snapshot handling to use R2
- Added `SNAPSHOT_S3_UPLOAD_BUCKET` for uploading snapshots
- Introduced `r2Endpoint` function for R2 endpoint configuration

> ✨ Ask PR-Codex anything about this PR by commenting with `/codex {your
question}`

<!-- end pr-codex -->

.changeset/shiny-carrots-behave.md

@@ -0,0 +1,5 @@
+---
+"@farcaster/hubble": patch
+---
+
+fix: Use R2 for snapshots

apps/hubble/src/cli.ts

@@ -9,7 +9,15 @@ import { mkdir, readFile, writeFile } from "fs/promises";
 import { Result, ResultAsync } from "neverthrow";
 import { dirname, resolve } from "path";
 import { exit } from "process";
-import { APP_VERSION, FARCASTER_VERSION, Hub, HubOptions, HubShutdownReason, S3_REGION } from "./hubble.js";
+import {
+  APP_VERSION,
+  FARCASTER_VERSION,
+  Hub,
+  HubOptions,
+  HubShutdownReason,
+  S3_REGION,
+  SNAPSHOT_S3_UPLOAD_BUCKET,
+} from "./hubble.js";
 import { logger } from "./utils/logger.js";
 import { addressInfoFromParts, hostPortFromString, ipMultiAddrStrFromAddressInfo, parseAddress } from "./utils/p2p.js";
 import { DEFAULT_RPC_CONSOLE, startConsole } from "./console/console.js";
@@ -25,10 +33,10 @@ import { startupCheck, StartupCheckStatus } from "./utils/startupCheck.js";
 import { mainnet, optimism } from "viem/chains";
 import { finishAllProgressBars } from "./utils/progressBars.js";
 import { MAINNET_BOOTSTRAP_PEERS } from "./bootstrapPeers.mainnet.js";
-import { GetCallerIdentityCommand, STSClient } from "@aws-sdk/client-sts";
 import axios from "axios";
-import { snapshotURLAndMetadata } from "./utils/snapshot.js";
+import { r2Endpoint, snapshotURLAndMetadata } from "./utils/snapshot.js";
 import { DEFAULT_DIAGNOSTIC_REPORT_URL, initDiagnosticReporter } from "./utils/diagnosticReport.js";
+import { ListObjectsV2Command, S3Client } from "@aws-sdk/client-s3";
 
 /** A CLI to accept options from the user and start the Hub */
 
@@ -942,16 +950,24 @@ app.parse(process.argv);
 // Verify that we have access to the AWS credentials.
 // Either via environment variables or via the AWS credentials file
 async function verifyAWSCredentials(): Promise<boolean> {
-  const sts = new STSClient({ region: S3_REGION });
+  const s3 = new S3Client({
+    region: S3_REGION,
+    endpoint: r2Endpoint(),
+    forcePathStyle: true,
+  });
 
   try {
-    const identity = await sts.send(new GetCallerIdentityCommand({}));
+    const params = {
+      Bucket: SNAPSHOT_S3_UPLOAD_BUCKET,
+      Prefix: "snapshots/",
+    };
 
-    logger.info({ accountId: identity.Account }, "Verified AWS credentials");
+    const result = await s3.send(new ListObjectsV2Command(params));
+    logger.info({ keys: result.KeyCount }, "Verified R2 credentials for snapshots");
 
     return true;
   } catch (error) {
-    logger.error({ err: error }, "Failed to verify AWS credentials. No S3 snapshot upload will be performed.");
+    logger.error({ err: error }, "Failed to verify R2 credentials. No snapshots performed.");
     return false;
   }
 }

apps/hubble/src/hubble.ts

@@ -100,8 +100,9 @@ export type HubSubmitSource = "gossip" | "rpc" | "eth-provider" | "l2-provider"
 export const APP_VERSION = packageJson.version;
 export const APP_NICKNAME = process.env["HUBBLE_NAME"] ?? "Farcaster Hub";
 
+export const SNAPSHOT_S3_UPLOAD_BUCKET = "farcaster-snapshots";
 export const SNAPSHOT_S3_DEFAULT_BUCKET = "download.farcaster.xyz";
-export const S3_REGION = "us-east-1";
+export const S3_REGION = "auto";
 
 export const FARCASTER_VERSION = "2024.3.20";
 export const FARCASTER_VERSIONS_SCHEDULE: VersionSchedule[] = [

apps/hubble/src/storage/jobs/dbSnapshotBackupJob.ts

@@ -8,7 +8,7 @@ import { uploadToS3 } from "../../utils/snapshot.js";
 import SyncEngine from "../../network/sync/syncEngine.js";
 import { S3Client, ListObjectsV2Command, DeleteObjectsCommand } from "@aws-sdk/client-s3";
 import fs from "fs";
-import { HubOptions, S3_REGION, SNAPSHOT_S3_DEFAULT_BUCKET } from "../../hubble.js";
+import { HubOptions, S3_REGION, SNAPSHOT_S3_UPLOAD_BUCKET } from "../../hubble.js";
 
 export const DEFAULT_DB_SNAPSHOT_BACKUP_JOB_CRON = "15 2 * * *"; // 2:15 am everyday
 
@@ -56,6 +56,7 @@ export class DbSnapshotBackupJobScheduler {
 
   async doJobs(): HubAsyncResult<void> {
     if (!this._options.enableSnapshotToS3) {
+      log.info({}, "Db Snapshot Backup job disabled, skipping");
       return ok(undefined);
     }
 
@@ -145,9 +146,8 @@ export class DbSnapshotBackupJobScheduler {
 
       log.warn({ oldFiles }, "Deleting old snapshot files from S3");
 
-      const s3Bucket = this._options.s3SnapshotBucket ?? SNAPSHOT_S3_DEFAULT_BUCKET;
       const deleteParams = {
-        Bucket: s3Bucket,
+        Bucket: SNAPSHOT_S3_UPLOAD_BUCKET,
         Delete: {
           Objects: oldFiles.map((file) => ({ Key: file.Key })),
         },
@@ -179,9 +179,8 @@ export class DbSnapshotBackupJobScheduler {
 
     // Note: We get the snapshots across all DB_SCHEMA versions
     // when determining which snapshots to delete, we only delete snapshots from the current DB_SCHEMA version
-    const s3Bucket = this._options.s3SnapshotBucket ?? SNAPSHOT_S3_DEFAULT_BUCKET;
     const params = {
-      Bucket: s3Bucket,
+      Bucket: SNAPSHOT_S3_UPLOAD_BUCKET,
       Prefix: `snapshots/${network}/`,
     };
 

apps/hubble/src/utils/snapshot.ts

@@ -2,7 +2,7 @@ import { DbStats, FarcasterNetwork, HubAsyncResult, HubError } from "@farcaster/
 import { LATEST_DB_SCHEMA_VERSION } from "../storage/db/migrations/migrations.js";
 import axios from "axios";
 import { err, ok, ResultAsync } from "neverthrow";
-import { S3_REGION, SNAPSHOT_S3_DEFAULT_BUCKET } from "../hubble.js";
+import { S3_REGION, SNAPSHOT_S3_DEFAULT_BUCKET, SNAPSHOT_S3_UPLOAD_BUCKET } from "../hubble.js";
 import { PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
 import fs from "fs";
 import { Upload } from "@aws-sdk/lib-storage";
@@ -76,17 +76,23 @@ export const snapshotURL = (
   return `https://${s3Bucket}/${snapshotDirectory(fcNetwork, prevVersionCounter)}`;
 };
 
+export const r2Endpoint = (): string => {
+  return process.env["R2_ENDPOINT"] ?? "";
+};
+
 export const uploadToS3 = async (
   fcNetwork: FarcasterNetwork,
   chunkedDirPath: string,
-  s3Bucket: string = SNAPSHOT_S3_DEFAULT_BUCKET,
+  s3Bucket: string = SNAPSHOT_S3_UPLOAD_BUCKET,
   messageCount?: number,
   timestamp?: number,
 ): HubAsyncResult<string> => {
   const startTimestamp = timestamp ?? Date.now();
 
   const s3 = new S3Client({
     region: S3_REGION,
+    endpoint: r2Endpoint(),
+    forcePathStyle: true,
   });
 
   // The AWS key is "snapshots/{network}/{DB_SCHEMA_VERSION}/snapshot-{yyyy-mm-dd}-{timestamp}.tar.gz"