Run the rollup backfill scan from the worker instead of the scheduler

apps/scheduler/src/schedulers/rollupBackfillScheduler.ts

@@ -1,66 +1,9 @@
-import { ListObjectsV2Command } from "@aws-sdk/client-s3";
-import { hoursToMilliseconds, minutesToMilliseconds } from "date-fns";
-import {
-  S3_BUCKET,
-  SNAPSHOT_RETENTION_HOURS,
-  addUtcDays,
-  formatUtcDay,
-  getEnvInt,
-  getS3Client,
-  parseUtcDay,
-  scheduleRollupBackfill,
-} from "@teerank/teerank";
+import { minutesToMilliseconds } from "date-fns";
+import { scheduleRollupBackfill } from "@teerank/teerank";
 import { schedule } from "../utils";
-import { prisma } from "../prisma";
-
-const ROLLUP_BACKFILL_DAYS_PER_TICK = getEnvInt('ROLLUP_BACKFILL_DAYS_PER_TICK', 4);
-
-async function listArchivedDays() {
-  const s3 = getS3Client();
-  const days: string[] = [];
-  let continuationToken: string | undefined;
-
-  do {
-    const result = await s3.send(new ListObjectsV2Command({
-      Bucket: S3_BUCKET,
-      Prefix: 'snapshots/',
-      Delimiter: '/',
-      ContinuationToken: continuationToken,
-    }));
-
-    for (const prefix of result.CommonPrefixes ?? []) {
-      const match = prefix.Prefix?.match(/dt=(\d{4}-\d{2}-\d{2})\/$/);
-      if (match !== null && match !== undefined) {
-        days.push(match[1]);
-      }
-    }
-
-    continuationToken = result.NextContinuationToken;
-  } while (continuationToken !== undefined);
-
-  return days.sort();
-}
 
 export function rollupBackfillScheduler() {
   schedule(minutesToMilliseconds(15), async () => {
-    const days = (await listArchivedDays()).slice(0, -1).filter((day) => {
-      const dayEndMs = addUtcDays(parseUtcDay(day), 1).getTime();
-      return dayEndMs + hoursToMilliseconds(SNAPSHOT_RETENTION_HOURS) <= Date.now();
-    });
-
-    if (days.length === 0) {
-      return;
-    }
-
-    const rolledUpDays = await prisma.playerDay.groupBy({
-      by: ['day'],
-    });
-    const rolledUp = new Set(rolledUpDays.map(({ day }) => formatUtcDay(day)));
-
-    const missing = days.filter((day) => !rolledUp.has(day)).slice(0, ROLLUP_BACKFILL_DAYS_PER_TICK);
-
-    for (const day of missing) {
-      await scheduleRollupBackfill({ day });
-    }
+    await scheduleRollupBackfill();
   });
 }

apps/worker/src/workers/rollupBackfill.ts

@@ -1,21 +1,62 @@
 import { GetObjectCommand, ListObjectsV2Command } from "@aws-sdk/client-s3";
 import { hoursToMilliseconds } from "date-fns";
 import {
-  RollupBackfillJobData,
   S3_BUCKET,
   SNAPSHOT_RETENTION_HOURS,
   addUtcDays,
+  formatUtcDay,
   getEnvInt,
   getS3Client,
   parseUtcDay,
   processRollupBackfillJobs,
 } from "@teerank/teerank";
+import { prisma } from "../prisma";
 import { SnapshotArchiveRow, decodeSnapshotRowsFromParquet } from "../parquet";
 import { DayAggregator, RollupSnapshot } from "../rollup/aggregateDay";
 import { writeDayRollup } from "../rollup/writeDayRollup";
-import { isDayRolledUp } from "./rollupDay";
 
 const ROLLUP_TIME_BUDGET_MS = getEnvInt('ROLLUP_TIME_BUDGET_MS', 10 * 60 * 1000);
