From eec998a76233bb33e2dddbd921fc81da2eba1e0a Mon Sep 17 00:00:00 2001 From: Ben Senescu <44480372+bensenescu@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:44:16 -0400 Subject: [PATCH] feat(rank-tracking): raise scheduler throughput 5x with rate-limit-derived sizing (#465) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(rank-tracking): raise scheduler budget to 2000 units/tick The 200-unit budget used ~2-6% of DataForSEO's 2,000 req/min account cap and would take days to drain the post-#462 backlog. Scheduled checks run through the task queue (1 task_post per 100 units + free task_get polls), so a full 2,000-unit tick peaks around 1,200 req/min — still leaving headroom for the other DataForSEO products on the account. Also raise the due-config fetch limit to 500 so skip-heavy stretches (free orgs, keywordless configs) drain more than 200 rows per tick. * fix(rank-tracking): retune budget to 1000 with accurate sizing and a tick deadline Review corrections to the 10x bump: task_get polling is one call per unit per round and rounds wake synchronized per tick, with up to three ~15-min poll windows overlapping the */5 cron — 2000 units/tick could saturate DataForSEO's 2000 req/min cap, silently aging throttled polls into the ~3x-cost live fallback billed to customers. 1000/tick keeps real headroom and is still ~45x steady-state demand. Add a 3-minute wall-clock deadline to the per-config loop (stoppedByDeadline in the tick summary): a skip-heavy 500-candidate tick pays serial Autumn round-trips per distinct org and could otherwise run into the 15-minute cron kill. Name the fetch limit (DUE_CONFIGS_PER_TICK) and correct its comment. Test fixtures now derive from MAX_KEYWORDS_PER_CONFIG instead of asserting an unreachable 1500-keyword config. * fix(cron): run the audit watchdog before the rank loop reconcileStaleAudits ran after runScheduledRankChecks in the same invocation, so a slow rank tick would delay the watchdog and a wall-clock kill would skip it entirely. * fix(cron): preserve watchdog failure signal; codex review polish Rethrow a caught reconcileStaleAudits error after the rank loop so the invocation still reports failed (matching pre-reorder semantics), use an inclusive deadline comparison, and note overlapping-tick poll residue in the sizing comment. --- src/server.ts | 19 +++++---- .../repositories/RankTrackingRepository.ts | 7 +++- .../services/scheduledRankChecks.test.ts | 40 +++++++++++++++---- .../services/scheduledRankChecks.ts | 28 +++++++++++-- src/server/workflows/rankCheckPaths.ts | 2 +- 5 files changed, 77 insertions(+), 19 deletions(-) diff --git a/src/server.ts b/src/server.ts index 9d476f3..5132a93 100644 --- a/src/server.ts +++ b/src/server.ts @@ -188,15 +188,20 @@ export default { env: Env, _ctx: ExecutionContext, ) { - // Scope a per-request Postgres client for the cron run (no-op in D1 mode). - // Caught so a rank-tracking failure can't suppress the audit watchdog below. + // Watchdog first: reconcile audits stuck in "running" whose workflow died + // without reaching mark-failed (OOM/CPU kills, expired instances). Runs + // before the rank loop so a slow tick can't delay or starve it. Its + // failure is held until after the rank checks so it can't suppress them, + // then rethrown so the invocation still reports as failed. + let watchdogError: unknown; try { - await withPgClient(() => runScheduledRankChecks(env)); + await withPgClient(() => reconcileStaleAudits()); } catch (err) { - console.error("[cron] Scheduled rank checks failed:", err); + watchdogError = err; + console.error("[cron] Stale-audit reconcile failed:", err); } - // Watchdog: reconcile audits stuck in "running" whose workflow died - // without reaching mark-failed (OOM/CPU kills, expired instances). - await withPgClient(() => reconcileStaleAudits()); + // Scope a per-request Postgres client for the cron run (no-op in D1 mode). + await withPgClient(() => runScheduledRankChecks(env)); + if (watchdogError) throw watchdogError; }, }; diff --git a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts index 0cc8933..0fb49d8 100644 --- a/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts +++ b/src/server/features/rank-tracking/repositories/RankTrackingRepository.ts @@ -115,6 +115,11 @@ async function updateConfig( ); } +// Caps per-tick loop work (claims, per-org plan checks) against the cron +// wall clock; paid-heavy ticks are stopped earlier by the unit budget and +// slow ticks by TICK_DEADLINE_MS in scheduledRankChecks.ts. +const DUE_CONFIGS_PER_TICK = 500; + async function getDueConfigsWithOrganization(nowIso: string) { return ( db @@ -150,7 +155,7 @@ async function getDueConfigsWithOrganization(nowIso: string) { asc(rankTrackingConfigs.nextCheckAt), asc(rankTrackingConfigs.id), ) - .limit(200) + .limit(DUE_CONFIGS_PER_TICK) ); } diff --git a/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts b/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts index 94611e7..345e1ab 100644 --- a/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts +++ b/src/server/features/rank-tracking/services/scheduledRankChecks.test.ts @@ -1,4 +1,5 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { MAX_KEYWORDS_PER_CONFIG } from "@/shared/rank-tracking"; type DueConfigRow = { id: string; @@ -177,15 +178,16 @@ describe("runScheduledRankChecks", () => { }); it("stops admitting configs once the task-unit budget is spent", async () => { - // 400 keywords × 2 devices = 800 units — admitted anyway as the tick's - // first start (oversized exemption), then exhausts the 200-unit budget. + // A legal-max config (MAX_KEYWORDS_PER_CONFIG × both devices) exceeds the + // whole budget but is admitted as the tick's first start (oversized + // exemption); the next config then hits the budget stop. mocks.getDueConfigsWithOrganization.mockResolvedValue([ dueConfig({ id: "config_big" }), dueConfig({ id: "config_next", nextCheckAt: "2026-01-02T00:00:00.000Z" }), ]); mocks.getKeywordCountsForConfigs.mockResolvedValue( new Map([ - ["config_big", 400], + ["config_big", MAX_KEYWORDS_PER_CONFIG], ["config_next", 5], ]), ); @@ -198,7 +200,10 @@ describe("runScheduledRankChecks", () => { expect(mocks.claimDueConfig).toHaveBeenCalledTimes(1); expect(mocks.customerHasPaidPlan).toHaveBeenCalledTimes(1); expect(logSpy).toHaveBeenCalledWith( - expect.objectContaining({ stoppedByBudget: true, unitsStarted: 800 }), + expect.objectContaining({ + stoppedByBudget: true, + unitsStarted: MAX_KEYWORDS_PER_CONFIG * 2, + }), ); }); @@ -252,8 +257,9 @@ describe("runScheduledRankChecks", () => { }); it("defers a config whose units would overflow the remaining budget", async () => { - // First config fits (10 units); the second would overflow (800), so it is - // deferred rather than admitted just because budget remains. + // First config fits (10 units); the second (legal-max, over the whole + // budget) would overflow, so it is deferred rather than admitted just + // because budget remains. mocks.getDueConfigsWithOrganization.mockResolvedValue([ dueConfig({ id: "config_small" }), dueConfig({ id: "config_big", nextCheckAt: "2026-01-02T00:00:00.000Z" }), @@ -261,7 +267,7 @@ describe("runScheduledRankChecks", () => { mocks.getKeywordCountsForConfigs.mockResolvedValue( new Map([ ["config_small", 5], - ["config_big", 400], + ["config_big", MAX_KEYWORDS_PER_CONFIG], ]), ); const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); @@ -277,6 +283,26 @@ describe("runScheduledRankChecks", () => { ); }); + it("stops the tick at the wall-clock deadline instead of running long", async () => { + mocks.getDueConfigsWithOrganization.mockResolvedValue([ + dueConfig({ id: "config_1" }), + dueConfig({ id: "config_2", nextCheckAt: "2026-01-02T00:00:00.000Z" }), + ]); + const start = Date.now(); + vi.spyOn(Date, "now") + .mockReturnValueOnce(start) // deadline anchor + .mockReturnValue(start + 4 * 60_000); // every later check is past it + const logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + + await runTick(); + + expect(mocks.beginRankCheckRun).not.toHaveBeenCalled(); + expect(mocks.claimDueConfig).not.toHaveBeenCalled(); + expect(logSpy).toHaveBeenCalledWith( + expect.objectContaining({ stoppedByDeadline: true, started: 0 }), + ); + }); + it("restores the original due time when a run is already active", async () => { mocks.beginRankCheckRun.mockResolvedValue({ ok: false, diff --git a/src/server/features/rank-tracking/services/scheduledRankChecks.ts b/src/server/features/rank-tracking/services/scheduledRankChecks.ts index 7f6f6be..c24cd57 100644 --- a/src/server/features/rank-tracking/services/scheduledRankChecks.ts +++ b/src/server/features/rank-tracking/services/scheduledRankChecks.ts @@ -10,9 +10,24 @@ import { // Work admitted per tick, in task units (keywords × devices). Admission // control, not a hard rate limit: the first start of a tick is always -// admitted, so an oversized config can never starve. 200 units on the -// 5-minute cron ≈ 57,600 units/day, ~9× current steady-state demand. -const SCHEDULED_TASK_UNIT_BUDGET = 200; +// admitted, so a config bigger than the budget (legal max: 1,000 keywords × +// 2 devices = 2,000 units) can never starve. Sized against DataForSEO's +// 2,000 requests/min account cap, where task_get polling is the binding +// term: one call per unit per poll round, rounds wake synchronized per tick, +// and up to three ticks' ~15-minute poll windows overlap the */5 cron — so a +// full tick can burst ~1,000 polls into a single minute, stacking with the +// residual rounds of the two prior ticks. Overruns aren't +// loud failures: throttled polls age into the live fallback at ~3× cost, +// billed to the customer, so keep real headroom under the cap. +// 1,000/tick ≈ 288,000 units/day, ~45× steady-state demand — it only binds +// during backlog catch-up. +const SCHEDULED_TASK_UNIT_BUDGET = 1000; + +// Wall-clock guard for the per-config loop: sub-hourly crons are killed at 15 +// minutes, and a skip-heavy tick pays serial Autumn round-trips per distinct +// org (worst case minutes, more when Autumn is degraded). Stopping early is +// safe — unprocessed configs stay due and the next tick resumes oldest-first. +const TICK_DEADLINE_MS = 3 * 60_000; // Cap on the per-tick list of configs blocked by an active run. Blocked // configs leave no durable trace on their row, so the summary names them. @@ -43,9 +58,11 @@ export async function runScheduledRankChecks(env: Env) { return check; }; + const deadline = Date.now() + TICK_DEADLINE_MS; let unitsStarted = 0; let started = 0; let stoppedByBudget = false; + let stoppedByDeadline = false; let skippedFree = 0; let skippedNoKeywords = 0; let concurrentChangeSkips = 0; @@ -56,6 +73,10 @@ export async function runScheduledRankChecks(env: Env) { let configErrors = 0; for (const config of dueConfigs) { + if (Date.now() >= deadline) { + stoppedByDeadline = true; + break; + } // Per-config containment: one bad row (e.g. malformed next_check_at, which // sorts first and would head every scan) or transient DB error must not // starve the rest of the tick or suppress the summary log. @@ -219,6 +240,7 @@ export async function runScheduledRankChecks(env: Env) { unitsStarted, budget: SCHEDULED_TASK_UNIT_BUDGET, stoppedByBudget, + stoppedByDeadline, skippedFree, skippedNoKeywords, concurrentChangeSkips, diff --git a/src/server/workflows/rankCheckPaths.ts b/src/server/workflows/rankCheckPaths.ts index f77971d..949ce39 100644 --- a/src/server/workflows/rankCheckPaths.ts +++ b/src/server/workflows/rankCheckPaths.ts @@ -166,7 +166,7 @@ const QUEUED_POLL_INTERVALS = [ /** Concurrent task_get requests within a collect step. */ const TASK_GET_CONCURRENCY = 25; -/** Max task_get calls per collect round (per-invocation subrequest budget). */ +/** Max task_get calls per collect round (bounds one round's fan-out). */ const TASK_GETS_PER_COLLECT = 500; // Collect steps may issue hundreds of task_get calls, so they get more room