Merge pull request #411 from Blobscan/swarm-syncer

Swarm batch information updater

.changeset/cold-terms-accept.md

@@ -0,0 +1,5 @@
+---
+"@blobscan/rest-api-server": minor
+---
+
+Added Swarm stamp syncer

.changeset/fair-rabbits-cross.md

@@ -0,0 +1,5 @@
+---
+"@blobscan/syncers": minor
+---
+
+Added swarm stamp syncer

.changeset/long-turkeys-impress.md

@@ -0,0 +1,11 @@
+---
+"@blobscan/rest-api-server": minor
+"@blobscan/syncers": minor
+---
+
+Refactored the stats syncer package to support general-purpose synchronization workers/queues.
+
+Key changes include:
+
+    •	Renamed the package to syncers.
+    •	Exported each syncer directly, removing the StatsSyncer managing entity.

.changeset/moody-fireants-itch.md

@@ -0,0 +1,5 @@
+---
+"@blobscan/db": minor
+---
+
+Added an updated at field to blob storages state model

.env.test

@@ -20,7 +20,7 @@ GOOGLE_STORAGE_ENABLED=true
 # GOOGLE_SERVICE_KEY=
 
 BEE_ENDPOINT=http://localhost:1633
-
+SWARM_BATCH_ID=f89e63edf757f06e89933761d6d46592d03026efb9871f9d244f34da86b6c242
 
 FILE_SYSTEM_STORAGE_PATH=test-blobscan-blobs
 

.nvmrc

@@ -1 +1 @@
-18.15.0
+20.12.2

apps/docs/src/app/docs/environment/page.md

@@ -65,10 +65,13 @@ At the moment Postgres is the default storage and Blobscan won't be able to run
 
 **Ethereum Swarm**
 
-| Variable                | Description          | Required | Default value |
-| ----------------------- | -------------------- | -------- | ------------- |
-| `SWARM_STORAGE_ENABLED` | Store blobs in Swarm | No       | `false`       |
-| `BEE_ENDPOINT`          | Bee endpoint         | No       | (empty)       |
+| Variable                   | Description                | Required                        | Default value  |
+| -------------------------- | -------------------------- | ------------------------------- | -------------- |
+| `SWARM_STORAGE_ENABLED`    | Store blobs in Swarm       | No                              | `false`        |
+| `SWARM_BATCH_ID`           | Swarm address of the stamp | If `SWARM_STORAGE_ENABLED=true` | (empty)        |
+| `SWARM_STAMP_CRON_PATTERN` | Cron pattern for swarm job | No                              | `*/15 * * * *` |
+| `BEE_ENDPOINT`             | Bee endpoint               | No                              | (empty)        |
+| `BEE_DEBUG_ENDPOINT`       | Bee debug endpoint         | No                              | (empty)        |
 
 ## Blob propagator
 

apps/rest-api-server/package.json

@@ -18,7 +18,7 @@
     "@blobscan/api": "workspace:^0.9.0",
     "@blobscan/logger": "workspace:^0.1.0",
     "@blobscan/open-telemetry": "workspace:^0.0.7",
-    "@blobscan/stats-syncer": "workspace:^0.1.8",
+    "@blobscan/syncers": "workspace:^0.1.9",
     "@blobscan/zod": "workspace:^0.1.0",
     "@opentelemetry/instrumentation-express": "^0.33.0",
     "@sentry/node": "^7.109.0",

apps/rest-api-server/src/env.ts

