feat: improve cron observability

Author: onmax

README.md

@@ -213,6 +213,22 @@ Where `env`: `preview`, `testnet`, or `testnet-preview` (omit for mainnet produc
 
 **Required secrets:** `ALBATROSS_RPC_NODE_URL`, `NUXT_SLACK_WEBHOOK_URL`
 
+### D1 Migrations
+
+When adding a new SQL migration under `server/db/migrations/`, apply it to the remote D1 database.
+
+For the `cron_runs` table:
+
+```bash
+pnpm db:apply:cron-runs:mainnet
+```
+
+Testnet:
+
+```bash
+pnpm db:apply:cron-runs:testnet
+```
+
 **Environments** (configured in `wrangler.json`):
 
 | Environment       | Dashboard URL                                                                  | Trigger            |

nuxt.config.ts

@@ -148,8 +148,8 @@ export default defineNuxtConfig({
       tasks: true,
     },
     scheduledTasks: {
-      // 12-hour sync: fetch missing epochs and update validator snapshots
-      '0 */12 * * *': ['sync:epochs', 'sync:snapshot'],
+      // 12-hour sync: wrapper task records run + executes sync tasks
+      '0 */12 * * *': ['cron:sync'],
     },
     openAPI: {
       meta: { title: name, description, version },

package.json

@@ -25,6 +25,8 @@
     "test": "vitest",
     "db:generate": "drizzle-kit generate",
     "db:delete": "drizzle-kit drop",
+    "db:apply:cron-runs:mainnet": "wrangler d1 execute validators-api-mainnet --remote --yes --command \"CREATE TABLE IF NOT EXISTS cron_runs (id integer PRIMARY KEY AUTOINCREMENT NOT NULL, cron text NOT NULL, network text NOT NULL, git_branch text, started_at text NOT NULL, finished_at text, status text NOT NULL, error_message text, meta text);\" && wrangler d1 execute validators-api-mainnet --remote --yes --command \"CREATE INDEX IF NOT EXISTS idx_cron_runs_started_at ON cron_runs (started_at);\"",
+    "db:apply:cron-runs:testnet": "wrangler d1 execute validators-api-testnet --remote --yes --command \"CREATE TABLE IF NOT EXISTS cron_runs (id integer PRIMARY KEY AUTOINCREMENT NOT NULL, cron text NOT NULL, network text NOT NULL, git_branch text, started_at text NOT NULL, finished_at text, status text NOT NULL, error_message text, meta text);\" && wrangler d1 execute validators-api-testnet --remote --yes --command \"CREATE INDEX IF NOT EXISTS idx_cron_runs_started_at ON cron_runs (started_at);\"",
     "release": "bumpp -r && nr -r publish",
     "validate:json-files": "tsx scripts/validate-json-files.ts"
   },

server/db/migrations/0003_add_cron_runs.sql

@@ -0,0 +1,14 @@
+CREATE TABLE `cron_runs` (
+	`id` integer PRIMARY KEY AUTOINCREMENT NOT NULL,
+	`cron` text NOT NULL,
+	`network` text NOT NULL,
+	`git_branch` text,
+	`started_at` text NOT NULL,
+	`finished_at` text,
+	`status` text NOT NULL,
+	`error_message` text,
+	`meta` text
+);
+--> statement-breakpoint
+CREATE INDEX `idx_cron_runs_started_at` ON `cron_runs` (`started_at`);
+

server/db/schema.ts

@@ -59,6 +59,20 @@ export const activity = sqliteTable('activity', {
   primaryKey({ columns: [table.validatorId, table.epochNumber] }),
 ])
 
+export const cronRuns = sqliteTable('cron_runs', {
+  id: integer('id').primaryKey({ autoIncrement: true }),
+  cron: text('cron').notNull(),
+  network: text('network').notNull(),
+  gitBranch: text('git_branch'),
+  startedAt: text('started_at').notNull(),
+  finishedAt: text('finished_at'),
+  status: text('status').notNull(),
+  errorMessage: text('error_message'),
+  meta: text('meta', { mode: 'json' }),
+}, table => [
+  index('idx_cron_runs_started_at').on(table.startedAt),
+])
+
 export const activityRelations = relations(activity, ({ one }) => ({
   validator: one(validators, {
     fields: [activity.validatorId],

server/tasks/cron/sync.ts

@@ -0,0 +1,111 @@
+import type { TaskEvent } from 'nitropack'
+import { consola } from 'consola'
+import { runTask } from 'nitropack/runtime'
+import { eq, tables, useDrizzle } from '~~/server/utils/drizzle'
+
+const CRON_EXPRESSION = '0 */12 * * *'
+const TASKS: string[] = ['sync:epochs', 'sync:snapshot']
+
+function toErrorString(error: unknown) {
+  if (error instanceof Error)
+    return `${error.message}${error.stack ? `\n${error.stack}` : ''}`
+  return String(error)
+}
+
+export default defineTask({
+  meta: {
+    name: 'cron:sync',
+    description: 'Main scheduled sync wrapper (records cron run to D1)',
+  },
+  async run(event: TaskEvent) {
+    const config = useSafeRuntimeConfig()
+    const { nimiqNetwork, gitBranch } = config.public
+
+    const startedAt = new Date().toISOString()
+
+    let cronRunId: number | undefined
+    try {
+      cronRunId = await useDrizzle()
+        .insert(tables.cronRuns)
+        .values({
+          cron: CRON_EXPRESSION,
+          network: nimiqNetwork,
+          gitBranch,
+          startedAt,
+          status: 'started',
+          meta: {
+            tasks: TASKS,
+            payload: event.payload ?? {},
+          },
+        })
+        .returning({ id: tables.cronRuns.id })
+        .get()
+        .then(r => r.id)
+    }
+    catch (error) {
+      consola.warn('[cron:sync] unable to record start (missing migration?)', error)
+    }
+
+    try {
+      const results: Record<string, unknown> = {}
+
+      for (const taskName of TASKS) {
+        consola.info(`[cron:sync] running ${taskName}`)
+        const res = await runTask(taskName, { payload: event.payload ?? {}, context: event.context ?? {} })
+        results[taskName] = (res as any)?.result ?? res
+      }
+
+      if (cronRunId) {
+        try {
+          await useDrizzle()
+            .update(tables.cronRuns)
+            .set({
+              finishedAt: new Date().toISOString(),
+              status: 'success',
+              meta: {
+                cron: CRON_EXPRESSION,
+                network: nimiqNetwork,
+                tasks: TASKS,
+                results,
+              },
+            })
+            .where(eq(tables.cronRuns.id, cronRunId))
+            .execute()
+        }
+        catch (error) {
+          consola.warn('[cron:sync] unable to record success', error)
+        }
+      }
+
+      return { result: { success: true, cronRunId } }
+    }
+    catch (error) {
+      const errorMessage = toErrorString(error)
+      consola.error('[cron:sync] failed', error)
+
+      if (cronRunId) {
+        try {
+          await useDrizzle()
+            .update(tables.cronRuns)
+            .set({
+              finishedAt: new Date().toISOString(),
+              status: 'error',
+              errorMessage,
+              meta: {
+                cron: CRON_EXPRESSION,
+                network: nimiqNetwork,
+                tasks: TASKS,
+              },
+            })
+            .where(eq(tables.cronRuns.id, cronRunId))
+            .execute()
+        }
+        catch (error) {
+          consola.warn('[cron:sync] unable to record error', error)
+        }
+      }
+
+      throw error
+    }
+  },
+})

server/utils/drizzle.ts

@@ -14,3 +14,5 @@ export type Activity = typeof schema.activity.$inferSelect
 export type NewActivity = typeof schema.activity.$inferInsert
 export type Score = typeof schema.scores.$inferSelect
 export type NewScore = typeof schema.scores.$inferInsert
+export type CronRun = typeof schema.cronRuns.$inferSelect
+export type NewCronRun = typeof schema.cronRuns.$inferInsert

wrangler.json

@@ -6,6 +6,13 @@
   "account_id": "cf9baad7d68d7ee717f3339731e81dfb",
   "compatibility_date": "2025-01-01",
   "compatibility_flags": ["nodejs_compat"],
+  "observability": {
+    "enabled": true,
+    "logs": {
+      "enabled": false,
+      "invocation_logs": true
+    }
+  },
   "triggers": { "crons": ["0 */12 * * *"] },
   "vars": { "NUXT_PUBLIC_NIMIQ_NETWORK": "main-albatross" },
   "d1_databases": [{ "binding": "DB", "database_id": "cc9f1d25-676b-4cb3-8af6-887e85a08baa", "database_name": "validators-api-mainnet" }],