diff --git a/prisma/migrations/0413_strava_backfill_resume/migration.sql b/prisma/migrations/0413_strava_backfill_resume/migration.sql new file mode 100644 index 000000000..821446acc --- /dev/null +++ b/prisma/migrations/0413_strava_backfill_resume/migration.sql @@ -0,0 +1,12 @@ +-- Where an unfinished Strava backfill picks up. +-- +-- One backfill run reads at most forty pages of a hundred activities. A +-- longer history ended the run as incomplete without a note of how far it +-- got, so every boot read the same newest forty pages again, up to 190 of +-- the 200 requests the whole instance may make in fifteen minutes, and the +-- older history was never reached. The column holds the start of the oldest +-- activity the last run read; the next run asks for activities before it. +-- Nullable and unset for every existing row: an unfinished backfill starts +-- once more from the newest page and records its place from then on. + +ALTER TABLE "users" ADD COLUMN "strava_backfill_resume_before" TIMESTAMP(3); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 11fa3b2f5..05b4527d1 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -316,6 +316,11 @@ model User { // First-connect full-history backfill marker (self-converging discovery // predicate `strava_backfill_completed_at IS NULL`), mirroring WHOOP/Fitbit. stravaBackfillCompletedAt DateTime? @map("strava_backfill_completed_at") + // Where an unfinished backfill picks up: the start of the oldest activity + // the last run read. A history longer than one run's page ceiling is read + // in runs that each walk further back from here, instead of every run + // starting again at the newest page. Cleared when the backfill completes. + stravaBackfillResumeBefore DateTime? @map("strava_backfill_resume_before") // Locale (null = use system/admin default) locale String? @map("locale") // "de" | "en" | null diff --git a/src/lib/data-wipe/wipe-plan.ts b/src/lib/data-wipe/wipe-plan.ts index f1867dbce..447f8f180 100644 --- a/src/lib/data-wipe/wipe-plan.ts +++ b/src/lib/data-wipe/wipe-plan.ts @@ -469,6 +469,7 @@ export const USER_RESET = { stravaAthleteId: null, stravaLastActivityAt: null, stravaBackfillCompletedAt: null, + stravaBackfillResumeBefore: null, healthKitConfigJson: Prisma.DbNull, healthKitLastSyncedAt: null, healthKitLastSyncTrigger: null, diff --git a/src/lib/export/backup-plan.ts b/src/lib/export/backup-plan.ts index 25a1246e1..91e5dc74c 100644 --- a/src/lib/export/backup-plan.ts +++ b/src/lib/export/backup-plan.ts @@ -960,6 +960,7 @@ export const USER_COLUMN_BACKUP_CLASS = { coachLastSeenAt: "OPERATIONAL", stravaLastActivityAt: "OPERATIONAL", stravaBackfillCompletedAt: "OPERATIONAL", + stravaBackfillResumeBefore: "OPERATIONAL", healthKitLastSyncedAt: "OPERATIONAL", healthKitLastSyncTrigger: "OPERATIONAL", healthKitLastBackgroundSyncAt: "OPERATIONAL", diff --git a/src/lib/jobs/integration-backfill-drain.ts b/src/lib/jobs/integration-backfill-drain.ts index df6ad556e..15ed48fdf 100644 --- a/src/lib/jobs/integration-backfill-drain.ts +++ b/src/lib/jobs/integration-backfill-drain.ts @@ -132,11 +132,15 @@ async function runAdmitted( return; } case "strava-backfill": { - const { imported } = await runStravaBackfillForUser(userId); + const { imported, continued } = await runStravaBackfillForUser( + userId, + data.slice ?? 0, + ); totals.imported += imported; + if (continued) totals.continued += 1; workerLog( "info", - `[strava-backfill] user=${userId} imported=${imported}`, + `[strava-backfill] user=${userId} imported=${imported}${continued ? " continued" : ""}`, ); return; } diff --git a/src/lib/jobs/strava-backfill.ts b/src/lib/jobs/strava-backfill.ts index 539396dad..5c36fc905 100644 --- a/src/lib/jobs/strava-backfill.ts +++ b/src/lib/jobs/strava-backfill.ts @@ -17,13 +17,16 @@ import { prisma } from "@/lib/db"; import { annotate } from "@/lib/logging/context"; import { getGlobalBoss } from "@/lib/jobs/boss-instance"; -import { integrationBackfillSourceOptions } from "@/lib/jobs/integration-backfill-admission"; +import { + enqueueIntegrationBackfillAdmission, + integrationBackfillSourceOptions, +} from "@/lib/jobs/integration-backfill-admission"; import { isReauthRequired } from "@/lib/integrations/status"; import { backfillVerdict, withPageWalkLedger, } from "@/lib/integrations/page-walk"; -import { syncUserStrava } from "@/lib/strava/sync"; +import { syncStravaHistory } from "@/lib/strava/sync"; export const STRAVA_BACKFILL_QUEUE = "strava-backfill"; @@ -40,23 +43,38 @@ export interface StravaBackfillPayload { } /** - * Per-user backfill handler. Runs a full-history sync for one account and - * stamps `stravaBackfillCompletedAt` only when the walk read the whole - * history, so the discovery query drops it. Idempotent: the - * `(userId, source, externalId)` workout upsert is key-stable, so a re-run + * The pause before the next run of an unfinished backfill: Strava's + * short-term rate window. One run spends up to forty page reads and 150 + * detail reads of the 200 the app may make in fifteen minutes, so the next + * run waits for a fresh window rather than opening on a 429 or starving the + * hourly syncs. + */ +export const STRAVA_BACKFILL_CONTINUATION_DELAY_SECONDS = 15 * 60; + +/** + * Per-user backfill handler. Runs one stretch of the full-history walk for + * one account and stamps `stravaBackfillCompletedAt` only when the walk read + * to the start of the history, so the discovery query drops it. Idempotent: + * the `(userId, source, externalId)` workout upsert is key-stable, so a re-run * (e.g. a reboot mid-walk) overwrites rather than duplicating. * * Verdict-gated (`backfillVerdict`, `integrations/page-walk.ts`), like the - * WHOOP and Fitbit backfills. A failed sync throws out of `syncUserStrava` - * (failure recorded first) and pg-boss retries. An activity walk that reached - * its page ceiling with history left is not complete: the marker stays unset - * without a retry, and boot discovery offers the account again. A connection - * parked for re-authorisation returns without stamping and without a retry, - * since every attempt would meet the same dead grant. + * WHOOP and Fitbit backfills. A failed sync throws out of the walk (failure + * recorded first) and pg-boss retries. A walk that reached its page ceiling + * with history left is not complete: it records where it got to + * (`stravaBackfillResumeBefore`, the start of the oldest activity it read) + * and sends the next run for after the rate window, which asks for the + * activities before that point. Without the note every run began again at + * the newest page, read the same forty pages and never reached the rest. A + * run that got no further than the last one records nothing and sends + * nothing, and the next boot offers the account again. A connection parked + * for re-authorisation returns without stamping and without a retry, since + * every attempt would meet the same dead grant. */ export async function runStravaBackfillForUser( userId: string, -): Promise<{ imported: number }> { + slice: number = 0, +): Promise<{ imported: number; continued: boolean }> { if (await isReauthRequired(userId, "strava")) { annotate({ action: { @@ -64,27 +82,56 @@ export async function runStravaBackfillForUser( details: { imported: 0 }, }, }); - return { imported: 0 }; + return { imported: 0, continued: false }; } - const { result: imported, truncated } = await withPageWalkLedger(() => - syncUserStrava(userId, { fullSync: true }), + const stored = await prisma.user.findUnique({ + where: { id: userId }, + select: { stravaBackfillResumeBefore: true }, + }); + const before = stored?.stravaBackfillResumeBefore ?? undefined; + const { result, truncated } = await withPageWalkLedger(() => + syncStravaHistory(userId, before), ); + const { imported, oldestStart } = result; // A failure has already thrown; what is left to decide is whether the walk // reached the beginning of the history. if (backfillVerdict({ failed: false, truncated }) === "incomplete") { + // One second past the oldest start, so an activity that shares it is + // read again rather than skipped; the upsert makes the overlap free. + const resumeBefore = oldestStart + ? new Date(oldestStart.getTime() + 1000) + : null; + const progressed = + resumeBefore !== null && + (before === undefined || resumeBefore.getTime() < before.getTime()); + if (progressed) { + await prisma.user.update({ + where: { id: userId }, + data: { stravaBackfillResumeBefore: resumeBefore }, + }); + } + const continued = progressed && (await sendNextRun(userId, slice)); annotate({ action: { name: "strava.backfill.truncated", - details: { imported, collections: truncated.join(",") }, + details: { + imported, + collections: truncated.join(","), + progressed, + continued, + }, }, }); - return { imported }; + return { imported, continued }; } await prisma.user.update({ where: { id: userId }, - data: { stravaBackfillCompletedAt: new Date() }, + data: { + stravaBackfillCompletedAt: new Date(), + stravaBackfillResumeBefore: null, + }, }); annotate({ @@ -93,7 +140,21 @@ export async function runStravaBackfillForUser( details: { imported }, }, }); - return { imported }; + return { imported, continued: false }; +} + +async function sendNextRun(userId: string, slice: number): Promise { + const boss = getGlobalBoss(); + if (!boss) return false; + const sent = await enqueueIntegrationBackfillAdmission( + boss, + { + kind: "strava-backfill", + data: { userId, enqueuedAt: new Date().toISOString(), slice: slice + 1 }, + }, + STRAVA_BACKFILL_CONTINUATION_DELAY_SECONDS, + ); + return Boolean(sent); } /** diff --git a/src/lib/strava/__tests__/sync-cursor-durability.test.ts b/src/lib/strava/__tests__/sync-cursor-durability.test.ts index 636e529ba..b4b9a9f84 100644 --- a/src/lib/strava/__tests__/sync-cursor-durability.test.ts +++ b/src/lib/strava/__tests__/sync-cursor-durability.test.ts @@ -182,6 +182,7 @@ describe("syncUserStrava cursor durability", () => { it("advances the backfill watermark after every workout is persisted", async () => { await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ imported: 3, + continued: false, }); expect(workouts.size).toBe(3); diff --git a/src/lib/strava/__tests__/sync-page-walk.test.ts b/src/lib/strava/__tests__/sync-page-walk.test.ts index 3f2f09f0e..395046631 100644 --- a/src/lib/strava/__tests__/sync-page-walk.test.ts +++ b/src/lib/strava/__tests__/sync-page-walk.test.ts @@ -11,13 +11,15 @@ const mocks = vi.hoisted(() => ({ fetchActivities: vi.fn(), annotate: vi.fn(), updateUser: vi.fn(), + findUser: vi.fn(), isReauthRequired: vi.fn(), + send: vi.fn(), })); vi.mock("@/lib/db", () => ({ prisma: { user: { - findUnique: vi.fn(async () => ({ stravaLastActivityAt: null })), + findUnique: mocks.findUser, updateMany: vi.fn(async () => ({ count: 1 })), update: mocks.updateUser, }, @@ -62,7 +64,7 @@ vi.mock("../client", async (importOriginal) => ({ fetchActivityById: vi.fn(async () => ({})), })); vi.mock("@/lib/jobs/boss-instance", () => ({ - getGlobalBoss: vi.fn(() => null), + getGlobalBoss: vi.fn(() => ({ send: mocks.send })), })); import { runStravaBackfillForUser } from "@/lib/jobs/strava-backfill"; @@ -71,27 +73,36 @@ import { syncUserStrava } from "../sync"; const PER_PAGE = 100; const BACKFILL_MAX_PAGES = 40; -/** A page of `n` activities; ids continue across pages. */ -function page(pageNo: number, n: number) { - return Array.from({ length: n }, (_, i) => { - const id = (pageNo - 1) * PER_PAGE + i + 1; - return { - id, - sport_type: "Run", - start_date: new Date(Date.UTC(2020, 0, 1) + id * 3_600_000).toISOString(), - moving_time: 1800, - elapsed_time: 1900, - average_heartrate: 140, - }; - }); +const HOUR = 3_600_000; +const T0 = Date.UTC(2020, 0, 1); + +/** Activity `id` of a history; a higher id started later. */ +function activity(id: number) { + return { + id, + sport_type: "Run", + start_date: new Date(T0 + id * HOUR).toISOString(), + moving_time: 1800, + elapsed_time: 1900, + average_heartrate: 140, + }; } -/** A history of `total` activities, served page by page. */ +/** + * A history of `total` activities served as Strava serves a walk without + * `after`: newest first, `before` (epoch seconds) cutting off the newer ones. + */ function history(total: number) { mocks.fetchActivities.mockImplementation( - async (_t: string, q: { page: number; perPage: number }) => { - const remaining = total - (q.page - 1) * q.perPage; - return page(q.page, Math.max(0, Math.min(q.perPage, remaining))); + async ( + _t: string, + q: { page: number; perPage: number; before?: number }, + ) => { + const ids = Array.from({ length: total }, (_, i) => total - i).filter( + (id) => q.before === undefined || T0 + id * HOUR < q.before * 1000, + ); + const from = (q.page - 1) * q.perPage; + return ids.slice(from, from + q.perPage).map(activity); }, ); } @@ -102,10 +113,30 @@ function truncatedAnnotations() { ); } +/** The resume point the user row holds, as the backfill writes it. */ +let resumeBefore: Date | null = null; + beforeEach(() => { vi.clearAllMocks(); + resumeBefore = null; mocks.isReauthRequired.mockResolvedValue(false); - mocks.updateUser.mockResolvedValue({ id: "user-1" }); + mocks.findUser.mockImplementation(async () => ({ + stravaLastActivityAt: null, + stravaBackfillResumeBefore: resumeBefore, + })); + mocks.updateUser.mockImplementation( + async ({ + data, + }: { + data: { stravaBackfillResumeBefore?: Date | null }; + }) => { + if ("stravaBackfillResumeBefore" in data) { + resumeBefore = data.stravaBackfillResumeBefore ?? null; + } + return { id: "user-1" }; + }, + ); + mocks.send.mockResolvedValue("job-next"); }); describe("Strava activity walk", () => { @@ -157,20 +188,42 @@ describe("Strava backfill verdict", () => { await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ imported: 3 * PER_PAGE - 1, + continued: false, }); expect(mocks.updateUser).toHaveBeenCalledWith({ where: { id: "user-1" }, - data: { stravaBackfillCompletedAt: expect.any(Date) }, + data: { + stravaBackfillCompletedAt: expect.any(Date), + stravaBackfillResumeBefore: null, + }, }); + expect(mocks.send).not.toHaveBeenCalled(); }); - it("leaves the marker unset, without a retry, when the walk was cut off", async () => { + it("records where a cut-off walk got to and sends the next run after the rate window", async () => { history((BACKFILL_MAX_PAGES + 1) * PER_PAGE); await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ imported: BACKFILL_MAX_PAGES * PER_PAGE, + continued: true, }); - expect(mocks.updateUser).not.toHaveBeenCalled(); + // The oldest activity read is id 101; the next run asks for those before. + expect(resumeBefore).toEqual(new Date(T0 + 101 * HOUR + 1000)); + expect(mocks.updateUser).not.toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + stravaBackfillCompletedAt: expect.any(Date), + }), + }), + ); + expect(mocks.send).toHaveBeenCalledWith( + "integration-backfill-admission", + expect.objectContaining({ + kind: "strava-backfill", + data: expect.objectContaining({ userId: "user-1", slice: 1 }), + }), + expect.objectContaining({ startAfter: 15 * 60 }), + ); expect(mocks.annotate).toHaveBeenCalledWith( expect.objectContaining({ action: expect.objectContaining({ name: "strava.backfill.truncated" }), @@ -178,12 +231,63 @@ describe("Strava backfill verdict", () => { ); }); + it("goes on from the recorded place instead of the newest page, and finishes", async () => { + const total = (BACKFILL_MAX_PAGES + 10) * PER_PAGE; + history(total); + + await runStravaBackfillForUser("user-1"); + const firstRun = mocks.fetchActivities.mock.calls.length; + mocks.fetchActivities.mockClear(); + await expect(runStravaBackfillForUser("user-1", 1)).resolves.toEqual({ + imported: 10 * PER_PAGE + 1, + continued: false, + }); + + expect(firstRun).toBe(BACKFILL_MAX_PAGES); + // Every page of the second run asked for activities before the place + // the first one recorded (the oldest it read was id 1001), and the walk + // ended at the oldest activity. + const queries = mocks.fetchActivities.mock.calls.map(([, q]) => q); + expect(new Set(queries.map((q) => q.before))).toEqual( + new Set([(T0 + 1001 * HOUR) / 1000 + 1]), + ); + expect(queries.length).toBe(11); + expect(resumeBefore).toBeNull(); + expect(mocks.updateUser).toHaveBeenLastCalledWith({ + where: { id: "user-1" }, + data: { + stravaBackfillCompletedAt: expect.any(Date), + stravaBackfillResumeBefore: null, + }, + }); + }); + + it("records nothing and sends nothing when a run got no further than the last", async () => { + history((BACKFILL_MAX_PAGES + 1) * PER_PAGE); + // A place older than anything the walk can read past: the provider kept + // handing back the same newest pages. + mocks.fetchActivities.mockImplementation(async (_t, q) => + Array.from({ length: PER_PAGE }, (_, i) => + activity(5_000 - (q.page - 1) * PER_PAGE - i), + ), + ); + resumeBefore = new Date(T0); + + await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ + imported: BACKFILL_MAX_PAGES * PER_PAGE, + continued: false, + }); + expect(resumeBefore).toEqual(new Date(T0)); + expect(mocks.send).not.toHaveBeenCalled(); + }); + it("returns without walking or stamping while the connection waits for re-authorisation", async () => { mocks.isReauthRequired.mockResolvedValue(true); history(PER_PAGE); await expect(runStravaBackfillForUser("user-1")).resolves.toEqual({ imported: 0, + continued: false, }); expect(mocks.fetchActivities).not.toHaveBeenCalled(); expect(mocks.updateUser).not.toHaveBeenCalled(); diff --git a/src/lib/strava/client.ts b/src/lib/strava/client.ts index 4428f14e1..a5407f915 100644 --- a/src/lib/strava/client.ts +++ b/src/lib/strava/client.ts @@ -225,6 +225,8 @@ export interface StravaDetailedActivity extends StravaSummaryActivity { export interface FetchActivitiesQuery { /** Epoch SECONDS cursor — return activities started after this instant. */ after?: number; + /** Epoch SECONDS — return activities started before this instant. */ + before?: number; page?: number; perPage?: number; } @@ -269,6 +271,8 @@ export async function fetchActivities( const params = new URLSearchParams(); if (typeof query.after === "number") params.set("after", String(Math.floor(query.after))); + if (typeof query.before === "number") + params.set("before", String(Math.floor(query.before))); params.set("page", String(query.page ?? 1)); params.set("per_page", String(query.perPage ?? 100)); const json = await stravaGet( diff --git a/src/lib/strava/sync.ts b/src/lib/strava/sync.ts index 89070f8fd..7405530d0 100644 --- a/src/lib/strava/sync.ts +++ b/src/lib/strava/sync.ts @@ -102,10 +102,43 @@ export async function syncUserStrava( userId: string, opts: SyncUserStravaOptions = {}, ): Promise { + return (await runStravaSync(userId, opts)).imported; +} + +/** What one history run read, for the backfill to resume from. */ +export interface StravaHistoryRun { + imported: number; + /** + * The start of the oldest activity the run read, when the pages came + * newest first as Strava serves a walk without `after`; null otherwise, + * so a place is never recorded from pages in an order the resume does not + * expect. + */ + oldestStart: Date | null; +} + +/** + * One run of the full-history walk, from `before` (or the newest activity) + * back as far as one run's page ceiling reaches. + */ +export async function syncStravaHistory( + userId: string, + before?: Date, +): Promise { + return runStravaSync(userId, { fullSync: true }, before); +} + +async function runStravaSync( + userId: string, + opts: SyncUserStravaOptions, + before?: Date, +): Promise { const conn = await getStravaConnection(userId); - if (!conn) return 0; + if (!conn) return { imported: 0, oldestStart: null }; const fullSync = opts.fullSync ?? false; + const beforeEpoch = + before === undefined ? undefined : Math.floor(before.getTime() / 1000); // Resolve the incremental cursor. On a full backfill we walk all history // (`after` omitted). Otherwise start at the stored newest-activity instant @@ -167,6 +200,9 @@ export async function syncUserStrava( let rows: StravaWorkoutRow[]; let newestStart: Date | null = null; + let oldestStart: Date | null = null; + let previousStart: number | null = null; + let newestFirst = true; try { // Strava pages by number; a short page is the last one. A walk that @@ -178,7 +214,12 @@ export async function syncUserStrava( async (cursor) => { const page = cursor ?? 1; const activities = await authed((t) => - fetchActivities(t, { after: afterEpoch, page, perPage: PER_PAGE }), + fetchActivities(t, { + after: afterEpoch, + before: beforeEpoch, + page, + perPage: PER_PAGE, + }), ); const mapped: StravaWorkoutRow[] = []; for (const a of activities) { @@ -212,11 +253,13 @@ export async function syncUserStrava( mapped.push(row); if (a.start_date) { const d = new Date(a.start_date); - if ( - !Number.isNaN(d.getTime()) && - (!newestStart || d > newestStart) - ) { - newestStart = d; + if (!Number.isNaN(d.getTime())) { + if (!newestStart || d > newestStart) newestStart = d; + if (!oldestStart || d < oldestStart) oldestStart = d; + if (previousStart !== null && d.getTime() > previousStart) { + newestFirst = false; + } + previousStart = d.getTime(); } } } @@ -270,7 +313,7 @@ export async function syncUserStrava( annotate({ action: { name: "strava.sync.complete", details: { imported } }, }); - return imported; + return { imported, oldestStart: newestFirst ? oldestStart : null }; } /**