+const ROLLUP_BACKFILL_DAYS_PER_TICK = getEnvInt('ROLLUP_BACKFILL_DAYS_PER_TICK', 4);
+
+async function listArchivedDays() {
+  const s3 = getS3Client();
+  const days: string[] = [];
+  let continuationToken: string | undefined;
+
+  do {
+    const result = await s3.send(new ListObjectsV2Command({
+      Bucket: S3_BUCKET,
+      Prefix: 'snapshots/',
+      Delimiter: '/',
+      ContinuationToken: continuationToken,
+    }));
+
+    for (const prefix of result.CommonPrefixes ?? []) {
+      const match = prefix.Prefix?.match(/dt=(\d{4}-\d{2}-\d{2})\/$/);
+      if (match !== null && match !== undefined) {
+        days.push(match[1]);
+      }
+    }
+
+    continuationToken = result.NextContinuationToken;
+  } while (continuationToken !== undefined);
+
+  return days.sort();
+}
+
+async function listMissingArchivedDays() {
+  const days = (await listArchivedDays()).slice(0, -1).filter((day) => {
+    const dayEndMs = addUtcDays(parseUtcDay(day), 1).getTime();
+    return dayEndMs + hoursToMilliseconds(SNAPSHOT_RETENTION_HOURS) <= Date.now();
+  });
+
+  const rolledUpDays = await prisma.playerDay.groupBy({
+    by: ['day'],
+  });
+  const rolledUp = new Set(rolledUpDays.map(({ day }) => formatUtcDay(day)));
+
+  return days.filter((day) => !rolledUp.has(day));
+}
 
 async function listDayObjectKeys(day: string) {
   const s3 = getS3Client();
@@ -79,27 +120,15 @@ function addArchiveRows(aggregator: DayAggregator, rows: SnapshotArchiveRow[], d
   }
 }
 
-export async function rollupBackfill(data: RollupBackfillJobData) {
+async function backfillDay(dayLabel: string) {
   const startedAt = Date.now();
-  const day = parseUtcDay(data.day);
+  const day = parseUtcDay(dayLabel);
   const dayEnd = addUtcDays(day, 1);
 
-  // The archive only holds all of a day's snapshots once the retention window
-  // has moved past the day's end.
-  if (dayEnd.getTime() + hoursToMilliseconds(SNAPSHOT_RETENTION_HOURS) > Date.now()) {
-    console.log(`Backfill for ${data.day} skipped: day may not be fully archived`);
-    return;
-  }
-
-  if (await isDayRolledUp(day)) {
-    console.log(`Backfill for ${data.day} skipped: already rolled up`);
-    return;
-  }
-
-  const keys = await listDayObjectKeys(data.day);
+  const keys = await listDayObjectKeys(dayLabel);
 
   if (keys.length === 0) {
-    console.log(`Backfill for ${data.day} skipped: no archive objects`);
+    console.log(`Backfill for ${dayLabel} skipped: no archive objects`);
     return;
   }
 
@@ -108,7 +137,7 @@ export async function rollupBackfill(data: RollupBackfillJobData) {
 
   for (const key of keys) {
     if (Date.now() - startedAt > ROLLUP_TIME_BUDGET_MS) {
-      throw new Error(`Backfill for ${data.day} exceeded time budget, nothing written`);
+      throw new Error(`Backfill for ${dayLabel} exceeded time budget, nothing written`);
     }
 
     const object = await s3.send(new GetObjectCommand({ Bucket: S3_BUCKET, Key: key }));
@@ -124,6 +153,14 @@ export async function rollupBackfill(data: RollupBackfillJobData) {
   await writeDayRollup(day, aggregator.finalize());
 }
 
+export async function rollupBackfill() {
+  const missing = await listMissingArchivedDays();
+
+  for (const day of missing.slice(0, ROLLUP_BACKFILL_DAYS_PER_TICK)) {
+    await backfillDay(day);
+  }
+}
+
 export async function startRollupBackfillWorker() {
   return processRollupBackfillJobs(rollupBackfill);
 }

libs/teerank/src/lib/bullmq/queueRollupBackfill.ts

@@ -1,8 +1,6 @@
 import { Job, Queue, Worker } from "bullmq";
 import { bullmqConnection } from "./config";
-import { z } from "zod";
 import { hoursToSeconds } from "date-fns";
-import { utcDaySchema } from "../schemas";
 
 let rollupBackfillQueue: Queue | null = null;
 
@@ -13,25 +11,18 @@ function getQueueRollupBackfill() {
   return rollupBackfillQueue;
 }
 
-const schema = z.object({
-  day: utcDaySchema,
-});
-
-export type RollupBackfillJobData = z.infer<typeof schema>;
-
-export async function scheduleRollupBackfill(data: RollupBackfillJobData) {
+export async function scheduleRollupBackfill() {
   const queue = getQueueRollupBackfill();
-  await queue.add(`rollup-backfill-${data.day}`, data, {
+  await queue.add('rollup-backfill-scan', {}, {
     deduplication: {
-      id: `rollup-backfill-${data.day}`,
+      id: 'rollup-backfill-scan',
     }
   });
 }
 
-export async function processRollupBackfillJobs(processor: (data: RollupBackfillJobData) => Promise<void>) {
-  const jobProcessor = async (job: Job) => {
-    const data = schema.parse(job.data);
-    await processor(data);
+export async function processRollupBackfillJobs(processor: () => Promise<void>) {
+  const jobProcessor = async (_job: Job) => {
+    await processor();
   }
 
   return new Worker(QUEUE_NAME_ROLLUP_BACKFILL, jobProcessor, {