From 222228d7f3880d6711cc76dee75e938507922744 Mon Sep 17 00:00:00 2001 From: needs <624097+needs@users.noreply.github.com> Date: Fri, 21 Aug 2026 01:12:09 +0200 Subject: [PATCH] Run the rollup backfill scan from the worker instead of the scheduler The scheduler app has no S3 credentials, so listing archived days from it failed every tick with ECONNREFUSED against the localhost fallback and no backfill job was ever enqueued. The scheduler now enqueues a single parameterless scan job, and the worker lists the archive, picks the missing days, and backfills them oldest first. Co-Authored-By: Claude Fable 5 --- .../src/schedulers/rollupBackfillScheduler.ts | 63 +--------------- apps/worker/src/workers/rollupBackfill.ts | 75 ++++++++++++++----- .../src/lib/bullmq/queueRollupBackfill.ts | 21 ++---- 3 files changed, 65 insertions(+), 94 deletions(-) diff --git a/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts b/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts index b1c279e..1f870df 100644 --- a/apps/scheduler/src/schedulers/rollupBackfillScheduler.ts +++ b/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(); }); } diff --git a/apps/worker/src/workers/rollupBackfill.ts b/apps/worker/src/workers/rollupBackfill.ts index 3789632..71ea3d5 100644 --- a/apps/worker/src/workers/rollupBackfill.ts +++ b/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); } diff --git a/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts b/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts index b6277a2..c3632af 100644 --- a/libs/teerank/src/lib/bullmq/queueRollupBackfill.ts +++ b/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; - -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) { - const jobProcessor = async (job: Job) => { - const data = schema.parse(job.data); - await processor(data); +export async function processRollupBackfillJobs(processor: () => Promise) { + const jobProcessor = async (_job: Job) => { + await processor(); } return new Worker(QUEUE_NAME_ROLLUP_BACKFILL, jobProcessor, {