From dbaca715f85fcbf1607989e8d2540ddb5cb317d8 Mon Sep 17 00:00:00 2001 From: peetzweg Date: Thu, 30 Jul 2026 11:50:11 +0200 Subject: [PATCH] Turn the org backfill into a daily background worker Make backfill-orgs a scheduled task that drains the recorded-but-unfilled org queue a bounded slice at a time, instead of a one-off script that has to run to completion. Extract the lock / budget-guard / runtime-cap plumbing that monthly-user-refresh proved out into src/lib/job-runner.ts, and rebuild the backfill on top of it as an engine + store + tests. Per-run caps (--max-requests, --max-runtime-minutes) stop the run cleanly and leave the rest for tomorrow; builtAt is stamped only once every enumerated member is fetched, so a partially-filled org stays queued and resumes from org_members.last_fetched. A 2,800-member org now gets chipped away over several nights rather than blocking the queue. Also raise the member pagination cap for the worker (the request path keeps its own) and report truncation, since three orgs in the queue have more than the 1,000 members the old cap could enumerate. Two reliability fixes found while testing against the dev database: - cleanup now releases the advisory lock first and defensively. A crash mid-teardown skipped pg_advisory_unlock, leaving the lock held by a server-side session that outlived the process, so the next run reported status=locked and did nothing. - pool shutdown can no longer flip the exit code. postgres.js threw asynchronously while tearing sockets down, which would show a completed run as a failed Coolify execution. A run that fails every org without fetching anything now gives up after three and exits non-zero, rather than printing "Done" and looking like success. --- CLAUDE.md | 22 + package.json | 3 +- scripts/backfill-orgs.ts | 478 +++++++++------------- scripts/monthly-user-refresh.ts | 20 +- src/lib/github.ts | 15 +- src/lib/job-runner.ts | 197 +++++++++ src/lib/monthly-user-refresh-store.ts | 44 +- src/lib/monthly-user-refresh.test.ts | 6 +- src/lib/monthly-user-refresh.ts | 126 +----- src/lib/org-backfill-store.ts | 194 +++++++++ src/lib/org-backfill.test.ts | 465 +++++++++++++++++++++ src/lib/org-backfill.ts | 567 ++++++++++++++++++++++++++ src/lib/org-cache.ts | 2 +- 13 files changed, 1709 insertions(+), 430 deletions(-) create mode 100644 src/lib/job-runner.ts create mode 100644 src/lib/org-backfill-store.ts create mode 100644 src/lib/org-backfill.test.ts create mode 100644 src/lib/org-backfill.ts diff --git a/CLAUDE.md b/CLAUDE.md index a772e1a..a7b0c2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,6 +22,28 @@ the self-hosting playbook live in the private `peetzweg/devops` repo. - **Ambient `GITHUB_TOKEN` in the shell shadows `.env`** and lacks `read:org` — prefix dev/bun script runs with `env -u GITHUB_TOKEN` when org lookups misbehave. +## Background jobs + +Recipe, proven by `monthly-user-refresh` and `backfill-orgs`: a CLI in `scripts/` (flags, +each mirrored by an env var) → bundled by `build:worker` into `.output/worker/.mjs` → +the Dockerfile already copies `.output`, so shipping a new job needs no Dockerfile change → +a Coolify **scheduled task** runs `node .output/worker/.mjs` (cron + `docker exec` into +the running app container; container TZ is UTC). Logic goes in a pure engine +(`src/lib/.ts`) over an injected store (`-store.ts`) so it unit-tests against fakes. + +Every job must: take its advisory lock from `LOCK_KEYS` in `src/lib/job-runner.ts` (unique key +per job); use `createBudgetGuard` so it can't drain the GitHub token below the floor that live +traffic needs; stop cleanly on a wall-clock cap instead of being killed; resume off a DB +freshness marker (**the missing row is the retry queue** — no job-state table); write only +idempotent upserts; isolate per-item failures to a narrow status allowlist; and emit one +greppable ` done status=… ` summary line. Stagger schedules: the prod box has no swap and +`docker exec` adds a second node process inside the app container. + +Two failure modes worth knowing: cleanup must release the lock **first and defensively**, or a +crash mid-teardown leaves it held by a server-side session that outlives the process and every +later run reports `status=locked` and silently does nothing; and never let pool shutdown throw, +or a completed run shows up as a failed Coolify execution. + ## Domain guardrails - Single-segment paths are GitHub logins (`$user` route); editorial pages live under the diff --git a/package.json b/package.json index 0aef984..99585a2 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,7 @@ "generate-routes": "tsr generate", "og": "tsx scripts/generate-og.ts", "build": "tsx scripts/generate-og.ts && vite build && pnpm run build:worker", - "build:worker": "esbuild scripts/monthly-user-refresh.ts --bundle --platform=node --target=node24 --format=esm --outfile=.output/worker/monthly-user-refresh.mjs", + "build:worker": "esbuild scripts/monthly-user-refresh.ts scripts/backfill-orgs.ts --bundle --platform=node --target=node24 --format=esm --outdir=.output/worker --out-extension:.js=.mjs", "start": "node .output/server/index.mjs", "preview": "vite preview", "test": "vitest run", @@ -23,6 +23,7 @@ "refresh-user": "bun scripts/refresh-user.ts", "monthly:user": "node --env-file-if-exists=.env --import tsx scripts/monthly-user-refresh.ts", "backfill": "bun scripts/backfill-contributions.ts", + "backfill:orgs": "node --env-file-if-exists=.env --import tsx scripts/backfill-orgs.ts", "backup": "bash scripts/backup.sh", "db:generate": "drizzle-kit generate", "db:migrate": "drizzle-kit migrate", diff --git a/scripts/backfill-orgs.ts b/scripts/backfill-orgs.ts index 52a6830..73928c1 100644 --- a/scripts/backfill-orgs.ts +++ b/scripts/backfill-orgs.ts @@ -1,309 +1,233 @@ /** - * Lifetime-totals backfill for organizations the request path won't build on demand. + * Fill organization lifetime totals in the background, a bounded slice at a time. * - * After #103, an org with more than MAX_ORG_MEMBERS (25) public members is *recorded* on lookup - * (an `entities` row, `builtAt` null) but never built live — building a mega-org on a page request - * would burn the shared token's hourly budget. This script fills those recorded-but-empty orgs - * (and any org you name explicitly, e.g. google/microsoft/github) with the SAME numbers the - * request path produces for small orgs: each public member's lifetime contributions *to that org*, - * summed. It writes only the existing tables — `org_members` rows + the `entities` roll-up — so it - * needs NO database migration. (The per-month resolution / company chart is separate future work - * on the refresh-orgs worker; this deliberately does none of that.) + * Bundled by `pnpm build` into `.output/worker/backfill-orgs.mjs`, which is what the production + * image runs as a Coolify scheduled task (daily, 06:00 UTC — deliberately clear of the monthly + * user refresh, which runs on the 1st at 03:00 and takes a couple of hours): * - * Politeness: paced well under GitHub's 5,000 points/hour with a reserved floor for live traffic, - * same model as backfill-contributions.ts. Run with bun (auto-loads .env; beware a shell-exported - * GITHUB_TOKEN overriding it — prefix with `env -u GITHUB_TOKEN` if your shell sets one): + * node .output/worker/backfill-orgs.mjs * - * bun scripts/backfill-orgs.ts # every recorded org not yet filled, SMALLEST first - * bun scripts/backfill-orgs.ts google microsoft # these orgs specifically (records them if new) - * bun scripts/backfill-orgs.ts --force # re-fetch every member (refresh, don't resume) + * Locally, `pnpm backfill:orgs` runs this source through tsx. * - * No-arg runs go smallest-org-first (fewest public members) so small orgs resolve quickly and - * mega-orgs fall to the back instead of blocking everyone behind one 1,000-member org. + * The engine (src/lib/org-backfill.ts) is built to *not* finish: it spends at most + * --max-requests GraphQL requests per run, keeps a floor of budget free for live site traffic, + * stops on a wall-clock cap, and resumes from `org_members.last_fetched` next time. So a + * 2,800-member org gets chipped away over several nights instead of blocking the queue, and every + * org the site has ever recorded eventually gets filled. * - * Safe to re-run and interrupt: every write is an idempotent upsert; a member's `lastFetched` - * marks it done, so a re-run SKIPS members already fetched (in this run or a previous, aborted - * one) and continues from where it stopped. An org's `builtAt` is stamped only after all its - * members are fetched, so an interrupted org stays unfilled and is picked up again. Pass --force - * to re-fetch already-filled members (refresh their numbers) instead of resuming. + * Execute-by-default. `--dry-run` reports what it would do without spending requests or writing. + * + * node .output/worker/backfill-orgs.mjs # drain the queue, smallest org first + * pnpm backfill:orgs --dry-run # preview the slice + * pnpm backfill:orgs google microsoft # these orgs specifically + * pnpm backfill:orgs --force jupyter # re-fetch every member, don't resume + * pnpm backfill:orgs --stale-after-days=90 --max-stale-orgs=5 # refresh the 5 oldest built orgs + * + * Member enumeration needs the read:org token from .env, so beware a shell-exported GITHUB_TOKEN + * shadowing it: prefix local runs with `env -u GITHUB_TOKEN` if your shell sets one. */ -import { and, asc, eq, isNull, sql } from "drizzle-orm"; import { db } from "#/lib/db"; -import { entities, orgMembers } from "#/lib/db/schema"; import { - fetchOrgMembers, fetchOrgMemberContributions, + fetchOrgMembers, fetchOrgProfile, - GitHubError, - type OrgMemberTotals, + fetchRateLimitBudget, yearlyWindows, } from "#/lib/github"; +import { runOrgBackfill } from "#/lib/org-backfill"; +import { createOrgBackfillStore } from "#/lib/org-backfill-store"; + +const VALID_FLAGS = new Set([ + "dry-run", + "force", + "h", + "help", + "max-orgs", + "max-requests", + "max-runtime-minutes", + "max-stale-orgs", + "member-pages", + "poll-every", + "rate-per-hour", + "remaining-floor", + "stale-after-days", +]); + +const { flags, logins } = parseArgv(process.argv.slice(2)); +const env = process.env; + +if (booleanFlag(flags, "help") || booleanFlag(flags, "h")) { + console.log(usage()); + process.exit(0); +} + +const DATABASE_URL = env.DATABASE_URL; +const GITHUB_TOKEN = env.GITHUB_TOKEN; -if (!process.env.DATABASE_URL) - throw new Error("DATABASE_URL is required (add it to .env)"); -const GITHUB_TOKEN = process.env.GITHUB_TOKEN; -if (!GITHUB_TOKEN) throw new Error("GITHUB_TOKEN is required (add it to .env)"); +if (!DATABASE_URL) throw new Error("DATABASE_URL is required."); +if (!GITHUB_TOKEN) throw new Error("GITHUB_TOKEN is required."); if (!db) throw new Error("Database client failed to initialise."); -const token = GITHUB_TOKEN; -const database = db; -const orgEntityId = (login: string) => `org:${login.trim().toLowerCase()}`; -const userEntityId = (login: string) => `user:${login.trim().toLowerCase()}`; +const maxRuntimeMinutes = numberValue( + flags, + "max-runtime-minutes", + env.ORG_BACKFILL_MAX_RUNTIME_MINUTES, + 120, +); + +const result = await runOrgBackfill({ + store: createOrgBackfillStore(db), + token: GITHUB_TOKEN, + now: new Date(), + logins, + dryRun: booleanFlag(flags, "dry-run") || isTrue(env.ORG_BACKFILL_DRY_RUN), + force: booleanFlag(flags, "force") || isTrue(env.ORG_BACKFILL_FORCE), + maxOrgs: numberValue(flags, "max-orgs", env.ORG_BACKFILL_MAX_ORGS, 25), + maxRequests: numberValue( + flags, + "max-requests", + env.ORG_BACKFILL_MAX_REQUESTS, + 1500, + ), + ratePerHour: numberValue( + flags, + "rate-per-hour", + env.ORG_BACKFILL_RATE_PER_HOUR, + 1200, + ), + remainingFloor: numberValue( + flags, + "remaining-floor", + env.ORG_BACKFILL_REMAINING_FLOOR, + 500, + ), + pollEvery: numberValue(flags, "poll-every", env.ORG_BACKFILL_POLL_EVERY, 25), + memberPages: numberValue( + flags, + "member-pages", + env.ORG_BACKFILL_MEMBER_PAGES, + 30, + ), + staleAfterDays: numberValue( + flags, + "stale-after-days", + env.ORG_BACKFILL_STALE_AFTER_DAYS, + 0, + { allowZero: true }, + ), + maxStaleOrgs: numberValue( + flags, + "max-stale-orgs", + env.ORG_BACKFILL_MAX_STALE_ORGS, + 0, + { allowZero: true }, + ), + maxRuntimeMs: maxRuntimeMinutes * 60_000, + // The token is shared with live site traffic, so the run polls its own GraphQL budget and + // refuses to spend below the floor. `rateLimit` queries cost 0 points. + fetchRateLimit: () => fetchRateLimitBudget(GITHUB_TOKEN), + fetchOrgProfile, + fetchOrgMembers, + fetchOrgMemberContributions, + yearlyWindows, + logger: console, +}); -// Requests/hour we aim to spend (≈ GraphQL points), well under the 5,000/hr limit so the live -// site keeps working. Override with REFRESH_RATE=. -const TARGET_RATE = Number(process.env.REFRESH_RATE ?? 2500); -// Never let the remaining budget drop below this — headroom for live traffic. -const REMAINING_FLOOR = 500; -const CHUNK_ROWS = 100; -// fetchOrgMemberContributions batches windows this many at a time; used only to pace, not for -// correctness — an over-estimate just spends the budget more conservatively. -const WINDOWS_PER_REQUEST = 6; -// --force re-fetches every member even if already filled (to refresh numbers). Default: resume, -// i.e. skip members that already have a lastFetched from any prior run. -const force = process.argv.slice(2).includes("--force"); +// Stopping on a cap is normal: the queue is the retry mechanism and tomorrow's run continues. +// A systemic fault is not, and a scheduled task that exits 0 while achieving nothing is invisible +// in Coolify — so those two exit non-zero and show up as a failed execution. +const systemicFailure = + result.stopReason === "consecutive_failures" || + (result.orgsFailed > 0 && result.membersFetched === 0 && !result.dryRun); -const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +await shutdown(systemicFailure ? 1 : 0); -interface RateLimit { - remaining: number; - resetAt: string; +/** + * Close the pool so Postgres isn't left holding idle backends until TCP keepalive reaps them — + * best-effort only. postgres.js can throw *asynchronously* while tearing sockets down (seen when + * the link to the database blipped mid-run), and an uncaught throw at this point would turn a run + * that already did its work and logged its summary into a failed Coolify execution. Shutdown is + * explicitly not allowed to change the outcome. + */ +async function shutdown(code: number): Promise { + process.on("uncaughtException", (err) => { + console.warn( + `org-backfill status=shutdown_error error=${JSON.stringify(String(err))}`, + ); + process.exit(code); + }); + await db?.$client.end({ timeout: 5 }).catch(() => {}); + process.exit(code); } -/** Query GitHub's current rate-limit budget. `rateLimit` queries themselves cost 0 points. */ -async function rateLimit(): Promise { - try { - const res = await fetch("https://api.github.com/graphql", { - method: "POST", - headers: { - Authorization: `Bearer ${token}`, - "Content-Type": "application/json", - "User-Agent": "commit-history-backfill-orgs", - }, - body: JSON.stringify({ - query: "query { rateLimit { remaining resetAt } }", - }), - }); - const json = (await res.json()) as { data?: { rateLimit: RateLimit } }; - return json.data?.rateLimit ?? null; - } catch { - return null; +function parseArgv(argv: string[]): { + flags: Map; + logins: string[]; +} { + const flags = new Map(); + const logins: string[] = []; + for (const arg of argv) { + if (!arg.startsWith("--")) { + logins.push(arg); + continue; + } + const [rawName, ...rest] = arg.slice(2).split("="); + const name = rawName.trim(); + if (!name) throw new Error(`Invalid flag "${arg}".`); + if (!VALID_FLAGS.has(name)) throw new Error(`Unknown flag "--${name}".`); + flags.set(name, rest.length === 0 ? true : rest.join("=")); } + return { flags, logins }; } -/** If we're near the reserved floor, sleep until GitHub's window resets (plus a small buffer). */ -async function respectFloor(): Promise { - const rl = await rateLimit(); - if (!rl) return; - if (rl.remaining > REMAINING_FLOOR) return; - const waitMs = Math.max(0, new Date(rl.resetAt).getTime() - Date.now()) + 2000; - console.log( - `… budget low (${rl.remaining} left) — pausing ${Math.ceil(waitMs / 1000)}s until reset`, - ); - await sleep(waitMs); +function booleanFlag(flags: Map, name: string): boolean { + return flags.get(name) === true; } -/** - * Fill one org with lifetime totals. Returns the (approximate) number of GitHub requests spent. - * Re-fetches the profile, (re-)enumerates members, fetches each member's org-scoped lifetime - * totals, then rolls the sums onto the org's `entities` row and stamps `builtAt`. - */ -async function fillOrg(login: string): Promise { - const orgId = orgEntityId(login); - const runStart = new Date(); - let requests = 0; - console.log(`${login}: resolving profile…`); - - // Profile first — validates the login, yields nodeId (keys every org-scoped query) + createdAt - // (bounds each member's windows). Upsert so a never-looked-up org (e.g. a fresh google) is - // recorded here rather than requiring a prior page visit. builtAt stays null until the roll-up. - const profile = await fetchOrgProfile(login, token); - requests += 1; - const profileCols = { - login: profile.login, - name: profile.name, - avatarUrl: profile.avatarUrl, - htmlUrl: profile.htmlUrl, - createdAt: new Date(profile.createdAt), - bio: profile.description, - location: profile.location, - websiteUrl: profile.websiteUrl, - twitterUsername: profile.twitterUsername, - publicRepos: profile.publicRepos, - isVerified: profile.isVerified, - githubNodeId: profile.nodeId, - memberCount: profile.memberCount, - lastFetched: runStart, - }; - await database - .insert(entities) - .values({ id: orgId, kind: "org", ...profileCols }) - .onConflictDoUpdate({ target: entities.id, set: profileCols }); - const orgCreated = new Date(profile.createdAt); - - // Enumerate members → a user stub (FK target, never overwrites a real user row) + a pending - // org_members row each. Both idempotent. - const members = await fetchOrgMembers(login, token); - requests += Math.max(1, Math.ceil(members.length / 100)); - for (let i = 0; i < members.length; i += CHUNK_ROWS) { - const chunk = members.slice(i, i + CHUNK_ROWS); - await database - .insert(entities) - .values( - chunk.map((m) => ({ - id: userEntityId(m.login), - kind: "user", - login: m.login, - name: m.name, - avatarUrl: m.avatarUrl, - htmlUrl: `https://github.com/${m.login}`, - createdAt: new Date(m.createdAt), - })), - ) - .onConflictDoNothing(); - await database - .insert(orgMembers) - .values( - chunk.map((m) => ({ - orgId, - memberId: userEntityId(m.login), - role: m.role, - source: "public_member", - })), - ) - .onConflictDoNothing(); - } - - // Resume support: a member's `lastFetched` is set once we've fetched it — in this run OR a - // previous, aborted one. By default we skip anything already fetched, so re-running continues - // from exactly where an abort stopped (e.g. mid-Microsoft) instead of restarting the org. - // --force re-fetches everyone, to refresh an already-filled org's numbers. - const existing = await database - .select({ - memberId: orgMembers.memberId, - lastFetched: orgMembers.lastFetched, - }) - .from(orgMembers) - .where(eq(orgMembers.orgId, orgId)); - const fetchedById = new Map(existing.map((r) => [r.memberId, r.lastFetched])); - const isDone = (login: string) => - !force && fetchedById.get(userEntityId(login)) != null; - const already = members.filter((m) => isDone(m.login)).length; - console.log( - ` ${members.length} public members${already ? ` (${already} already done — resuming)` : ""} — fetching lifetime totals, ~${Math.round((Math.max(1, Math.ceil(20 / WINDOWS_PER_REQUEST)) / TARGET_RATE) * 3600)}s each`, - ); - - let done = 0; - for (const m of members) { - const memberId = userEntityId(m.login); - if (isDone(m.login)) continue; - - // A member can't have contributed to the org before either account existed. - const start = new Date( - Math.max(orgCreated.getTime(), new Date(m.createdAt).getTime()), - ); - const windows = yearlyWindows(start, runStart); - let totals: OrgMemberTotals = { - commits: 0, - issues: 0, - pullRequests: 0, - reviews: 0, - }; - try { - totals = await fetchOrgMemberContributions( - m.login, - profile.nodeId, - token, - windows, - ); - } catch (e) { - // Isolate per-member failures so one bad member can't sink the whole org: - // - 404: deleted/renamed member — gone forever. - // - 400: login our validator can't accept (shouldn't happen now that legacy - // hyphen-ending usernames pass, but stay defensive). - // Record zeros + mark done so resume doesn't wedge on it. Rate limits (403/429) and - // 5xx still propagate → the run aborts and resumes later rather than storing false zeros. - if (!(e instanceof GitHubError && (e.status === 404 || e.status === 400))) { - throw e; - } - console.log(` – ${m.login.padEnd(22)} skipped (${(e as GitHubError).status})`); - } - const req = Math.max(1, Math.ceil(windows.length / WINDOWS_PER_REQUEST)); - requests += req; +function isTrue(value: string | undefined): boolean { + return value?.toLowerCase() === "true"; +} - await database - .update(orgMembers) - .set({ ...totals, lastFetched: new Date() }) - .where( - and(eq(orgMembers.orgId, orgId), eq(orgMembers.memberId, memberId)), - ); - done += 1; - console.log( - ` ✓ ${m.login.padEnd(22)} ${totals.commits.toLocaleString()} commits (${done}/${members.length})`, +function numberValue( + flags: Map, + name: string, + envValue: string | undefined, + fallback: number, + opts: { allowZero?: boolean } = {}, +): number { + const flagValue = flags.get(name); + if (flagValue === true) throw new Error(`--${name} requires a value.`); + const raw = typeof flagValue === "string" ? flagValue : envValue; + if (raw == null || raw === "") return fallback; + const value = Number(raw); + const floor = opts.allowZero ? 0 : 1; + if (!Number.isInteger(value) || value < floor) { + throw new Error( + `--${name} must be an integer >= ${floor}${opts.allowZero ? " (0 disables it)" : ""}.`, ); - // Politeness pacing: spread this member's request cost across the target hourly rate. - await sleep((req / TARGET_RATE) * 3_600_000); } - - // Roll up org totals from org_members (departed rows, if any, stay frozen — same as #97) and - // stamp builtAt so the org now looks finished to the request path. - const [sums] = await database - .select({ - commits: sql`coalesce(sum(${orgMembers.commits}), 0)`, - pullRequests: sql`coalesce(sum(${orgMembers.pullRequests}), 0)`, - reviews: sql`coalesce(sum(${orgMembers.reviews}), 0)`, - issues: sql`coalesce(sum(${orgMembers.issues}), 0)`, - }) - .from(orgMembers) - .where(eq(orgMembers.orgId, orgId)); - await database - .update(entities) - .set({ - totalCommits: Number(sums?.commits ?? 0), - totalPullRequests: Number(sums?.pullRequests ?? 0), - totalReviews: Number(sums?.reviews ?? 0), - totalIssues: Number(sums?.issues ?? 0), - builtAt: new Date(), - lastFetched: new Date(), - }) - .where(eq(entities.id, orgId)); - - console.log( - `✓ ${login.padEnd(24)} ${Number(sums?.commits ?? 0).toLocaleString()} commits · ${done} fetched${already ? ` + ${already} already done` : ""} / ${members.length} members`, - ); - return requests; + return value; } -const args = process.argv.slice(2).filter((a) => !a.startsWith("-")); - -let targets: { login: string }[]; -if (args.length > 0) { - targets = args.map((login) => ({ login })); - console.log(`Backfilling ${targets.length} org(s) by name\n`); -} else { - // Every recorded-but-unfilled org (builtAt null), SMALLEST first (fewest public members) so - // small orgs resolve quickly and mega-orgs (microsoft, google) fall to the back instead of - // blocking everyone. An org interrupted mid-fill still has builtAt null, so it's picked up - // again and resumes from where it stopped (members already fetched are skipped). - const rows = await database - .select({ login: entities.login }) - .from(entities) - .where(and(eq(entities.kind, "org"), isNull(entities.builtAt))) - .orderBy(sql`${entities.memberCount} asc nulls last`, asc(entities.id)); - targets = rows; - console.log( - `${rows.length} recorded org(s) not yet filled, smallest first, ~${TARGET_RATE} req/hr\n`, - ); -} - -let spent = 0; -for (const { login } of targets) { - await respectFloor(); - try { - spent += await fillOrg(login); - } catch (e) { - const msg = e instanceof Error ? e.message : String(e); - console.log(`✗ ${login.padEnd(24)} ${msg} — continuing`); - } +function usage(): string { + return [ + "Usage:", + " node .output/worker/backfill-orgs.mjs [flags] [org…] # built image", + " pnpm backfill:orgs [flags] [org…] # local source via tsx", + "", + "Named orgs bypass the queue. With none, drains recorded-but-unfilled orgs, smallest first.", + "", + "Flags override env vars:", + " --dry-run env ORG_BACKFILL_DRY_RUN", + " --force env ORG_BACKFILL_FORCE", + " --max-orgs=N env ORG_BACKFILL_MAX_ORGS, default 25", + " --max-requests=N env ORG_BACKFILL_MAX_REQUESTS, default 1500", + " --rate-per-hour=N env ORG_BACKFILL_RATE_PER_HOUR, default 1200", + " --remaining-floor=N env ORG_BACKFILL_REMAINING_FLOOR, default 500", + " --poll-every=N env ORG_BACKFILL_POLL_EVERY, default 25", + " --member-pages=N env ORG_BACKFILL_MEMBER_PAGES, default 30 (×100 members)", + " --max-runtime-minutes=N env ORG_BACKFILL_MAX_RUNTIME_MINUTES, default 120", + " --stale-after-days=N env ORG_BACKFILL_STALE_AFTER_DAYS, default 0 (off)", + " --max-stale-orgs=N env ORG_BACKFILL_MAX_STALE_ORGS, default 0 (off)", + ].join("\n"); } -console.log(`\nDone. ~${spent.toLocaleString()} requests spent.`); diff --git a/scripts/monthly-user-refresh.ts b/scripts/monthly-user-refresh.ts index 2e01db5..c276887 100644 --- a/scripts/monthly-user-refresh.ts +++ b/scripts/monthly-user-refresh.ts @@ -72,8 +72,24 @@ await runMonthlyUserRefresh({ logger: console, }); -await db.$client.end(); -process.exit(0); +await shutdown(0); + +/** + * Close the pool so Postgres isn't left holding an idle backend until TCP keepalive reaps it — + * best-effort only. postgres.js can throw *asynchronously* while tearing sockets down (seen when + * the link to the database blipped mid-run), and an uncaught throw at this point would turn a run + * that already did its work and logged its summary into a failed Coolify execution. + */ +async function shutdown(code: number): Promise { + process.on("uncaughtException", (err) => { + console.warn( + `monthly-user-refresh status=shutdown_error error=${JSON.stringify(String(err))}`, + ); + process.exit(code); + }); + await db?.$client.end({ timeout: 5 }).catch(() => {}); + process.exit(code); +} interface Config { help: boolean; diff --git a/src/lib/github.ts b/src/lib/github.ts index dd41788..c49bfdf 100644 --- a/src/lib/github.ts +++ b/src/lib/github.ts @@ -541,21 +541,28 @@ export async function fetchOrgProfile( // Hard stop for the members pagination loop (pages of 100). The org build enforces its own, // lower member cap before fetching contributions — this is just a runaway/mega-org backstop. +// The backfill worker raises it (see `maxPages`), since a 2,800-member org is exactly the case +// the request path refuses and the worker exists to handle. const MAX_MEMBER_PAGES = 10; /** * Enumerate an org's members visible to the token — public members only unless the token itself * belongs to the org. Each node carries `createdAt`, so the caller can window every member's * contribution fetch without per-member profile requests. + * + * `truncated` means the page cap stopped us with more members still to come. Callers that stamp an + * org as built need it: rolling up a truncated membership silently undercounts the org forever. */ export async function fetchOrgMembers( rawLogin: string, token: string, -): Promise { + opts: { maxPages?: number } = {}, +): Promise<{ members: OrgMember[]; truncated: boolean }> { const login = assertLogin(rawLogin); + const maxPages = Math.max(1, opts.maxPages ?? MAX_MEMBER_PAGES); const members: OrgMember[] = []; let cursor: string | null = null; - for (let page = 0; page < MAX_MEMBER_PAGES; page++) { + for (let page = 0; page < maxPages; page++) { const after: string = cursor ? `, after: "${cursor}"` : ""; const data = await graphql<{ organization: { @@ -587,10 +594,10 @@ export async function fetchOrgMembers( if (!edge.node) continue; members.push({ ...edge.node, role: edge.role }); } - if (!conn.pageInfo.hasNextPage) return members; + if (!conn.pageInfo.hasNextPage) return { members, truncated: false }; cursor = conn.pageInfo.endCursor; } - return members; + return { members, truncated: true }; } /** A member's summed lifetime contributions to one org (org-scoped, not their global totals). */ diff --git a/src/lib/job-runner.ts b/src/lib/job-runner.ts new file mode 100644 index 0000000..bd81bb2 --- /dev/null +++ b/src/lib/job-runner.ts @@ -0,0 +1,197 @@ +/** + * Shared plumbing for the background workers that run as Coolify scheduled tasks + * (`.output/worker/*.mjs`, see the Dockerfile). Extracted from monthly-user-refresh once a second + * job needed the same three guarantees: + * + * 1. one run at a time (Postgres advisory lock), + * 2. never drain the shared GitHub token below a reserved floor, + * 3. stop cleanly at a wall-clock cap instead of being killed by the task timeout. + * + * Every job here is expected to be *resumable off its own data* — a missing row is the retry queue — + * so stopping early is normal and cheap, and exiting 0 with work left over is a success. + */ + +import type { DB } from "#/lib/db"; +import type { RateLimitBudget } from "#/lib/github"; + +export interface JobLogger { + info(message: string): void; + warn(message: string): void; + error(message: string): void; +} + +export const quietLogger: JobLogger = { + info: () => {}, + warn: () => {}, + error: () => {}, +}; + +export const defaultSleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)); + +export function positiveInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value <= 0) { + throw new Error(`${name} must be a positive integer.`); + } + return value; +} + +export function nonNegativeInteger(value: number, name: string): number { + if (!Number.isInteger(value) || value < 0) { + throw new Error(`${name} must be a non-negative integer.`); + } + return value; +} + +export interface JobLock { + tryLock(): Promise; + releaseLock(): Promise; +} + +/** + * Session-scoped advisory lock pinned to one reserved connection. + * + * `database.execute` takes an arbitrary connection out of the postgres.js pool, and unlocking from + * a different connection than the one holding the lock silently returns false. Mutual exclusion + * would work either way (a second process is a second session), but a release that quietly no-ops + * is a trap, so the connection is held for the run's lifetime and released in `releaseLock`. + * + * Keys must be unique per job — see the LOCK_KEYS table below. + */ +export function createJobLock( + database: DB, + key1: number, + key2: number, + label: string, +): JobLock { + let lockConn: Awaited> | null = null; + + return { + async tryLock() { + lockConn = await database.$client.reserve(); + const [row] = await lockConn>` + select pg_try_advisory_lock(${key1}, ${key2}) as acquired`; + const acquired = Boolean(row?.acquired); + if (!acquired) { + lockConn.release(); + lockConn = null; + } + return acquired; + }, + + async releaseLock() { + if (!lockConn) return; + try { + const [row] = await lockConn>` + select pg_advisory_unlock(${key1}, ${key2}) as released`; + if (!row?.released) { + console.warn( + `${label} lock=release_returned_false (was it held by this session?)`, + ); + } + } finally { + lockConn.release(); + lockConn = null; + } + }, + }; +} + +/** + * Advisory-lock keys, one row per background job. Same first key, distinct second — collisions + * would make two unrelated jobs exclude each other, which is much harder to notice than a job + * that never runs. + */ +export const LOCK_KEYS = { + monthlyUserRefresh: [20260701, 1], + orgBackfill: [20260701, 2], +} as const; + +export interface BudgetGuard { + /** True to proceed; false when the run must stop rather than dip below the floor. */ + ensure(force?: boolean): Promise; + /** Account for requests just spent, between polls. */ + spend(requests?: number): void; + /** Total requests this guard has been told about — the run's spend estimate. */ + spent(): number; +} + +/** + * Keeps a run above `remainingFloor` GraphQL points. The token is shared with live site traffic, so + * a batch job that drains the hourly quota takes the site down with it. Polling costs nothing but a + * round-trip, so we poll every `pollEvery` requests and decrement locally in between. + * + * Below the floor the only options are "wait for the window to reset" or "stop". Waiting is allowed + * only if it fits inside the remaining max-runtime budget: a scheduled job that sleeps past its + * window is worse than one that exits 0 and leaves the rest for the next pass. + */ +export function createBudgetGuard(opts: { + /** Log prefix, e.g. `org-backfill` — every line is `