diff --git a/src/lib/environment/__tests__/air-quality-history.test.ts b/src/lib/environment/__tests__/air-quality-history.test.ts index 4b58cccb4..2d96853e5 100644 --- a/src/lib/environment/__tests__/air-quality-history.test.ts +++ b/src/lib/environment/__tests__/air-quality-history.test.ts @@ -1,19 +1,19 @@ /** * v1.42 — the air-quality history backfill's pure parts: how pending days * are cut into request ranges, what a range weighs, the ceilings that keep - * the history from taking the nightly fetch's room, where a day without a - * row is placed, and when an account is offered a run again. + * the history from taking the nightly fetch's room, the entry-day snapshot a + * chain hands on, and when an account is offered a run again. */ import { describe, expect, it } from "vitest"; import { AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS, - historyRangeWeight, - historyRequestFits, + AIR_QUALITY_HISTORY_CEILING, planHistoryRanges, readAirQualityHistoryState, - resolveHistoryLocation, + readEntryDaysSnapshot, } from "../air-quality-history"; +import { airQualityChunkWeight } from "../open-meteo-air-quality"; import { historyDue } from "@/lib/jobs/environment-air-quality-history"; import { shiftDateKey } from "@/lib/tz/format"; @@ -64,100 +64,35 @@ describe("planHistoryRanges", () => { const days = daysFrom("2023-01-01", 122, 3); const ranges = planHistoryRanges(days); const ranged = ranges.reduce( - (sum, r) => sum + historyRangeWeight(r, []), + (sum, r) => sum + airQualityChunkWeight(r.startDate, r.endDate), 0, ); - const perDay = - days.length * - historyRangeWeight( - { startDate: days[0], endDate: days[0], days: [days[0]] }, - [], - ); + const perDay = days.length * airQualityChunkWeight(days[0], days[0]); expect(ranged).toBeLessThan(perDay / 4); }); }); -describe("historyRangeWeight", () => { - it("adds the weather only over the span of the days without a row", () => { - const range = { - startDate: "2023-01-01", - endDate: "2023-03-31", - days: ["2023-01-01", "2023-03-31"], - }; - const airOnly = historyRangeWeight(range, []); - expect(airOnly).toBeCloseTo((1.7 * 90) / 14); - // One missing day: one weather request at its minimum weight (12 vars). - expect(historyRangeWeight(range, ["2023-03-31"])).toBeCloseTo( - airOnly + 1.2, - ); - }); -}); - -describe("historyRequestFits", () => { - it("admits a request under every ceiling", () => { - expect(historyRequestFits({}, 10)).toBe(true); - }); - - it("holds the account to three quarters of its daily share", () => { +describe("the history's ceiling", () => { + it("holds the history to three quarters of the account's share and half of each instance window", () => { expect(AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS).toBe(300); - expect(historyRequestFits({ "account-day": 295 }, 5)).toBe(true); - expect(historyRequestFits({ "account-day": 295 }, 5.1)).toBe(false); - }); - - it("holds the history to half of each instance window", () => { - expect(historyRequestFits({ minute: 245 }, 5)).toBe(true); - expect(historyRequestFits({ minute: 246 }, 5)).toBe(false); - expect(historyRequestFits({ hour: 1999 }, 2)).toBe(false); - expect(historyRequestFits({ day: 3995 }, 10)).toBe(false); + expect(AIR_QUALITY_HISTORY_CEILING).toEqual({ + instanceShare: 0.5, + accountDayCalls: 300, + }); }); }); -describe("resolveHistoryLocation", () => { - const home = { - lat: 51.5, - lon: 7.2, - label: "Home", - timezone: "Europe/Berlin", - since: "2026-09-01", - }; - const trip = { - startDate: "2024-07-01", - endDate: "2024-07-14", - lat: 38.7, - lon: -9.1, - label: "Lisbon", - }; - - it("places a day of a trip at the trip", () => { - expect(resolveHistoryLocation("2024-07-05", home, [trip])).toMatchObject({ - lat: 38.7, - source: "TRAVEL", - }); - }); - - it("places a day from the home's effective date on at the home", () => { - expect(resolveHistoryLocation("2026-09-01", home, [trip])).toMatchObject({ - lat: 51.5, - source: "HOME", - }); - }); - - it("leaves a day before the home was set without a place, unless a period covers it", () => { - expect(resolveHistoryLocation("2026-08-31", home, [trip])).toBeNull(); - expect(resolveHistoryLocation("2020-03-01", home, [trip])).toBeNull(); - expect(resolveHistoryLocation("2024-07-14", home, [trip])).toMatchObject({ - source: "TRAVEL", - }); - }); - - it("places every day at a home stored without an effective date", () => { - expect( - resolveHistoryLocation("2015-01-01", { ...home, since: null }, []), - ).toMatchObject({ lat: 51.5, source: "HOME" }); - }); - - it("places nothing without a home or a period", () => { - expect(resolveHistoryLocation("2020-03-01", null, [trip])).toBeNull(); +describe("entry-day snapshot", () => { + it("reads a snapshot from a payload and refuses anything else", () => { + const snapshot = { + timezone: "Europe/Berlin", + cutoff: "2026-10-01", + days: ["2024-01-01"], + }; + expect(readEntryDaysSnapshot(snapshot)).toEqual(snapshot); + expect(readEntryDaysSnapshot(undefined)).toBeNull(); + expect(readEntryDaysSnapshot({ ...snapshot, days: [1] })).toBeNull(); + expect(readEntryDaysSnapshot({ ...snapshot, cutoff: 1 })).toBeNull(); }); }); diff --git a/src/lib/environment/__tests__/request-budget.test.ts b/src/lib/environment/__tests__/request-budget.test.ts index b0426a31e..49fc49fc8 100644 --- a/src/lib/environment/__tests__/request-budget.test.ts +++ b/src/lib/environment/__tests__/request-budget.test.ts @@ -45,6 +45,7 @@ import { admitWeight, openMeteoCallWeight, reserveOpenMeteoCalls, + windowsUnderCeiling, type BudgetBucket, } from "../request-budget"; @@ -134,3 +135,44 @@ describe("the per-account share", () => { expect(await reserveOpenMeteoCalls(1)).toBe(true); }); }); + +describe("a caller's own ceiling", () => { + const ceiling = { instanceShare: 0.5, accountDayCalls: 300 }; + + it("lowers each instance window to its share and the account's day to its calls", () => { + const windows = windowsUnderCeiling( + [...OPEN_METEO_BUDGET_WINDOWS, OPEN_METEO_ACCOUNT_DAY_WINDOW], + ceiling, + ); + expect(windows.map((w) => [w.name, w.limit, w.windowMs])).toEqual([ + ["minute", 250, 60_000], + ["hour", 2_000, 3_600_000], + ["day", 4_000, 86_400_000], + ["account-day", 300, 86_400_000], + ]); + expect(windowsUnderCeiling(OPEN_METEO_BUDGET_WINDOWS, undefined)).toBe( + OPEN_METEO_BUDGET_WINDOWS, + ); + }); + + it("refuses past the ceiling what the budget alone would admit, and charges nothing", async () => { + store.clear(); + const later = new Date(Date.now() + 3_600_000); + store.set("open-meteo-budget:account:acct-c:day", { + count: 295 * 100, + resetAt: later, + }); + expect(await reserveOpenMeteoCalls(5, "acct-c", ceiling)).toBe(true); + expect(await reserveOpenMeteoCalls(0.1, "acct-c", ceiling)).toBe(false); + expect(store.get("open-meteo-budget:account:acct-c:day")!.count).toBe( + 300 * 100, + ); + // The same request without the ceiling is still inside the budget. + expect(await reserveOpenMeteoCalls(0.1, "acct-c")).toBe(true); + + store.clear(); + store.set("open-meteo-budget:minute", { count: 246 * 100, resetAt: later }); + expect(await reserveOpenMeteoCalls(5, "acct-c", ceiling)).toBe(false); + expect(await reserveOpenMeteoCalls(4, "acct-c", ceiling)).toBe(true); + }); +}); diff --git a/src/lib/environment/air-quality-history.ts b/src/lib/environment/air-quality-history.ts index 7d2ea80a2..c192eff28 100644 --- a/src/lib/environment/air-quality-history.ts +++ b/src/lib/environment/air-quality-history.ts @@ -15,13 +15,15 @@ * (only the air-quality part is fetched then, as the nightly gap fill * does); * - otherwise in the dated location period that covers it, else at the - * home, but the home only from `homeSince` on (or on every day when no - * `homeSince` is stored). A day before the home was set is never placed - * there: the weather and air of a place the person was not at would be - * wrong data in the correlations, the Coach and the day view, and an - * empty day is honest where a wrong one is not. Such a day is filled - * only when a period covers it. A day without a row gets its weather row - * with the air quality, because the air quality is stored on it. + * home, but the home only from `homeSince` on, by the same rule the + * nightly fetch places a day with (`resolveLocationForDay`). A home + * without a `homeSince` places no day at all, there as here. A day + * before the home was set is never placed there: the weather and air of + * a place the person was not at would be wrong data in the + * correlations, the Coach and the day view, and an empty day is honest + * where a wrong one is not. Such a day is filled only when a period + * covers it. A day without a row gets its weather row with the air + * quality, because the air quality is stored on it. * * A day inside a period whose sealed location does not open is skipped (fail * closed, as in the service), and so is a day with no place at all; neither @@ -40,7 +42,19 @@ * over 14 (`openMeteoCallWeight`), so a lone day costs as much as two weeks: * joining short gaps is free, splitting long ones saves the gap. The newest * ranges go first, so the window the correlations read fills first. The - * weather of a range is asked only over the span of its days without a row. + * weather of a range is asked only over the span of its days without a row, + * and before its air quality: a weather request the feed or the budget + * refuses then costs no air-quality call, and when the air quality fails + * after the weather came back, the weather rows are stored without it, so + * the next run asks for the air quality alone and nothing paid is asked for + * twice. A day the archive answer leaves out gets no row: it stays pending + * rather than carrying an empty weather row that nothing would fill again. + * + * The first request the feed refuses (a 429, a 5xx, a network error) ends + * the run with status `error` and no follow-up; the next nightly discovery + * tries again. Before every range the run re-reads the module, the + * account's air-quality switch and `ENVIRONMENT_AIR_QUALITY_DISABLED`, so + * switching any of them off stops a run between two ranges. * * ## Budget * @@ -48,7 +62,11 @@ * share (`request-budget.ts`). On top of that the history holds itself to * {@link AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS} of the account's 400 daily * calls and to half of each instance window, so the nightly fetch, the gap - * fill and a backfill someone asked for keep their room. A run that reaches a + * fill and a backfill someone asked for keep their room. The ceiling is + * checked in the budget's own locked transaction, together with the charge + * ({@link AIR_QUALITY_HISTORY_CEILING}), so two worker processes, or the + * history beside the nightly fetch, cannot both read room only one of them + * gets. A run that reaches a * ceiling stops; the nightly discovery picks the account up again, so a long * history spreads over several days by construction. * @@ -56,14 +74,16 @@ * * A day is done once its row carries `aqFetchedAt`; done days are never asked * for again, and the writes only touch rows still without it (or create the - * row). Each run stores `{ total, done, checkedAt, complete }` on + * row). The days with entries are read once per chain and handed from run + * to run in the follow-up's payload ({@link EntryDaysSnapshot}), so a chain + * of fifty runs scans the entry tables once, not fifty times. Each run + * stores `{ total, done, checkedAt, complete }` on * `User.environmentAqHistoryJson`, which the settings card shows as * "Backfilling air quality: x of y days". */ import { Prisma } from "@/generated/prisma/client"; import { prisma, toJson } from "@/lib/db"; import { - airQualityChunkWeight, earliestAirQualityDay, fetchDailyAirQuality, isAirQualityActive, @@ -73,16 +93,13 @@ import { type DailyAirQualityObservation, } from "@/lib/environment/open-meteo-air-quality"; import { - archiveRequestWeight, fetchDailyEnvironment, type DailyEnvironmentObservation, } from "@/lib/environment/open-meteo"; import { OPEN_METEO_ACCOUNT_DAY_WINDOW, - OPEN_METEO_BUDGET_WINDOWS, OpenMeteoBudgetExhaustedError, - readOpenMeteoBudgetUsage, - type BudgetUsage, + type BudgetCeiling, } from "@/lib/environment/request-budget"; import { readLocation, sealLocation } from "@/lib/environment/location-cipher"; import { @@ -93,11 +110,11 @@ import { inAnyPeriod, readTravelOverrides, resolveHome, + resolveLocationForDay, utcDayKey, - type HomeLocation, type ResolvedLocation, - type TravelOverride, } from "@/lib/environment/service"; +import { isModuleEnabled } from "@/lib/modules/gate"; import { shiftDateKey } from "@/lib/tz/format"; import { startOfLocalDayKey } from "@/lib/tz/local-day"; @@ -115,6 +132,11 @@ export const AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS = Math.floor( ); /** The history's share of each instance-wide window. */ export const AIR_QUALITY_HISTORY_INSTANCE_SHARE = 0.5; +/** Both ceilings, as every history request hands them to the budget. */ +export const AIR_QUALITY_HISTORY_CEILING: BudgetCeiling = { + instanceShare: AIR_QUALITY_HISTORY_INSTANCE_SHARE, + accountDayCalls: AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS, +}; /** One pending day and where it is fetched. */ export interface HistoryDay { @@ -159,60 +181,6 @@ export function planHistoryRanges( return ranges; } -/** - * The weight of one range in calls: its air-quality request, plus the - * weather request over the span of its days without a row, when it has any. - */ -export function historyRangeWeight( - range: HistoryRange, - missingRowDays: readonly string[], -): number { - const air = airQualityChunkWeight(range.startDate, range.endDate); - if (missingRowDays.length === 0) return air; - const sorted = [...missingRowDays].sort(); - return air + archiveRequestWeight(sorted[0], sorted[sorted.length - 1]); -} - -/** - * Whether a request of `weight` calls fits under the history's ceilings, given - * what each window already holds. Pure; exported for tests. - */ -export function historyRequestFits( - usage: BudgetUsage, - weight: number, -): boolean { - for (const w of OPEN_METEO_BUDGET_WINDOWS) { - const ceiling = w.limit * AIR_QUALITY_HISTORY_INSTANCE_SHARE; - if ((usage[w.name] ?? 0) + weight > ceiling) return false; - } - return ( - (usage["account-day"] ?? 0) + weight <= - AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS - ); -} - -/** - * Where a day without a row is placed: the period that covers it, else the - * home from `homeSince` on, or on every day when no `homeSince` is stored - * (see the header). Null for a day before the home with no period. Pure; - * exported for tests. - */ -export function resolveHistoryLocation( - day: string, - home: HomeLocation | null, - travels: readonly TravelOverride[], -): ResolvedLocation | null { - for (const t of travels) { - if (day >= t.startDate && day <= t.endDate) { - return { lat: t.lat, lon: t.lon, label: t.label, source: "TRAVEL" }; - } - } - if (home && (home.since === null || day >= home.since)) { - return { lat: home.lat, lon: home.lon, label: home.label, source: "HOME" }; - } - return null; -} - /** The stored progress, as the settings card reads it. */ export interface AirQualityHistoryState { version: 1; @@ -258,6 +226,13 @@ export function readAirQualityHistoryState( * mood entry, a taken dose, a symptom or a workout. Distinct in SQL, so a * dense record costs a scan, not a transfer. `tz` is a zone the account * stored (home or profile), bound as a parameter. + * + * Measured on Postgres 16 with 4 000 days of readings: 4 000 rows answer in + * about 5 ms, 400 000 rows (100 a day) in about 75 ms, almost all of it the + * measurement scan and its hash aggregate (`EXPLAIN ANALYZE`: a parallel + * sequential scan feeding a HashAggregate, 6.4 MB per worker). The cost + * grows with the readings, not with the days, so it is read once per chain + * and handed on ({@link EntryDaysSnapshot}), not once per run. */ export async function readEntryDays( userId: string, @@ -301,21 +276,53 @@ export async function readEntryDays( .sort(); } +/** + * The days with entries as one run read them, handed to the next run of the + * same chain so the entry tables are scanned once per chain. Valid only for + * the zone and the cutoff it was read under; a run under another one (the + * home moved, or the chain crossed midnight) reads afresh. + */ +export interface EntryDaysSnapshot { + timezone: string; + cutoff: string; + days: string[]; +} + +/** Tolerant read of a snapshot from a job payload; null for anything else. */ +export function readEntryDaysSnapshot(raw: unknown): EntryDaysSnapshot | null { + if (raw === null || typeof raw !== "object") return null; + const r = raw as Record; + if ( + typeof r.timezone !== "string" || + typeof r.cutoff !== "string" || + !Array.isArray(r.days) || + !r.days.every((d) => typeof d === "string") + ) { + return null; + } + return { timezone: r.timezone, cutoff: r.cutoff, days: r.days as string[] }; +} + /** What the history still has to do for an account, and how far it is. */ export interface HistoryWork { pending: HistoryDay[]; total: number; done: number; timezone: string; + /** The entry days this collection used, for the chain's next run. */ + entryDays: EntryDaysSnapshot; } /** * Collect the pending days and the counts. Null when the account is gone or - * its air quality is off (for the account or the instance). + * its air quality is off (for the account or the instance). With `entryDays` + * from the chain's previous run under the same zone and cutoff, the entry + * tables are not read again. */ export async function collectAirQualityHistory( userId: string, now: Date = new Date(), + entryDays: EntryDaysSnapshot | null = null, ): Promise { const user = await prisma.user.findUnique({ where: { id: userId }, @@ -329,11 +336,23 @@ export async function collectAirQualityHistory( const timezone = home?.timezone ?? user.timezone; const cutoff = shiftDateKey(utcDayKey(now), -AIR_QUALITY_HISTORY_SETTLE_DAYS); if (cutoff < AIR_QUALITY_EARLIEST_DAY) { - return { pending: [], total: 0, done: 0, timezone }; + return { + pending: [], + total: 0, + done: 0, + timezone, + entryDays: { timezone, cutoff, days: [] }, + }; } - const [entryDays, rows] = await Promise.all([ - readEntryDays(userId, timezone, AIR_QUALITY_EARLIEST_DAY, cutoff), + const reuse = + entryDays !== null && + entryDays.timezone === timezone && + entryDays.cutoff === cutoff; + const [days, rows] = await Promise.all([ + reuse + ? entryDays.days + : readEntryDays(userId, timezone, AIR_QUALITY_EARLIEST_DAY, cutoff), prisma.environmentContext.findMany({ where: { userId, date: { gte: AIR_QUALITY_EARLIEST_DAY, lte: cutoff } }, select: { @@ -352,7 +371,7 @@ export async function collectAirQualityHistory( const pending: HistoryDay[] = []; let total = 0; let done = 0; - for (const date of entryDays) { + for (const date of days) { const row = rowByDate.get(date); let location: ResolvedLocation | null; if (row) { @@ -366,7 +385,7 @@ export async function collectAirQualityHistory( } else if (inAnyPeriod(date, travel.unreadable)) { location = null; } else { - location = resolveHistoryLocation(date, home, travel.overrides); + location = resolveLocationForDay(date, home, travel.overrides); } if (!location) continue; // Outside what the source serves at this place: left empty, not counted. @@ -378,31 +397,38 @@ export async function collectAirQualityHistory( pending.push({ date, location, hasRow: row != null }); } } - return { pending, total, done, timezone }; -} - -/** A weather observation with nothing in it, for a day the archive skipped. */ -function emptyWeather(date: string): DailyEnvironmentObservation { return { - date, - tempMin: null, - tempMax: null, - tempMean: null, - apparentMean: null, - apparentMax: null, - sunshineSec: null, - daylightSec: null, - precipSum: null, - pressureMean: null, - pressureDelta: null, - humidityMean: null, - cloudMean: null, - weatherCode: null, + pending, + total, + done, + timezone, + entryDays: { timezone, cutoff, days }, }; } +/** + * Whether the history may still send for the account: the account exists, + * its environment module is on, and its air quality is on for the account + * and the instance. Read before every range, so switching any of them off + * stops a run between two ranges, not only at the next run. + */ +async function historyStillActive(userId: string): Promise { + const user = await prisma.user.findUnique({ + where: { id: userId }, + select: { environmentAirQualityEnabled: true }, + }); + if (!user || !isAirQualityActive(user.environmentAirQualityEnabled)) { + return false; + } + return isModuleEnabled(userId, "environment"); +} + export interface AirQualityHistoryRunResult { - /** `inactive`: the account is gone or its air quality is off. */ + /** + * `inactive`: the account is gone, or its module or air quality is off + * (at the start or between two ranges). `error`: the feed refused a + * request, which ends the run and its chain. + */ status: "inactive" | "complete" | "progress" | "budget" | "error"; /** Days whose air quality this run stored. */ filled: number; @@ -412,20 +438,30 @@ export interface AirQualityHistoryRunResult { done: number; /** Days still pending after this run. */ remaining: number; + /** The entry days this run used, for its follow-up; null when inactive. */ + entryDays: EntryDaysSnapshot | null; } /** * One run of the history for an account: up to `maxRanges` ranges, newest * first, inside the history's budget ceilings. Stores the progress and says - * why it stopped. + * why it stopped. `entryDays` is the previous run's snapshot in a chain. */ export async function runAirQualityHistory( userId: string, - opts: { now?: Date; maxRanges?: number } = {}, + opts: { + now?: Date; + maxRanges?: number; + entryDays?: EntryDaysSnapshot | null; + } = {}, ): Promise { const now = opts.now ?? new Date(); const maxRanges = opts.maxRanges ?? AIR_QUALITY_HISTORY_RANGES_PER_RUN; - const work = await collectAirQualityHistory(userId, now); + const work = await collectAirQualityHistory( + userId, + now, + opts.entryDays ?? null, + ); if (!work) { return { status: "inactive", @@ -434,6 +470,7 @@ export async function runAirQualityHistory( total: 0, done: 0, remaining: 0, + entryDays: null, }; } @@ -461,46 +498,23 @@ export async function runAirQualityHistory( ) .sort((a, b) => b.range.endDate.localeCompare(a.range.endDate)); + const budget = { accountId: userId, ceiling: AIR_QUALITY_HISTORY_CEILING }; let filled = 0; let fetches = 0; - let stop: "budget" | "error" | null = null; - let errors = 0; + let stop: "budget" | "error" | "inactive" | null = null; let attempted = 0; for (const { group, range } of planned) { if (attempted >= maxRanges) break; + if (!(await historyStillActive(userId))) { + stop = "inactive"; + break; + } attempted += 1; const days = range.days.map((d) => group.byDate.get(d)!); const missing = days.filter((d) => !d.hasRow).map((d) => d.date); - const usage = await readOpenMeteoBudgetUsage(userId, new Date()); - if (!historyRequestFits(usage, historyRangeWeight(range, missing))) { - stop = "budget"; - break; - } - - const air = await fetchDailyAirQuality( - { - lat: group.lat, - lon: group.lon, - timezone: work.timezone, - startDate: range.startDate, - endDate: range.endDate, - }, - { accountId: userId }, - ); - if (air.stopped === "budget") { - stop = "budget"; - break; - } - if (air.stopped === "error") { - errors += 1; - continue; - } - fetches += 1; - const airByDay = new Map( - air.days.map((d) => [d.date, d]), - ); - const airFor = (date: string) => airByDay.get(date) ?? uncoverableDay(date); + // The weather first, so a refused or failed archive request costs no + // air-quality call. let weather: Map | null = null; if (missing.length > 0) { try { @@ -512,55 +526,88 @@ export async function runAirQualityHistory( startDate: missing[0], endDate: missing[missing.length - 1], }, - { accountId: userId }, + budget, ); fetches += 1; weather = new Map(observations.map((o) => [o.date, o])); } catch (err) { - if (err instanceof OpenMeteoBudgetExhaustedError) { - stop = "budget"; - } else { - errors += 1; - } - // The days with a row are stored below; those without one stay - // pending until the weather can be fetched with them. + stop = + err instanceof OpenMeteoBudgetExhaustedError ? "budget" : "error"; + if (stop === "error") fetches += 1; + break; } } + const air = await fetchDailyAirQuality( + { + lat: group.lat, + lon: group.lon, + timezone: work.timezone, + startDate: range.startDate, + endDate: range.endDate, + }, + budget, + ); + if (air.stopped !== "budget") fetches += 1; + const airByDay = new Map( + air.days.map((d) => [d.date, d]), + ); + // A day the feed did not answer after a full fetch is one it cannot + // cover; after a stopped fetch it is simply not fetched yet. + const airFor = (date: string): DailyAirQualityObservation | null => + airByDay.get(date) ?? + (air.stopped === null ? uncoverableDay(date) : null); + const fetchedAt = new Date(); const ops: Prisma.PrismaPromise[] = []; + const fillsAir: boolean[] = []; for (const day of days) { + const dayAir = airFor(day.date); if (day.hasRow) { + if (!dayAir) continue; ops.push( prisma.environmentContext.updateMany({ // Still unfetched, so a weather rewrite that landed meanwhile // (with its own air quality) is not overwritten. where: { userId, date: day.date, aqFetchedAt: null }, - data: airQualityColumns(airFor(day.date), fetchedAt), + data: airQualityColumns(dayAir, fetchedAt), }), ); - } else if (weather) { - ops.push( - buildUpsert( - userId, - day.location, - sealLocation(day.location), - weather.get(day.date) ?? emptyWeather(day.date), - { active: true, day: airFor(day.date), now: fetchedAt }, - ), - ); + fillsAir.push(true); + continue; } + // A day the archive answer left out gets no row: it stays pending + // instead of carrying empty weather nothing would fetch again. + const observation = weather?.get(day.date); + if (!observation) continue; + // Without its air quality (the air request stopped), the weather row + // is still stored: it was paid for, and the next run then asks for + // the air quality alone. + ops.push( + buildUpsert( + userId, + day.location, + sealLocation(day.location), + observation, + { active: true, day: dayAir, now: fetchedAt }, + ), + ); + fillsAir.push(dayAir !== null); } if (ops.length > 0) { const results = await prisma.$transaction(ops); - for (const r of results) { + results.forEach((r, i) => { + if (!fillsAir[i]) return; filled += r && typeof r === "object" && "count" in r ? Number((r as { count: number }).count) : 1; - } + }); + } + if (air.stopped) { + stop = air.stopped; + break; } - if (stop) break; } const done = Math.min(work.total, work.done + filled); @@ -578,12 +625,14 @@ export async function runAirQualityHistory( }); const status: AirQualityHistoryRunResult["status"] = - remaining === 0 - ? "complete" - : stop === "budget" - ? "budget" - : errors > 0 && filled === 0 - ? "error" - : "progress"; - return { status, filled, fetches, total: work.total, done, remaining }; + remaining === 0 ? "complete" : (stop ?? "progress"); + return { + status, + filled, + fetches, + total: work.total, + done, + remaining, + entryDays: work.entryDays, + }; } diff --git a/src/lib/environment/open-meteo-air-quality.ts b/src/lib/environment/open-meteo-air-quality.ts index 85d44b373..d782b3f20 100644 --- a/src/lib/environment/open-meteo-air-quality.ts +++ b/src/lib/environment/open-meteo-air-quality.ts @@ -72,6 +72,7 @@ import { enumerateDayCount } from "@/lib/environment/day-span"; import { openMeteoCallWeight, reserveOpenMeteoCalls, + type BudgetCeiling, } from "@/lib/environment/request-budget"; /** Hosted default; override with `OPENMETEO_AIR_QUALITY_URL` (self-host). */ @@ -395,7 +396,7 @@ export async function fetchDailyAirQuality( startDate: string; endDate: string; }, - budget: { accountId?: string } = {}, + budget: { accountId?: string; ceiling?: BudgetCeiling } = {}, ): Promise { const days: DailyAirQualityObservation[] = []; // Days before the feed's first day at this location are answered here, @@ -415,7 +416,9 @@ export async function fetchDailyAirQuality( for (const chunk of chunks) { const variables = variablesForChunk(chunk.endDate); const weight = airQualityChunkWeight(chunk.startDate, chunk.endDate); - if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { + if ( + !(await reserveOpenMeteoCalls(weight, budget.accountId, budget.ceiling)) + ) { return { days, stopped: "budget" }; } try { diff --git a/src/lib/environment/open-meteo.ts b/src/lib/environment/open-meteo.ts index ed3245e14..a0f8e7e43 100644 --- a/src/lib/environment/open-meteo.ts +++ b/src/lib/environment/open-meteo.ts @@ -26,6 +26,7 @@ import { OpenMeteoBudgetExhaustedError, openMeteoCallWeight, reserveOpenMeteoCalls, + type BudgetCeiling, } from "@/lib/environment/request-budget"; /** Hosted archive default; override with `OPENMETEO_BASE_URL` (self-host). */ @@ -278,10 +279,12 @@ export async function fetchDailyEnvironment( startDate: string; endDate: string; }, - budget: { accountId?: string } = {}, + budget: { accountId?: string; ceiling?: BudgetCeiling } = {}, ): Promise { const weight = archiveRequestWeight(args.startDate, args.endDate); - if (!(await reserveOpenMeteoCalls(weight, budget.accountId))) { + if ( + !(await reserveOpenMeteoCalls(weight, budget.accountId, budget.ceiling)) + ) { 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 05f6d46fa..220f8d335 100644 --- a/src/lib/environment/request-budget.ts +++ b/src/lib/environment/request-budget.ts @@ -114,16 +114,54 @@ export function admitWeight( return { admitted: true, next }; } +/** + * A lower ceiling a caller holds itself to inside the budget: a share of each + * instance window, and a number of calls inside the account's daily share. + * The air-quality history backfill uses one so the nightly fetch and + * on-demand requests keep their room. The ceiling is checked against + * everything the windows hold, not only against the caller's own calls, and + * the request is charged to the windows as usual. + */ +export interface BudgetCeiling { + /** Share of each instance window the request may fill up to (0..1). */ + instanceShare: number; + /** Calls the account's daily window may hold after the request. */ + accountDayCalls: number; +} + +/** + * The windows with their limits lowered to `ceiling`. Only the limits + * change, so the buckets written are the same as without a ceiling. Pure; + * exported for tests. + */ +export function windowsUnderCeiling( + windows: readonly BudgetWindow[], + ceiling: BudgetCeiling | undefined, +): readonly BudgetWindow[] { + if (!ceiling) return windows; + return windows.map((w) => ({ + ...w, + limit: + w.name === "account-day" + ? Math.min(w.limit, ceiling.accountDayCalls) + : w.limit * ceiling.instanceShare, + })); +} + /** * 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. 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. + * With `ceiling`, the request is admitted only under that lower ceiling, and + * the check runs under the same lock as the charge, so two processes cannot + * both read room that only one of them gets. */ export async function reserveOpenMeteoCalls( weight: number, accountId?: string, + ceiling?: BudgetCeiling, ): Promise { const weightCenti = Math.max(1, Math.ceil(weight * CENTI)); const windows: readonly BudgetWindow[] = accountId @@ -150,7 +188,12 @@ export async function reserveOpenMeteoCalls( const name = nameOfKey.get(row.key); if (name) buckets.set(name, { count: row.count, resetAt: row.resetAt }); } - const decision = admitWeight(buckets, weightCenti, new Date(), windows); + const decision = admitWeight( + buckets, + weightCenti, + new Date(), + windowsUnderCeiling(windows, ceiling), + ); if (!decision.admitted) return false; for (const [name, bucket] of decision.next) { const key = keyOf(name); @@ -171,39 +214,3 @@ export class OpenMeteoBudgetExhaustedError extends Error { this.name = "OpenMeteoBudgetExhaustedError"; } } - -/** Calls already charged to each live window, by window name. */ -export type BudgetUsage = Partial>; - -/** - * Read how much of each window is in use, in calls, without charging - * anything. A window never written or already reset reads as absent (zero). - * With `accountId` the account's daily share is read as well. Used by the - * air-quality history backfill, which holds itself under a ceiling of its own - * so the nightly fetch and on-demand requests keep their room. - */ -export async function readOpenMeteoBudgetUsage( - accountId?: string, - now: Date = new Date(), -): Promise { - const keys = new Map( - OPEN_METEO_BUDGET_WINDOWS.map((w) => [ - `${BUCKET_PREFIX}:${w.name}`, - w.name, - ]), - ); - if (accountId) { - keys.set(`${BUCKET_PREFIX}:account:${accountId}:day`, "account-day"); - } - const rows = await prisma.rateLimit.findMany({ - where: { key: { in: [...keys.keys()] } }, - select: { key: true, count: true, resetAt: true }, - }); - const usage: BudgetUsage = {}; - for (const row of rows) { - const name = keys.get(row.key); - if (!name || row.resetAt.getTime() <= now.getTime()) continue; - usage[name] = row.count / CENTI; - } - return usage; -} diff --git a/src/lib/jobs/environment-air-quality-history.ts b/src/lib/jobs/environment-air-quality-history.ts index f63423b19..ad6acea4a 100644 --- a/src/lib/jobs/environment-air-quality-history.ts +++ b/src/lib/jobs/environment-air-quality-history.ts @@ -16,8 +16,16 @@ * ceiling stopped it, it sends its own follow-up half a minute later. A run * the budget stopped sends none: the next nightly discovery resumes it, which * is what spreads a long history over several days. The module gate and the - * air-quality switches are re-read on every run, so switching either off - * ends the chain. + * air-quality switches are re-read before every range, so switching either + * off ends the run and the chain. + * + * A run the feed refused a request in (429, 5xx, network) ends with status + * `error`: it reports a failed job and sends no follow-up, and the sends + * carry `retryLimit: 0`, so pg-boss does not retry it straight into the + * same refusal either. The next nightly discovery tries again. + * + * The follow-up carries the days with entries the run read + * (`entryDays`), so a chain scans the entry tables once. * * The queue name MUST be registered in `allQueues` in * `src/lib/jobs/reminder/register-maintenance.ts`, or the sends silently @@ -37,7 +45,9 @@ import { import { isAirQualityOperatorDisabled } from "@/lib/environment/open-meteo-air-quality"; import { readAirQualityHistoryState, + readEntryDaysSnapshot, runAirQualityHistory, + type EntryDaysSnapshot, } from "@/lib/environment/air-quality-history"; import { workerLog } from "./reminder/shared"; @@ -59,8 +69,16 @@ export interface EnvironmentAqHistoryPayload { /** Absent on the discovery tick. */ userId?: string; continuation?: number; + /** The previous run's entry days, on a follow-up only. */ + entryDays?: EntryDaysSnapshot; } +/** + * No pg-boss retry: a run fails when the feed refused it, and an immediate + * retry would only ask the same feed again. The next discovery is the retry. + */ +const AQ_HISTORY_SEND_OPTIONS = { retryLimit: 0 } as const; + function singletonKey(userId: string): string { return `environment-aq-history:${userId}`; } @@ -80,6 +98,7 @@ export async function enqueueAirQualityHistory( ENVIRONMENT_AQ_HISTORY_QUEUE, { userId } satisfies EnvironmentAqHistoryPayload, { + ...AQ_HISTORY_SEND_OPTIONS, singletonKey: singletonKey(userId), ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), }, @@ -149,6 +168,7 @@ export async function discoverAirQualityHistory( ENVIRONMENT_AQ_HISTORY_QUEUE, { userId: candidate.id } satisfies EnvironmentAqHistoryPayload, { + ...AQ_HISTORY_SEND_OPTIONS, singletonKey: singletonKey(candidate.id), ...(startAfterSeconds > 0 ? { startAfter: startAfterSeconds } : {}), }, @@ -203,7 +223,9 @@ export async function handleEnvironmentAqHistory( return jobDone({ skipped: "environment_module_disabled" }); } - const result = await runAirQualityHistory(userId); + const result = await runAirQualityHistory(userId, { + entryDays: readEntryDaysSnapshot(payload.entryDays), + }); const continuation = payload.continuation ?? 0; let continued = false; if ( @@ -216,8 +238,10 @@ export async function handleEnvironmentAqHistory( { userId, continuation: continuation + 1, + ...(result.entryDays ? { entryDays: result.entryDays } : {}), } satisfies EnvironmentAqHistoryPayload, { + ...AQ_HISTORY_SEND_OPTIONS, singletonKey: singletonKey(userId), startAfter: AQ_HISTORY_CONTINUATION_DELAY_SECONDS, }, @@ -228,13 +252,18 @@ export async function handleEnvironmentAqHistory( "info", `[environment-aq-history] user=${userId} status=${result.status} filled=${result.filled} done=${result.done}/${result.total} fetches=${result.fetches}${continued ? " continued" : ""}`, ); - // Every range failed at the feed: say so, so the failing-jobs card sees a + // The feed refused a request: say so, so the failing-jobs card sees a // feed that keeps refusing. A budget stop is not a failure: nothing was // sent past the ceiling, and the next discovery resumes where it stopped. if (result.status === "error") { - return jobFailed("air-quality history: every range failed", undefined, { - fetches: result.fetches, - }); + return jobFailed( + "air-quality history: the feed refused a request", + undefined, + { + fetches: result.fetches, + days_stored: result.filled, + }, + ); } return jobDone({ outcome: result.status, diff --git a/tests/integration/environment-air-quality-history.test.ts b/tests/integration/environment-air-quality-history.test.ts index b71bf55ae..249ecd696 100644 --- a/tests/integration/environment-air-quality-history.test.ts +++ b/tests/integration/environment-air-quality-history.test.ts @@ -21,8 +21,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS, + AIR_QUALITY_HISTORY_CEILING, runAirQualityHistory, } from "@/lib/environment/air-quality-history"; +import { reserveOpenMeteoCalls } from "@/lib/environment/request-budget"; import { openLocation, sealLocation } from "@/lib/environment/location-cipher"; import { handleEnvironmentAqHistory } from "@/lib/jobs/environment-air-quality-history"; import { getPrismaClient, truncateAllTables } from "./setup"; @@ -49,8 +51,19 @@ function daysBetween(start: string, end: string): string[] { return out; } +interface StubOptions { + /** HTTP status the archive answers with instead of data. */ + archiveStatus?: number; + /** HTTP status the air-quality API answers with instead of data. */ + airStatus?: number; + /** Days the archive leaves out of its answer. */ + omitArchiveDays?: readonly string[]; + /** Runs before every air-quality answer. */ + onAir?: () => Promise; +} + /** Answers both feeds with a full day of values for every requested day. */ -function openMeteoStub() { +function openMeteoStub(opts: StubOptions = {}) { return vi.fn(async (input: unknown) => { const url = new URL(String(input)); const days = daysBetween( @@ -58,6 +71,8 @@ function openMeteoStub() { url.searchParams.get("end_date")!, ); if (url.pathname === "/v1/air-quality") { + await opts.onAir?.(); + if (opts.airStatus) return new Response("", { status: opts.airStatus }); const variables = url.searchParams.get("hourly")!.split(","); const time = days.flatMap((d) => Array.from( @@ -69,11 +84,15 @@ function openMeteoStub() { for (const v of variables) hourly[v] = time.map(() => 12); return Response.json({ hourly }); } - const daily: Record = { time: days }; - for (const field of url.searchParams.get("daily")!.split(",")) { - daily[field] = days.map(() => 5); + if (opts.archiveStatus) { + return new Response("", { status: opts.archiveStatus }); } - const time = days.flatMap((d) => + const answered = days.filter((d) => !opts.omitArchiveDays?.includes(d)); + const daily: Record = { time: answered }; + for (const field of url.searchParams.get("daily")!.split(",")) { + daily[field] = answered.map(() => 5); + } + const time = answered.flatMap((d) => Array.from( { length: 24 }, (_, h) => `${d}T${String(h).padStart(2, "0")}:00`, @@ -194,6 +213,22 @@ function requests() { return fetchSpy.mock.calls.map(([input]) => new URL(String(input))); } +function useStub(opts: StubOptions) { + fetchSpy = openMeteoStub(opts); + vi.stubGlobal("fetch", fetchSpy); +} + +/** `[path, start_date]` of every request sent, in order. */ +function sentRanges() { + return requests().map((u) => [u.pathname, u.searchParams.get("start_date")]); +} + +async function rowOn(day: string) { + return getPrismaClient().environmentContext.findUnique({ + where: { userId_date: { userId: OWNER, date: day } }, + }); +} + describe("air-quality history backfill", () => { it("fills every past day with entries the source reaches, once", async () => { const prisma = getPrismaClient(); @@ -366,3 +401,189 @@ describe("air-quality history backfill", () => { expect(sends).toEqual([]); }); }); + +describe("air-quality history: placing, failing and stopping", () => { + it("places no day at a home stored without an effective date, as the nightly fetch does", async () => { + await getPrismaClient().user.update({ + where: { id: OWNER }, + data: { homeSince: null }, + }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + // The two 2020 trip days, the 2024 trip day, the day with a row and the + // day already done; the home day without a row is not placed anywhere. + expect(result).toMatchObject({ status: "complete", total: 5, done: 5 }); + expect(await rowOn("2025-09-10")).toBeNull(); + expect(sentRanges().map(([, start]) => start)).not.toContain("2025-09-10"); + }); + + it("stops at the first refused archive request and sends no air-quality request for it", async () => { + useStub({ archiveStatus: 503 }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result).toMatchObject({ status: "error", filled: 0, fetches: 1 }); + expect(sentRanges()).toEqual([["/v1/archive", "2025-09-10"]]); + }); + + it("stops at the first refused air-quality request and keeps the weather it already paid for", async () => { + useStub({ airStatus: 429 }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result).toMatchObject({ status: "error", filled: 0, fetches: 2 }); + expect(sentRanges()).toEqual([ + ["/v1/archive", "2025-09-10"], + ["/v1/air-quality", "2025-09-10"], + ]); + // The weather row is stored, without air quality. + expect(await rowOn("2025-09-10")).toMatchObject({ + tempMax: 5, + pm25Mean: null, + aqFetchedAt: null, + }); + + // The next run asks for that day's air quality only. + useStub({}); + const next = await runAirQualityHistory(OWNER, { now: NOW }); + expect(next.status).toBe("complete"); + expect(sentRanges().filter(([, start]) => start === "2025-09-10")).toEqual([ + ["/v1/air-quality", "2025-09-10"], + ]); + expect(await rowOn("2025-09-10")).toMatchObject({ + tempMax: 5, + pm25Mean: 12, + }); + }); + + it("sends no follow-up and no retry after a refused request", async () => { + useStub({ archiveStatus: 503 }); + const sends: Array<{ data: unknown; options: unknown }> = []; + const boss = { + send: vi.fn(async (_name: string, data: unknown, options: unknown) => { + sends.push({ data, options }); + return "job-id"; + }), + }; + const outcome = await handleEnvironmentAqHistory(boss as never, { + userId: OWNER, + }); + expect(outcome.ok).toBe(false); + expect(sends).toEqual([]); + }); + + it("creates no row for a day the archive answer leaves out", async () => { + useStub({ omitArchiveDays: ["2020-03-05"] }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result).toMatchObject({ status: "progress", remaining: 1 }); + expect(await rowOn("2020-03-05")).toBeNull(); + expect(await rowOn("2020-03-01")).toMatchObject({ pm25Mean: 12 }); + }); + + it.each([ + [ + "the account's air quality", + () => + getPrismaClient() + .user.update({ + where: { id: OWNER }, + data: { environmentAirQualityEnabled: false }, + }) + .then(() => undefined), + ], + [ + "the environment module", + () => + getPrismaClient() + .user.update({ + where: { id: OWNER }, + data: { modulePreferencesJson: { environment: false } }, + }) + .then(() => undefined), + ], + [ + "the operator switch", + async () => { + vi.stubEnv("ENVIRONMENT_AIR_QUALITY_DISABLED", "true"); + }, + ], + ])("stops between two ranges once %s is switched off", async (_, off) => { + let first = true; + useStub({ + onAir: async () => { + if (first) await off(); + first = false; + }, + }); + const result = await runAirQualityHistory(OWNER, { now: NOW }); + expect(result.status).toBe("inactive"); + expect( + requests().filter((u) => u.pathname === "/v1/air-quality"), + ).toHaveLength(1); + }); + + it("holds the history's ceiling under concurrent reservations", async () => { + const prisma = getPrismaClient(); + const key = `open-meteo-budget:account:${OWNER}:day`; + await prisma.rateLimit.create({ + data: { + key, + count: (AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS - 1) * 100, + resetAt: new Date(Date.now() + 3_600_000), + }, + }); + const admitted = await Promise.all( + Array.from({ length: 6 }, () => + reserveOpenMeteoCalls(0.5, OWNER, AIR_QUALITY_HISTORY_CEILING), + ), + ); + expect(admitted.filter(Boolean)).toHaveLength(2); + const bucket = await prisma.rateLimit.findUniqueOrThrow({ where: { key } }); + expect(bucket.count).toBe(AIR_QUALITY_HISTORY_ACCOUNT_DAY_CALLS * 100); + }); + + it("reads the days with entries once per chain and hands them to the follow-up", async () => { + // Nine more home days, each more than two weeks apart: more ranges than + // one run works through, so the run sends a follow-up. + for (let i = 0; i < 9; i++) { + await measurementOn( + new Date(Date.parse("2025-01-10T00:00:00Z") + i * 20 * 86_400_000) + .toISOString() + .slice(0, 10), + ); + } + const sends: Array<{ data: Record; options: unknown }> = + []; + const boss = { + send: vi.fn( + async ( + _name: string, + data: Record, + options: unknown, + ) => { + sends.push({ data, options }); + return "job-id"; + }, + ), + }; + const outcome = await handleEnvironmentAqHistory(boss as never, { + userId: OWNER, + }); + expect(outcome.ok).toBe(true); + expect(sends).toHaveLength(1); + const followUp = sends[0]!.data as { + userId: string; + continuation: number; + entryDays: { timezone: string; cutoff: string; days: string[] }; + }; + expect(followUp.entryDays.timezone).toBe("Europe/Berlin"); + // A failed run is not retried straight into the same refusal. + expect(sends[0]!.options).toMatchObject({ retryLimit: 0 }); + expect(followUp.entryDays.days).toContain("2025-01-10"); + + // An entry added after the chain read its days is left to the next + // chain: the follow-up does not scan the entry tables again. + await measurementOn("2025-11-20"); + fetchSpy.mockClear(); + await handleEnvironmentAqHistory(boss as never, followUp); + expect(await rowOn("2025-11-20")).toBeNull(); + // A fresh chain reads it. + const fresh = await runAirQualityHistory(OWNER); + expect(fresh.entryDays!.days).toContain("2025-11-20"); + }); +});