diff --git a/src/app/api/environment/backfill/__tests__/route.test.ts b/src/app/api/environment/backfill/__tests__/route.test.ts new file mode 100644 index 000000000..2d1cea45f --- /dev/null +++ b/src/app/api/environment/backfill/__tests__/route.test.ts @@ -0,0 +1,117 @@ +/** + * v1.42 — one account must not drain the instance-wide Open-Meteo budget. + * + * The backfill route shared the analytics-read bucket (120 a minute) and sent + * every explicit range without a key, so one account could queue a + * two-year backfill again and again. It is now held to three a rolling hour, + * and to one queued backfill per twenty-minute slot, keyed on the account. + */ +import { NextRequest } from "next/server"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { send, store } = vi.hoisted(() => ({ + send: vi.fn(), + store: new Map(), +})); + +vi.mock("@/lib/api-handler", async () => { + const actual = + await vi.importActual( + "@/lib/api-handler", + ); + return { + ...actual, + apiHandler: Promise>( + h: T, + ): T => h, + requireAuth: vi.fn(async () => ({ user: { id: "acct-1" } })), + }; +}); +vi.mock("@/lib/modules/gate", () => ({ + requireModuleEnabled: vi.fn(async () => ({ enabled: true })), +})); +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: () => ({ send }), +})); +vi.mock("@/lib/rate-limit", async () => { + const actual = + await vi.importActual( + "@/lib/rate-limit", + ); + const counted = (limit: number) => async (key: string) => { + const count = (store.get(key) ?? 0) + 1; + store.set(key, count); + return { + allowed: count <= limit, + limit, + remaining: Math.max(0, limit - count), + resetAt: Date.now() + 1000, + }; + }; + return { + ...actual, + checkAnalyticsReadRateLimit: (id: string) => + counted(120)(`analytics-read:${id}`), + checkEnvironmentBackfillRateLimit: (id: string) => + counted(actual.ENVIRONMENT_BACKFILL_LIMIT)(`environment-backfill:${id}`), + }; +}); +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + findUnique: vi.fn(async () => ({ + homeLat: null, + homeLocationEncrypted: new Uint8Array([1]), + homeSince: new Date("2025-01-01T00:00:00Z"), + })), + }, + }, +})); + +import { POST } from "../route"; + +function post() { + return new NextRequest("http://localhost/api/environment/backfill", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ startDate: "2025-01-01", endDate: "2026-10-01" }), + }); +} + +beforeEach(() => { + store.clear(); + send.mockReset().mockResolvedValue("job-id"); +}); + +describe("POST /api/environment/backfill budget protection", () => { + it("sends with a per-account singleton key and slot", async () => { + const res = await POST(post()); + expect(res.status).toBe(202); + expect(send).toHaveBeenCalledWith( + "environment-fetch", + expect.objectContaining({ userId: "acct-1" }), + expect.objectContaining({ + singletonKey: "environment-backfill:acct-1", + singletonSeconds: 20 * 60, + }), + ); + }); + + it("answers 409 when the account's slot already holds a backfill", async () => { + send.mockResolvedValue(null); + const res = await POST(post()); + expect(res.status).toBe(409); + const body = await res.json(); + expect(body.meta.errorCode).toBe("environment.backfill_pending"); + }); + + it("refuses the fourth backfill in an hour with 429", async () => { + for (let i = 0; i < 3; i++) { + expect((await POST(post())).status).toBe(202); + } + const res = await POST(post()); + expect(res.status).toBe(429); + expect(send).toHaveBeenCalledTimes(3); + }); +}); diff --git a/src/app/api/environment/backfill/route.ts b/src/app/api/environment/backfill/route.ts index 7c6f5f5ed..690a338ea 100644 --- a/src/app/api/environment/backfill/route.ts +++ b/src/app/api/environment/backfill/route.ts @@ -6,6 +6,11 @@ * The span is capped (the worker re-checks the cap too). Requires a home * location (otherwise there is nothing to resolve). Module-gated + rate-limited; * `userId` is narrowed from auth. + * + * Every backfill draws on the instance-wide Open-Meteo budget, so one account + * is held to three a rolling hour (429), one queued per twenty minutes (409 + * `environment.backfill_pending`), and its own share of the daily budget in + * the worker (`OPEN_METEO_ACCOUNT_DAY_SHARE`). */ import { NextRequest } from "next/server"; @@ -18,7 +23,10 @@ import { } from "@/lib/api-response"; import { annotate } from "@/lib/logging/context"; import { requireModuleEnabled } from "@/lib/modules/gate"; -import { checkAnalyticsReadRateLimit } from "@/lib/rate-limit"; +import { + checkAnalyticsReadRateLimit, + checkEnvironmentBackfillRateLimit, +} from "@/lib/rate-limit"; import { prisma } from "@/lib/db"; import { environmentBackfillSchema } from "@/lib/validations/environment"; import { @@ -26,7 +34,7 @@ import { defaultBackfillRange, utcDayKey, } from "@/lib/environment/service"; -import { enqueueEnvironmentFetch } from "@/lib/jobs/environment-fetch"; +import { enqueueEnvironmentBackfill } from "@/lib/jobs/environment-fetch"; const MS_PER_DAY = 24 * 60 * 60 * 1000; @@ -96,11 +104,24 @@ export const POST = apiHandler(async (request: NextRequest) => { ); } - const enqueued = await enqueueEnvironmentFetch({ + const backfillLimit = await checkEnvironmentBackfillRateLimit(user.id); + if (!backfillLimit.allowed) { + return apiError("Too many backfills. Please retry later.", 429, { + errorCode: "environment.backfill_rate_limited", + }); + } + + const outcome = await enqueueEnvironmentBackfill({ userId: user.id, startDate, endDate, }); + if (outcome === "already_queued") { + return apiError("A backfill for this account is already queued.", 409, { + errorCode: "environment.backfill_pending", + }); + } + const enqueued = outcome === "enqueued"; annotate({ action: { name: "environment.backfill.enqueue" }, diff --git a/src/lib/environment/__tests__/request-budget.test.ts b/src/lib/environment/__tests__/request-budget.test.ts index a03c21364..b0426a31e 100644 --- a/src/lib/environment/__tests__/request-budget.test.ts +++ b/src/lib/environment/__tests__/request-budget.test.ts @@ -40,6 +40,7 @@ vi.mock("@/lib/db", () => { }); import { + OPEN_METEO_ACCOUNT_DAY_WINDOW, OPEN_METEO_BUDGET_WINDOWS, admitWeight, openMeteoCallWeight, @@ -113,3 +114,23 @@ describe("reserveOpenMeteoCalls", () => { expect(store.get("open-meteo-budget:day")!.count).toBe(dayBefore); }); }); + +describe("the per-account share", () => { + it("refuses one account past its share while another account still gets through", async () => { + store.clear(); + expect(OPEN_METEO_ACCOUNT_DAY_WINDOW.limit).toBe(400); + // A two-year backfill weighs about 63 calls; six of them fit in the + // share, the seventh does not. + for (let i = 0; i < 6; i++) { + expect(await reserveOpenMeteoCalls(62.57, "acct-a")).toBe(true); + } + expect(await reserveOpenMeteoCalls(62.57, "acct-a")).toBe(false); + expect(store.get("open-meteo-budget:account:acct-a:day")!.count).toBe( + 6 * 6257, + ); + // The instance day is far from full, so another account is admitted. + expect(await reserveOpenMeteoCalls(62.57, "acct-b")).toBe(true); + // A request made for no account is bound by the instance windows only. + expect(await reserveOpenMeteoCalls(1)).toBe(true); + }); +}); diff --git a/src/lib/environment/open-meteo-air-quality.ts b/src/lib/environment/open-meteo-air-quality.ts index 6471bb967..4d2c8a917 100644 --- a/src/lib/environment/open-meteo-air-quality.ts +++ b/src/lib/environment/open-meteo-air-quality.ts @@ -350,13 +350,16 @@ export interface AirQualityFetchResult { * problem: the caller gets what was fetched and why it stopped, and the days * not returned stay unfetched for the next run. */ -export async function fetchDailyAirQuality(args: { - lat: number; - lon: number; - timezone: string; - startDate: string; - endDate: string; -}): Promise { +export async function fetchDailyAirQuality( + args: { + lat: number; + lon: number; + timezone: string; + startDate: string; + endDate: string; + }, + budget: { accountId?: string } = {}, +): Promise { const days: DailyAirQualityObservation[] = []; // Days before the feed's first day are answered here, without a request. for ( @@ -379,7 +382,7 @@ export async function fetchDailyAirQuality(args: { variables.length, enumerateDayCount(chunk.startDate, chunk.endDate), ); - if (!(await reserveOpenMeteoCalls(weight))) { + if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { return { days, stopped: "budget" }; } try { diff --git a/src/lib/environment/open-meteo.ts b/src/lib/environment/open-meteo.ts index 60d1b9a41..f936d5c4a 100644 --- a/src/lib/environment/open-meteo.ts +++ b/src/lib/environment/open-meteo.ts @@ -256,20 +256,24 @@ export async function geocodeLocation( * cloud) are requested alongside and aggregated to per-day values here. Throws * a {@link SafeFetchError} on egress failure so the caller can classify it, * and an {@link OpenMeteoBudgetExhaustedError} when the instance-wide request - * budget refuses the call (nothing is sent then). + * budget, or the account's share of it (`budget.accountId`), refuses the call + * (nothing is sent then). The account never reaches the request. */ -export async function fetchDailyEnvironment(args: { - lat: number; - lon: number; - timezone: string; - startDate: string; - endDate: string; -}): Promise { +export async function fetchDailyEnvironment( + args: { + lat: number; + lon: number; + timezone: string; + startDate: string; + endDate: string; + }, + budget: { accountId?: string } = {}, +): Promise { const weight = openMeteoCallWeight( ARCHIVE_VARIABLE_COUNT, enumerateDayCount(args.startDate, args.endDate), ); - if (!(await reserveOpenMeteoCalls(weight))) { + if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { throw new OpenMeteoBudgetExhaustedError(); } const url = new URL(`${ARCHIVE_BASE_URL}/v1/archive`); diff --git a/src/lib/environment/request-budget.ts b/src/lib/environment/request-budget.ts index d72980b12..58f0c31c5 100644 --- a/src/lib/environment/request-budget.ts +++ b/src/lib/environment/request-budget.ts @@ -34,7 +34,7 @@ import { prisma } from "@/lib/db"; /** One budget window. */ export interface BudgetWindow { /** Bucket key suffix, also the window's name. */ - name: "minute" | "hour" | "day"; + name: "minute" | "hour" | "day" | "account-day"; windowMs: number; /** Calls admitted per window (whole calls). */ limit: number; @@ -46,6 +46,26 @@ export const OPEN_METEO_BUDGET_WINDOWS: readonly BudgetWindow[] = [ { name: "day", windowMs: 86_400_000, limit: 8_000 }, ]; +/** + * The share of the daily budget one account may use. The instance-wide + * windows alone let a single account drain the day for everyone (repeated + * backfills of a two-year range), so a request made for an account is also + * charged to that account's own daily bucket. One twentieth of the day is + * 400 calls: more than two full backfills of weather and air quality, and + * still leaves the rest of the day for nineteen other accounts. + */ +export const OPEN_METEO_ACCOUNT_DAY_SHARE = 0.05; + +const DAY_LIMIT = + OPEN_METEO_BUDGET_WINDOWS.find((w) => w.name === "day")?.limit ?? 0; + +/** The per-account daily window, charged only when a request names an account. */ +export const OPEN_METEO_ACCOUNT_DAY_WINDOW: BudgetWindow = { + name: "account-day", + windowMs: 86_400_000, + limit: Math.floor(DAY_LIMIT * OPEN_METEO_ACCOUNT_DAY_SHARE), +}; + const BUCKET_PREFIX = "open-meteo-budget"; const LOCK_KEY = "open-meteo-budget"; @@ -97,13 +117,23 @@ export function admitWeight( /** * Ask the budget for one request of `weight` calls. Returns true and charges * every window when it fits, false (charging nothing) when any window is - * full. + * full. With `accountId`, the account's own daily share is checked and + * charged as well (`OPEN_METEO_ACCOUNT_DAY_SHARE`); requests made for no + * particular account (the geocoder) are bound by the instance windows only. */ -export async function reserveOpenMeteoCalls(weight: number): Promise { +export async function reserveOpenMeteoCalls( + weight: number, + accountId?: string, +): Promise { const weightCenti = Math.max(1, Math.ceil(weight * CENTI)); - const keys = OPEN_METEO_BUDGET_WINDOWS.map( - (w) => `${BUCKET_PREFIX}:${w.name}`, - ); + const windows: readonly BudgetWindow[] = accountId + ? [...OPEN_METEO_BUDGET_WINDOWS, OPEN_METEO_ACCOUNT_DAY_WINDOW] + : OPEN_METEO_BUDGET_WINDOWS; + const keyOf = (name: BudgetWindow["name"]) => + name === "account-day" + ? `${BUCKET_PREFIX}:account:${accountId}:day` + : `${BUCKET_PREFIX}:${name}`; + const nameOfKey = new Map(windows.map((w) => [keyOf(w.name), w.name])); return prisma.$transaction(async (tx) => { // `pg_advisory_xact_lock` returns void, which the client cannot // deserialize as a column; selecting FROM it yields a plain row. @@ -112,21 +142,18 @@ export async function reserveOpenMeteoCalls(weight: number): Promise { FROM pg_advisory_xact_lock(hashtextextended(${LOCK_KEY}, 0)) `; const rows = await tx.rateLimit.findMany({ - where: { key: { in: keys } }, + where: { key: { in: [...nameOfKey.keys()] } }, select: { key: true, count: true, resetAt: true }, }); const buckets = new Map(); for (const row of rows) { - const name = row.key.slice(BUCKET_PREFIX.length + 1) as - BudgetWindow["name"] | string; - if (name === "minute" || name === "hour" || name === "day") { - buckets.set(name, { count: row.count, resetAt: row.resetAt }); - } + const name = nameOfKey.get(row.key); + if (name) buckets.set(name, { count: row.count, resetAt: row.resetAt }); } - const decision = admitWeight(buckets, weightCenti, new Date()); + const decision = admitWeight(buckets, weightCenti, new Date(), windows); if (!decision.admitted) return false; for (const [name, bucket] of decision.next) { - const key = `${BUCKET_PREFIX}:${name}`; + const key = keyOf(name); await tx.rateLimit.upsert({ where: { key }, create: { key, count: bucket.count, resetAt: bucket.resetAt }, diff --git a/src/lib/environment/service.ts b/src/lib/environment/service.ts index b0e5b9277..bd26c43f0 100644 --- a/src/lib/environment/service.ts +++ b/src/lib/environment/service.ts @@ -330,13 +330,16 @@ export async function fetchAndStoreEnvironment(args: { const max = groupDays[groupDays.length - 1]; let observations: DailyEnvironmentObservation[]; try { - observations = await fetchDailyEnvironment({ - lat: loc.lat, - lon: loc.lon, - timezone, - startDate: min, - endDate: max, - }); + observations = await fetchDailyEnvironment( + { + lat: loc.lat, + lon: loc.lon, + timezone, + startDate: min, + endDate: max, + }, + { accountId: userId }, + ); } catch (err) { if (err instanceof OpenMeteoBudgetExhaustedError) { // Nothing was sent. The remaining groups wait for the next run. @@ -354,13 +357,16 @@ export async function fetchAndStoreEnvironment(args: { // air quality; the weather is stored regardless. let airByDay: Map | null = null; if (airQuality) { - const result = await fetchDailyAirQuality({ - lat: loc.lat, - lon: loc.lon, - timezone, - startDate: min, - endDate: max, - }); + const result = await fetchDailyAirQuality( + { + lat: loc.lat, + lon: loc.lon, + timezone, + startDate: min, + endDate: max, + }, + { accountId: userId }, + ); if (result.stopped === "budget") budgetBlocked = true; airByDay = new Map(result.days.map((d) => [d.date, d])); fetches += result.days.length > 0 ? 1 : 0; @@ -574,13 +580,16 @@ export async function fillAirQualityGaps( let filled = 0; for (const group of byLocation.values()) { for (const range of nearbyRanges(group.days)) { - const result = await fetchDailyAirQuality({ - lat: group.lat, - lon: group.lon, - timezone, - startDate: range[0], - endDate: range[range.length - 1], - }); + const result = await fetchDailyAirQuality( + { + lat: group.lat, + lon: group.lon, + timezone, + startDate: range[0], + endDate: range[range.length - 1], + }, + { accountId: userId }, + ); const wanted = new Set(range); const now = new Date(); const ops = result.days diff --git a/src/lib/jobs/environment-fetch.ts b/src/lib/jobs/environment-fetch.ts index d069d9ae3..23c838882 100644 --- a/src/lib/jobs/environment-fetch.ts +++ b/src/lib/jobs/environment-fetch.ts @@ -103,6 +103,32 @@ export async function enqueueEnvironmentFetch( return true; } +/** One account's backfill slot: at most one backfill job per slot. */ +export const ENVIRONMENT_BACKFILL_SINGLETON_SECONDS = 20 * 60; + +/** + * Enqueue the on-demand backfill a person asked for from the settings + * surface. Unlike the travel-period refresh, it carries a per-account + * `singletonKey` with a time slot, so one account has at most one backfill + * per twenty minutes no matter how often the button is pressed; pg-boss + * enforces the slot (`singleton_on`) under every queue policy. A send the + * slot refuses comes back as `already_queued`, and the route says so instead + * of reporting a job that will not run. + */ +export async function enqueueEnvironmentBackfill(payload: { + userId: string; + startDate: string; + endDate: string; +}): Promise<"enqueued" | "no_worker" | "already_queued"> { + const boss = getGlobalBoss(); + if (!boss) return "no_worker"; + const id = await boss.send(ENVIRONMENT_FETCH_QUEUE, payload, { + singletonKey: `environment-backfill:${payload.userId}`, + singletonSeconds: ENVIRONMENT_BACKFILL_SINGLETON_SECONDS, + }); + return id ? "enqueued" : "already_queued"; +} + /** * Discovery fan-out: enqueue one per-user fetch job for every account with the * module on and a home location set. Idempotent (singletonKey-coalesced per diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 72a5a0989..801a38ec1 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -754,8 +754,10 @@ const schedules: ScheduleEntry[] = [ * Any policy would collapse every keyless explicit-range backfill onto the * shared empty key, so two different requested date ranges would silently * become one. That is exactly the class of silent work-dropping a policy is - * supposed to prevent, so the queue keeps `standard` until the enqueue side - * gives the explicit-range variant a key of its own. + * supposed to prevent, so the queue keeps `standard`. The settings + * backfill (v1.42) is held per account by a keyed time slot + * (`singletonSeconds`), which pg-boss enforces under any policy; the + * travel-period refresh stays keyless so two periods never merge. * - APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, * DATA_BACKUP_QUEUE and PR_DETECTION_QUEUE send keylessly by design (each * import / backup / detection run is a distinct unit of work). A policy diff --git a/src/lib/openapi/error-codes.ts b/src/lib/openapi/error-codes.ts index 06c7a87bc..95f87bb50 100644 --- a/src/lib/openapi/error-codes.ts +++ b/src/lib/openapi/error-codes.ts @@ -177,6 +177,8 @@ export const ERROR_CODE_CATALOGUE: Readonly> = ], encryption: ["encryption.keyBackup.stale", "encryption.key_mismatch"], environment: [ + "environment.backfill_pending", + "environment.backfill_rate_limited", "environment.invalid", "environment.no_home", "environment.range_too_large", diff --git a/src/lib/openapi/routes/environment.ts b/src/lib/openapi/routes/environment.ts index 39faae489..c14803c29 100644 --- a/src/lib/openapi/routes/environment.ts +++ b/src/lib/openapi/routes/environment.ts @@ -361,7 +361,7 @@ export const environmentPaths: NonNullable = { post: { tags: ["Environment"], summary: "Enqueue a historical environment fetch", - description: `Asks the worker to fetch and store past days rather than waiting for the rolling nightly lookback. Returns 202 — the work has been QUEUED, not done; poll \`GET /api/environment\` and watch \`context.days\`.\n\nAn omitted bound defaults to the conservative \`[homeSince .. today]\` range. An explicit start earlier than \`homeSince\` is accepted and then resolves to a skip inside the worker: the span cannot fabricate weather for the pre-home past, so a wide range can legitimately return \`enqueued: true\` and add no days at all.\n\nThe span is capped at ${ENVIRONMENT_MAX_BACKFILL_DAYS} days, and the worker re-checks the cap independently. Body cap 4 KiB. Module-gated, and it draws on the shared analytics-read rate bucket.`, + description: `Asks the worker to fetch and store past days rather than waiting for the rolling nightly lookback. Returns 202 — the work has been QUEUED, not done; poll \`GET /api/environment\` and watch \`context.days\`.\n\nAn omitted bound defaults to the conservative \`[homeSince .. today]\` range. An explicit start earlier than \`homeSince\` is accepted and then resolves to a skip inside the worker: the span cannot fabricate weather for the pre-home past, so a wide range can legitimately return \`enqueued: true\` and add no days at all.\n\nThe span is capped at ${ENVIRONMENT_MAX_BACKFILL_DAYS} days, and the worker re-checks the cap independently. Body cap 4 KiB. Module-gated, and it draws on the shared analytics-read rate bucket.\n\nEvery backfill draws on the instance-wide Open-Meteo budget, so one account is held to three a rolling hour (429 \`environment.backfill_rate_limited\`), at most one queued backfill per twenty-minute slot (409 \`environment.backfill_pending\`), and, inside the worker, its own share of the daily request budget; days the share does not reach are filled by the nightly run.`, requestBody: { required: true, content: { "application/json": { schema: backfillRequest } }, @@ -391,7 +391,7 @@ export const environmentPaths: NonNullable = { ...moduleDisabledResponse, "409": { description: - "No home location is set, so there is nothing to resolve the days against. `meta.errorCode` = `environment.no_home`.", + "No home location is set, so there is nothing to resolve the days against (`meta.errorCode` = `environment.no_home`), or a backfill for this account is already queued in the current twenty-minute slot (`environment.backfill_pending`).", content: { "application/json": { schema: errorEnvelope } }, }, "413": { diff --git a/src/lib/rate-limit.ts b/src/lib/rate-limit.ts index 5a8816b4b..6914c167c 100644 --- a/src/lib/rate-limit.ts +++ b/src/lib/rate-limit.ts @@ -117,6 +117,24 @@ export async function checkAnalyticsReadRateLimit( ); } +/** + * Environment backfill: three a rolling hour per account. Each one can ask + * the shared Open-Meteo budget for a two-year range, so the general + * analytics ceiling (120 a minute) is far too loose for it. + */ +export const ENVIRONMENT_BACKFILL_LIMIT = 3; +const ENVIRONMENT_BACKFILL_WINDOW_MS = 60 * 60 * 1000; + +export async function checkEnvironmentBackfillRateLimit( + userId: string, +): Promise { + return checkRateLimit( + `environment-backfill:${userId}`, + ENVIRONMENT_BACKFILL_LIMIT, + ENVIRONMENT_BACKFILL_WINDOW_MS, + ); +} + /** * Shared per-account ceiling for the single-record writes. *