[#901] Make scheduled report runs retryable without duplication

drizzle/migrations/0027_scheduled_report_runs.sql

@@ -0,0 +1,26 @@
+CREATE TABLE "scheduled_report_runs" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"scheduled_report_id" uuid NOT NULL,
+	"schedule_key" text NOT NULL,
+	"run_for" timestamp with time zone NOT NULL,
+	"status" text DEFAULT 'pending' NOT NULL,
+	"attempt" integer DEFAULT 0 NOT NULL,
+	"lease_token" text,
+	"lease_until" timestamp with time zone,
+	"started_at" timestamp with time zone,
+	"completed_at" timestamp with time zone,
+	"next_attempt_at" timestamp with time zone,
+	"output_ref" text,
+	"last_error" text,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL,
+	"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
+	CONSTRAINT "scheduled_report_runs_identity_unique" UNIQUE("scheduled_report_id", "run_for")
+);
+--> statement-breakpoint
+ALTER TABLE "scheduled_report_runs" ADD CONSTRAINT "scheduled_report_runs_scheduled_report_id_fk" FOREIGN KEY ("scheduled_report_id") REFERENCES "public"."scheduled_reports"("id") ON DELETE cascade ON UPDATE no action;
+--> statement-breakpoint
+CREATE INDEX "scheduled_report_runs_identity_idx" ON "scheduled_report_runs" USING btree ("scheduled_report_id", "run_for");
+--> statement-breakpoint
+CREATE INDEX "scheduled_report_runs_ready_idx" ON "scheduled_report_runs" USING btree ("status", "next_attempt_at");
+--> statement-breakpoint
+CREATE INDEX "scheduled_report_runs_lease_idx" ON "scheduled_report_runs" USING btree ("status", "lease_until");

src/db/schema.ts

@@ -180,7 +180,6 @@ export const predictions = pgTable("predictions", {
   claimTxHash: text("claim_tx_hash"),
   /** Timestamp when the claim transaction was submitted. Null until claimed. */
   claimedAt: timestamp("claimed_at", { withTimezone: true }),
-  createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
   createdAt: timestamp("created_at", { withTimezone: true })
     .notNull()
     .defaultNow(),
@@ -612,6 +611,58 @@ export const scheduledReports = pgTable(
 export type ScheduledReport = typeof scheduledReports.$inferSelect;
 export type NewScheduledReport = typeof scheduledReports.$inferInsert;
 
+/**
+ * One durable execution attempt for a scheduled report period.
+ *
+ * `scheduleKey` is deliberately stored instead of inferred by workers.  The
+ * unique pair (scheduledReportId, runFor) is the database-level idempotency
+ * boundary: a queue redelivery, process restart, or two dispatchers may all
+ * ask for the same period without creating a second output.
+ */
+export const scheduledReportRuns = pgTable(
+  "scheduled_report_runs",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    scheduledReportId: uuid("scheduled_report_id")
+      .notNull()
+      .references(() => scheduledReports.id, { onDelete: "cascade" }),
+    scheduleKey: text("schedule_key").notNull(),
+    runFor: timestamp("run_for", { withTimezone: true }).notNull(),
+    status: text("status").notNull().default("pending"),
+    attempt: integer("attempt").notNull().default(0),
+    leaseToken: text("lease_token"),
+    leaseUntil: timestamp("lease_until", { withTimezone: true }),
+    startedAt: timestamp("started_at", { withTimezone: true }),
+    completedAt: timestamp("completed_at", { withTimezone: true }),
+    nextAttemptAt: timestamp("next_attempt_at", { withTimezone: true }),
+    outputRef: text("output_ref"),
+    lastError: text("last_error"),
+    createdAt: timestamp("created_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+    updatedAt: timestamp("updated_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+  },
+  (t) => ({
+    scheduledReportRunsIdentityIdx: index("scheduled_report_runs_identity_idx").on(
+      t.scheduledReportId,
+      t.runFor,
+    ),
+    scheduledReportRunsReadyIdx: index("scheduled_report_runs_ready_idx").on(
+      t.status,
+      t.nextAttemptAt,
+    ),
+    scheduledReportRunsLeaseIdx: index("scheduled_report_runs_lease_idx").on(
+      t.status,
+      t.leaseUntil,
+    ),
+  }),
+);
+
+export type ScheduledReportRun = typeof scheduledReportRuns.$inferSelect;
+export type NewScheduledReportRun = typeof scheduledReportRuns.$inferInsert;
+
 // ---------------------------------------------------------------------------
 // Market Watchers
 // ---------------------------------------------------------------------------
@@ -677,4 +728,3 @@ export const referrals = pgTable(
 
 export type Referral = typeof referrals.$inferSelect;
 export type NewReferral = typeof referrals.$inferInsert;
-

src/metrics/registry.ts

@@ -58,6 +58,26 @@ export const indexerLagLedgers = new Gauge({
   registers: [register],
 });
 
+export const scheduledReportRunsTotal = new Counter({
+  name: "scheduled_report_runs_total",
+  help: "Scheduled report runs by terminal outcome",
+  labelNames: ["status"] as const,
+  registers: [register],
+});
+
+export const scheduledReportRetriesTotal = new Counter({
+  name: "scheduled_report_retries_total",
+  help: "Scheduled report retry attempts by reason",
+  labelNames: ["reason"] as const,
+  registers: [register],
+});
+
+export const scheduledReportLeaseConflictsTotal = new Counter({
+  name: "scheduled_report_lease_conflicts_total",
+  help: "Scheduled report jobs skipped because another worker owns the lease",
+  registers: [register],
+});
+
 export const webhookDeliveriesTotal = new Counter({
   name: "webhook_deliveries_total",
   help: "Total number of webhook deliveries, segmented by outcome status (success, failed)",

src/queue/index.ts

@@ -15,6 +15,7 @@ export const webhookQueueName = "webhook-deliveries";
 export const backupVerificationQueueName = "backup-verification";
 export const reconciliationQueueName = "reconciliation";
 export const marketResolutionQueueName = "market-resolution";
+export const scheduledReportQueueName = "scheduled-report-runs";
 
 export const webhookQueue = new Queue(webhookQueueName, {
   //  IORedis types conflict with BullMQ
@@ -36,4 +37,9 @@ export const marketResolutionQueue = new Queue(marketResolutionQueueName, {
   connection: redisConnection,
 });
 
+export const scheduledReportQueue = new Queue(scheduledReportQueueName, {
+  // IORedis types conflict with BullMQ
+  connection: redisConnection,
+});
+
 export { Queue, Worker, QueueEvents };