@@ -11,6 +11,9 @@ import {
 export const env = createEnv({
   envOptions: {
     server: {
+      // FIXME
+      // BEE_ENDPOINT: requiredStorageConfigSchema("SWARM", z.string().url()),
+      BEE_ENDPOINT: z.string().optional(),
       BLOBSCAN_API_BASE_URL: z
         .string()
         .url()
@@ -22,8 +25,11 @@ export const env = createEnv({
       METRICS_ENABLED: booleanSchema.default("false"),
       REDIS_URI: z.string().default("redis://localhost:6379"),
       DENCUN_FORK_SLOT: z.coerce.number().optional(),
+      SWARM_STAMP_CRON_PATTERN: z.string().default("*/15 * * * *"),
       STATS_SYNCER_DAILY_CRON_PATTERN: z.string().default("30 0 * * * *"),
       STATS_SYNCER_OVERALL_CRON_PATTERN: z.string().default("*/15 * * * *"),
+      SWARM_BATCH_ID: z.string().optional(),
+      SWARM_STORAGE_ENABLED: booleanSchema.default("false"),
       SENTRY_DSN_API: z.string().url().optional(),
     },
 

apps/rest-api-server/src/index.ts

@@ -14,28 +14,16 @@ import {
   gracefulShutdown as apiGracefulShutdown,
 } from "@blobscan/api";
 import { collectDefaultMetrics } from "@blobscan/open-telemetry";
-import { StatsSyncer } from "@blobscan/stats-syncer";
 
 import { env } from "./env";
 import { logger } from "./logger";
 import { morganMiddleware } from "./morgan";
 import { openApiDocument } from "./openapi";
-import { getNetworkDencunForkSlot } from "./utils";
+import { setUpSyncers } from "./syncers";
 
 collectDefaultMetrics();
 
-const statsSyncer = new StatsSyncer({
-  redisUri: env.REDIS_URI,
-  lowestSlot:
-    env.DENCUN_FORK_SLOT ?? getNetworkDencunForkSlot(env.NETWORK_NAME),
-});
-
-statsSyncer.start({
-  cronPatterns: {
-    daily: env.STATS_SYNCER_DAILY_CRON_PATTERN,
-    overall: env.STATS_SYNCER_OVERALL_CRON_PATTERN,
-  },
-});
+const closeSyncers = setUpSyncers();
 
 const app = express();
 
@@ -73,7 +61,9 @@ async function gracefulShutdown(signal: string) {
   logger.debug(`Received ${signal}. Shutting down...`);
 
   await apiGracefulShutdown()
-    .finally(() => statsSyncer.close())
+    .finally(async () => {
+      await closeSyncers();
+    })
     .finally(() => {
       server.close(() => {
         logger.debug("Server shut down successfully");
@@ -82,7 +72,11 @@ async function gracefulShutdown(signal: string) {
 }
 
 // Listen for TERM signal .e.g. kill
-process.on("SIGTERM", () => void gracefulShutdown("SIGTERM"));
+process.on("SIGTERM", async () => {
+  await gracefulShutdown("SIGTERM");
+});
 
 // Listen for INT signal e.g. Ctrl-C
-process.on("SIGINT", () => void gracefulShutdown("SIGINT"));
+process.on("SIGINT", async () => {
+  await gracefulShutdown("SIGINT");
+});

apps/rest-api-server/src/syncers.ts

@@ -0,0 +1,61 @@
+import type { BaseSyncer } from "@blobscan/syncers";
+import {
+  DailyStatsSyncer,
+  OverallStatsSyncer,
+  SwarmStampSyncer,
+  createRedisConnection,
+} from "@blobscan/syncers";
+
+import { env } from "./env";
+import { logger } from "./logger";
+import { getNetworkDencunForkSlot } from "./utils";
+
+export function setUpSyncers() {
+  const connection = createRedisConnection(env.REDIS_URI);
+  const syncers: BaseSyncer[] = [];
+
+  if (env.SWARM_STORAGE_ENABLED) {
+    if (!env.SWARM_BATCH_ID) {
+      logger.error(`Can't initialize Swarm stamp job: no batch ID provided`);
+    } else if (!env.BEE_ENDPOINT) {
+      logger.error("Can't initialize Swarm stamp job: no Bee endpoint provided");
+    } else {
+      syncers.push(
+        new SwarmStampSyncer({
+          cronPattern: env.SWARM_STAMP_CRON_PATTERN,
+          redisUriOrConnection: connection,
+          batchId: env.SWARM_BATCH_ID,
+          beeEndpoint: env.BEE_ENDPOINT,
+        })
+      );
+    }
+  }
+
+  syncers.push(
+    new DailyStatsSyncer({
+      cronPattern: env.STATS_SYNCER_DAILY_CRON_PATTERN,
+      redisUriOrConnection: connection,
+    })
+  );
+
+  syncers.push(
+    new OverallStatsSyncer({
+      cronPattern: env.STATS_SYNCER_OVERALL_CRON_PATTERN,
+      redisUriOrConnection: connection,
+      lowestSlot:
+        env.DENCUN_FORK_SLOT ?? getNetworkDencunForkSlot(env.NETWORK_NAME),
+    })
+  );
+
+  Promise.all(syncers.map((syncer) => syncer.start()));
+
+  return () => {
+    let teardownPromise = Promise.resolve();
+
+    for (const syncer of syncers) {
+      teardownPromise = teardownPromise.finally(() => syncer.close());
+    }
+
+    return teardownPromise;
+  };
+}

package.json

@@ -50,6 +50,7 @@
     "@vitest/coverage-v8": "^0.34.3",
     "@vitest/ui": "^0.34.1",
     "dotenv-cli": "^7.2.1",
+    "msw": "^2.3.1",
     "prettier": "^2.8.8",
     "prettier-plugin-tailwindcss": "^0.2.8",
     "ts-node": "^10.9.1",

packages/db/prisma/migrations/20240613112056_add_updated_at_field/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "blob_storages_state" ADD COLUMN     "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP;

packages/db/prisma/schema.prisma

@@ -53,10 +53,12 @@ model BlockchainSyncState {
   @@map("blockchain_sync_state")
 }
 
+//TODO: Rename to SwarmBatchStorage
 model BlobStoragesState {
-  id           Int     @id @default(autoincrement())
-  swarmDataId  String? @map("swarm_data_id")
-  swarmDataTTL Int?    @map("swarm_data_ttl")
+  id            Int      @id @default(autoincrement())
+  swarmDataId   String?  @map("swarm_data_id") // TODO: rename to batchId
+  swarmDataTTL  Int?     @map("swarm_data_ttl") // TODO: rename to batchTtl
+  updatedAt     DateTime @default(now()) @map("updated_at")
 
   @@map("blob_storages_state")
 }
@@ -259,7 +261,7 @@ model BlobDailyStats {
 }
 
 // NextAuth.js Models
-// NOTE: When using postgresql, mysql or sqlserver, 
+// NOTE: When using postgresql, mysql or sqlserver,
 // uncomment the @db.Text annotations below
 // @see https://next-auth.js.org/schemas/models
 // model Account {