Merge branch 'fix/v1436-gh' into release/v1.43.6

This commit is contained in:
Marc-André Bombeck committed 2026-10-19 20:20:28 +02:00
commit d833d39d7d
9 files changed
+602 -105

No files matched your search

+1 -1
View File
@@ -2,7 +2,7 @@ name: Integration tests (long)
# The integration tests that take their size from the environment run small
# on every pull request and at full size here, once a night. Today that is
# the Google Health multi-year backfill: 45 and 60 days of heart rate on a
# the Google Health multi-year backfill: 120 and 160 days of heart rate on a
# pull request, three and four years (some 3.7 million readings, over twenty
# minutes) here. See the header of
# tests/integration/google-health-backfill-multi-year.test.ts.
+22 -8
View File
@@ -187,14 +187,23 @@ year of history fills depends on the device. Two bounds apply instead:
- **Progress.** A walk has to reach older readings as it goes. If Google
hands out 20 pages in a row that reach no further back than the ones before
them (the same page again, or a page token that goes round in a circle),
the walk stops, and that data type's history ends at the oldest reading
written so far. Retrying would only meet the same wall, so it is not
retried; a reconnect starts the backfill over.
the walk stops where it got to. A passing fault in Google's paging looks
the same as the end of what it holds, so the first time this happens the
data type only waits: it keeps its place, the rest of the backfill goes
on, and a day later it is walked again from there. If that second attempt
also stops without getting anywhere, the data type's history ends at the
oldest reading written so far. If it moves on, it is an ordinary walk
again. A walk that stops before it has written a single reading notes no
end at all (there is no date it could be said to end at); the slice fails
as stuck and is retried like any other. A reconnect starts the backfill
over.
Either way the data type counts as finished, the backfill completes once
every other type is in, and the integration card stops saying that older
history is loading. Instead it says, per data type, how far back the import
reaches and why it ends there.
Either way the data type counts as finished once its end is final, the
backfill completes once every other type is in, and the integration card
stops saying that older history is loading. Instead it says, per data type,
how far back the import reaches and why it ends there. While a type waits
for its second attempt the backfill is not complete, and the card still
says older history is loading.
Before 1.43.6 every page-by-page data type had a ceiling of pages counted
across all slices (10,000 for heart rate, 1,000 for the others). A dense
@@ -249,7 +258,12 @@ reading:
- `edges`: where every unfinished page-by-page data type stands now.
- `ended`: data types whose history ended short of its beginning, each
with its measurement `type`, the instant it ends at (`at`) and the
`reason` (`horizon` or `no_progress`).
`reason` (`horizon` or `no_progress`). A `no_progress` entry also
carries `attempts` and `endedAt`; with one attempt it is waiting for
its second, a day after `endedAt`, and is not yet published.
- `action: google_health.backfill.retry_scheduled` marks a slice that found
nothing left but a data type waiting for its second attempt and queued the
next slice for when it is due (`delaySeconds`).
- `meta["googleHealth.backfill.ended"]` appears on the slice where a data
type ended short, with the same entries. A walk stopped for lack of
progress also leaves a warning:
@@ -29,10 +29,12 @@ import {
googleHealthBackfillHorizonYears,
googleHealthWalkProgress,
noteGoogleHealthWalkPage,
type GoogleHealthBackfillEnd,
} from "../backfill-bounds";
import {
googleHealthBackfillEndedList,
googleHealthBackfillProgressAfterCompletion,
googleHealthBackfillRetryPending,
readGoogleHealthBackfillEnded,
recordGoogleHealthBackfillEnd,
runGoogleHealthCollection,
@@ -253,69 +255,202 @@ describe("a collection ended short", () => {
prismaMock.googleHealthConnection.update.mockResolvedValue({});
});
const HOUR = 3_600_000;
const END = {
type: "PULSE" as const,
at: "2024-05-11T00:00:00.000Z",
reason: "no_progress" as const,
};
const HORIZON_END = {
type: "WEIGHT" as const,
at: "2006-10-17T12:00:00.000Z",
reason: "horizon" as const,
};
/** An end without progress `attempts` deep, the last `hoursAgo` back. */
const parked = (attempts: number, hoursAgo: number) => ({
...END,
attempts,
endedAt: new Date(Date.now() - hoursAgo * HOUR).toISOString(),
});
it("counts as finished even though its walk ended truncated, and is noted with where and why", async () => {
const done = new Set<string>();
const { ended, collections } = await runWithGoogleHealthClientOutcome(() =>
runWithGoogleHealthBackfillResume("u1", done, () =>
runGoogleHealthCollection("fetchHeartRate", async () => {
// The walk stopped for no progress: truncated, and the end noted.
noteGoogleHealthDeferred();
await recordGoogleHealthBackfillEnd("fetchHeartRate", END);
return 0;
}),
/** One slice walking heart rate with `walk`, from `ended`. */
async function slice(
walk: () => Promise<number>,
ended = new Map<string, GoogleHealthBackfillEnd>(),
done = new Set<string>(),
) {
return runWithGoogleHealthClientOutcome(() =>
runWithGoogleHealthBackfillResume(
"u1",
done,
() => runGoogleHealthCollection("fetchHeartRate", walk),
new Map(),
new Set(),
ended,
),
).then((r) => r.result);
}
/** A walk stopped for no progress: truncated, and the end noted. */
const stalls = vi.fn(async () => {
noteGoogleHealthDeferred();
await recordGoogleHealthBackfillEnd("fetchHeartRate", END);
return 0;
});
it("parks a first end without progress: not finished, edge kept, not published", async () => {
const done = new Set<string>();
const { ended, collections } = await slice(stalls, new Map(), done);
expect(collections).toEqual([
expect.objectContaining({ key: "fetchHeartRate", status: "done" }),
expect.objectContaining({ key: "fetchHeartRate", status: "stopped" }),
]);
expect([...ended]).toEqual([["fetchHeartRate", END]]);
expect(done.has("fetchHeartRate")).toBe(false);
const note = ended.get("fetchHeartRate")!;
expect(note).toMatchObject({ ...END, attempts: 1 });
expect(Date.now() - Date.parse(note.endedAt!)).toBeLessThan(60_000);
const written = prismaMock.googleHealthConnection.update.mock.calls.at(
-1,
)![0].data.backfillProgress as unknown;
expect(
readGoogleHealthBackfillEnded(written).get("fetchHeartRate"),
).toEqual(END);
).toEqual(note);
// Waiting for another try is not where the history ends.
expect(googleHealthBackfillEndedList(written)).toEqual([]);
expect(googleHealthBackfillProgressAfterCompletion(ended)).toBeNull();
});
it("is skipped by the next slice instead of walked into the same wall", async () => {
it("is not walked again before its day is up", async () => {
const walk = vi.fn(async () => 1);
const { skipped } = await runWithGoogleHealthBackfillResume(
"u1",
new Set(),
() => runGoogleHealthCollection("fetchHeartRate", walk),
new Map(),
new Set(),
new Map([["fetchHeartRate", END]]),
const { skipped } = await slice(
walk,
new Map<string, GoogleHealthBackfillEnd>([
["fetchHeartRate", parked(1, 23)],
]),
);
expect(walk).not.toHaveBeenCalled();
expect(skipped).toEqual(["fetchHeartRate"]);
});
it("outlives the completion stamp, and nothing else does", () => {
it("is walked again once its day is up, and a second end without progress is final", async () => {
const done = new Set<string>();
const { ended, collections } = await slice(
stalls,
new Map<string, GoogleHealthBackfillEnd>([
["fetchHeartRate", parked(1, 25)],
]),
done,
);
expect(stalls).toHaveBeenCalledTimes(1);
expect(collections).toEqual([
expect.objectContaining({ key: "fetchHeartRate", status: "done" }),
]);
expect(done.has("fetchHeartRate")).toBe(true);
expect(ended.get("fetchHeartRate")).toMatchObject({ ...END, attempts: 2 });
expect(
googleHealthBackfillProgressAfterCompletion(ended)!.ended.fetchHeartRate,
).toMatchObject({ ...END, attempts: 2 });
});
it("goes back to an ordinary walk when the second attempt gets somewhere", async () => {
const { ended, collections } = await slice(
async () => {
// Moved on, then ran out of slice: no end this time.
noteGoogleHealthDeferred();
return 5;
},
new Map<string, GoogleHealthBackfillEnd>([
["fetchHeartRate", parked(1, 25)],
]),
);
expect(collections).toEqual([
expect.objectContaining({ key: "fetchHeartRate", status: "stopped" }),
]);
expect(ended.has("fetchHeartRate")).toBe(false);
});
it("is skipped for good once final, and so is the horizon", async () => {
const walk = vi.fn(async () => 1);
const { skipped } = await slice(
walk,
new Map<string, GoogleHealthBackfillEnd>([
["fetchHeartRate", parked(2, 100)],
]),
);
expect(walk).not.toHaveBeenCalled();
expect(skipped).toEqual(["fetchHeartRate"]);
const atHorizon = await slice(
walk,
new Map<string, GoogleHealthBackfillEnd>([
["fetchHeartRate", { ...HORIZON_END, type: "PULSE" as const }],
]),
);
expect(walk).not.toHaveBeenCalled();
expect(atHorizon.skipped).toEqual(["fetchHeartRate"]);
});
it("finishes the collection at once at the horizon", async () => {
const done = new Set<string>();
const { collections } = await slice(
async () => {
await recordGoogleHealthBackfillEnd("fetchHeartRate", {
...HORIZON_END,
type: "PULSE",
});
return 3;
},
new Map(),
done,
);
expect(collections).toEqual([
expect.objectContaining({ key: "fetchHeartRate", status: "done" }),
]);
expect(done.has("fetchHeartRate")).toBe(true);
});
it("outlives the completion stamp when final, and nothing else does", () => {
expect(googleHealthBackfillProgressAfterCompletion(new Map())).toBeNull();
expect(
googleHealthBackfillProgressAfterCompletion(
new Map([["fetchHeartRate", END]]),
new Map<string, GoogleHealthBackfillEnd>([
["fetchWeight", HORIZON_END],
["fetchHeartRate", parked(1, 1)],
]),
),
).toEqual({ ended: { fetchHeartRate: END } });
).toEqual({ ended: { fetchWeight: HORIZON_END } });
});
it("is published per type, and a malformed note is dropped", () => {
it("names the earliest retry among the parked ones", () => {
const early = parked(1, 20);
expect(
googleHealthBackfillRetryPending(
new Map<string, GoogleHealthBackfillEnd>([
["fetchWeight", HORIZON_END],
["fetchHeartRate", early],
["fetchHrv", parked(1, 2)],
["fetchSpo2", parked(2, 30)],
]),
),
).toEqual({
keys: ["fetchHeartRate", "fetchHrv"],
dueAt: Date.parse(early.endedAt) + 24 * HOUR,
});
expect(
googleHealthBackfillRetryPending(
new Map<string, GoogleHealthBackfillEnd>([
["fetchWeight", HORIZON_END],
]),
),
).toBeNull();
});
it("is published per type when final, and a malformed note is dropped", () => {
expect(
googleHealthBackfillEndedList({
ended: {
fetchHeartRate: END,
fetchWeight: {
type: "WEIGHT",
at: "2006-10-17T12:00:00.000Z",
reason: "horizon",
},
fetchHeartRate: parked(2, 30),
fetchRhr: parked(1, 1),
fetchWeight: HORIZON_END,
fetchBodyFat: { type: "BODY_FAT", at: "nope", reason: "horizon" },
fetchSpo2: { type: "NOT_A_TYPE", at: END.at, reason: "horizon" },
fetchHrv: { type: "PULSE", at: END.at, reason: "because" },
+46
View File
@@ -140,4 +140,50 @@ export interface GoogleHealthBackfillEnd {
type: MeasurementType;
at: string;
reason: GoogleHealthBackfillEndReason;
/**
* `no_progress` only: how many separate attempts have ended this way, and
* when the last one did (see `googleHealthBackfillEndIsFinal`).
*/
attempts?: number;
endedAt?: string;
}
/**
* Separate attempts that must end without progress before a data type's
* history counts as ending there. One is not enough: Google handing out the
* same page for a while is as likely a passing fault in its paging as the end
* of what it holds, and a reconnect would be the only way back.
*/
export const GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS = 2;
/** How long after an attempt ended without progress the next one starts. */
export const GOOGLE_HEALTH_NO_PROGRESS_RETRY_MS = 24 * 60 * 60 * 1000;
/**
* Whether an end is where the history stays: always at the horizon, and
* after `GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS` attempts without progress.
* Until then the data type waits for its next attempt, unfinished, and is not
* published as ending anywhere.
*/
export function googleHealthBackfillEndIsFinal(
end: GoogleHealthBackfillEnd,
): boolean {
return (
end.reason === "horizon" ||
(end.attempts ?? 1) >= GOOGLE_HEALTH_NO_PROGRESS_ATTEMPTS
);
}
/**
* When an end that is not final may be walked again (epoch ms), or null for
* a final one. An end without a readable time is due at once.
*/
export function googleHealthBackfillEndRetryAt(
end: GoogleHealthBackfillEnd,
): number | null {
if (googleHealthBackfillEndIsFinal(end)) return null;
const endedAt = end.endedAt ? Date.parse(end.endedAt) : Number.NaN;
return Number.isNaN(endedAt)
? 0
: endedAt + GOOGLE_HEALTH_NO_PROGRESS_RETRY_MS;
}
+110 -19
View File
@@ -22,7 +22,11 @@ import { MeasurementType as MeasurementTypes } from "@/generated/prisma/enums";
import { encrypt, decrypt } from "@/lib/crypto";
import { annotate, getEvent } from "@/lib/logging/context";
import type { GoogleHealthCollectionKey } from "./collection-keys";
import type { GoogleHealthBackfillEnd } from "./backfill-bounds";
import {
googleHealthBackfillEndIsFinal,
googleHealthBackfillEndRetryAt,
type GoogleHealthBackfillEnd,
} from "./backfill-bounds";
import { userDayKey } from "@/lib/tz/format";
import { isP2002 } from "@/lib/prisma-errors";
import {
@@ -491,6 +495,11 @@ interface BackfillResumeTracker {
edgeFilterUnsupported: Set<string>;
/** Collections whose walk ended short of the history's start, and where. */
ended: Map<string, GoogleHealthBackfillEnd>;
/**
* Collections walked again this attempt after an earlier one ended without
* progress, with how many attempts had ended that way.
*/
retrying: Map<string, number>;
}
const backfillResumeStorage = new AsyncLocalStorage<BackfillResumeTracker>();
@@ -567,10 +576,22 @@ export function readGoogleHealthBackfillEnded(
if (typeof e.type !== "string" || !(e.type in MeasurementTypes)) {
continue;
}
const attempts =
typeof e.attempts === "number" &&
Number.isInteger(e.attempts) &&
e.attempts >= 1
? e.attempts
: undefined;
const endedAt =
typeof e.endedAt === "string" && !Number.isNaN(Date.parse(e.endedAt))
? new Date(e.endedAt).toISOString()
: undefined;
out.set(key, {
type: e.type as MeasurementType,
at: new Date(at).toISOString(),
reason: e.reason,
...(attempts !== undefined ? { attempts } : {}),
...(endedAt !== undefined ? { endedAt } : {}),
});
}
return out;
@@ -587,11 +608,16 @@ export interface GoogleHealthBackfillEndedEntry {
reason: GoogleHealthBackfillEnd["reason"];
}
/** The `backfillProgress` column's ended collections, as published. */
/**
* The `backfillProgress` column's ended collections, as published: only the
* final ends. One still waiting for another attempt is not where its history
* ends yet.
*/
export function googleHealthBackfillEndedList(
value: unknown,
): GoogleHealthBackfillEndedEntry[] {
return [...readGoogleHealthBackfillEnded(value).values()]
.filter(googleHealthBackfillEndIsFinal)
.map((e) => ({ type: e.type, endsAt: e.at, reason: e.reason }))
.sort((a, b) => a.type.localeCompare(b.type));
}
@@ -603,23 +629,60 @@ export function googleHealthBackfillEndedList(
export function googleHealthBackfillProgressAfterCompletion(
ended: Map<string, GoogleHealthBackfillEnd>,
): {
ended: Record<string, { type: string; at: string; reason: string }>;
ended: Record<string, GoogleHealthBackfillEndJson>;
} | null {
return ended.size > 0 ? { ended: endedJson(ended) } : null;
const final = new Map(
[...ended].filter(([, e]) => googleHealthBackfillEndIsFinal(e)),
);
return final.size > 0 ? { ended: endedJson(final) } : null;
}
/** One ended collection as the column stores it. */
type GoogleHealthBackfillEndJson = {
type: string;
at: string;
reason: string;
attempts?: number;
endedAt?: string;
};
/** The ended collections as the column stores them. */
function endedJson(
ended: Map<string, GoogleHealthBackfillEnd>,
): Record<string, { type: string; at: string; reason: string }> {
): Record<string, GoogleHealthBackfillEndJson> {
return Object.fromEntries(
[...ended].map(([key, e]) => [
key,
{ type: e.type, at: e.at, reason: e.reason },
{
type: e.type,
at: e.at,
reason: e.reason,
...(e.attempts !== undefined ? { attempts: e.attempts } : {}),
...(e.endedAt !== undefined ? { endedAt: e.endedAt } : {}),
},
]),
);
}
/**
* The collections waiting for another attempt after one ended without
* progress, and the earliest instant (epoch ms) one of them is due, or null
* when none waits.
*/
export function googleHealthBackfillRetryPending(
ended: Map<string, GoogleHealthBackfillEnd>,
): { keys: string[]; dueAt: number } | null {
const keys: string[] = [];
let dueAt = Infinity;
for (const [key, e] of ended) {
const at = googleHealthBackfillEndRetryAt(e);
if (at === null) continue;
keys.push(key);
dueAt = Math.min(dueAt, at);
}
return keys.length > 0 ? { keys, dueAt } : null;
}
/**
* Read the `backfillProgress` column into the collections Google refused the
* edge filter on.
@@ -728,6 +791,7 @@ export async function runWithGoogleHealthBackfillResume<T>(
collections: [],
edgeFilterUnsupported,
ended,
retrying: new Map(),
};
const result = await backfillResumeStorage.run(tracker, fn);
return {
@@ -779,20 +843,32 @@ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): {
/**
* Record that a collection's backfill ended short of the start of its
* history, at `at`, because of `reason` (see `backfill-bounds.ts`). The
* collection then counts as finished for this backfill
* (`runGoogleHealthCollection`), and the note outlives the completion stamp
* so the card can say where the history ends. Outside a backfill this does
* nothing.
* history, at `at`, because of `reason` (see `backfill-bounds.ts`). A final
* end (`googleHealthBackfillEndIsFinal`) counts the collection as finished
* for this backfill (`runGoogleHealthCollection`) and outlives the
* completion stamp, so the card can say where the history ends. A first end
* without progress parks the collection with its edge until its next
* attempt is due. Outside a backfill this does nothing.
*/
export async function recordGoogleHealthBackfillEnd(
key: GoogleHealthCollectionKey,
end: GoogleHealthBackfillEnd,
end: Pick<GoogleHealthBackfillEnd, "type" | "at" | "reason">,
): Promise<void> {
const resume = backfillResumeStorage.getStore();
if (!resume) return;
const ended = new Map(resume.ended);
ended.set(key, end);
ended.set(
key,
end.reason === "no_progress"
? {
...end,
// Counted across attempts: the one walked again this attempt
// carries the count it was parked with.
attempts: (resume.retrying.get(key) ?? 0) + 1,
endedAt: new Date().toISOString(),
}
: end,
);
annotate({
meta: { "googleHealth.backfill.ended": Object.fromEntries(ended) },
});
@@ -974,7 +1050,13 @@ export async function runGoogleHealthCollection(
): Promise<number> {
const resume = backfillResumeStorage.getStore();
if (!resume) return fn();
if (resume.done.has(key) || resume.ended.has(key)) {
const parked = resume.ended.get(key);
if (
resume.done.has(key) ||
(parked &&
(googleHealthBackfillEndIsFinal(parked) ||
googleHealthBackfillEndRetryAt(parked)! > Date.now()))
) {
resume.skipped.push(key);
return 0;
}
@@ -988,6 +1070,13 @@ export async function runGoogleHealthCollection(
});
return 0;
}
if (parked) {
// Its earlier attempt ended without progress and its wait is over: walk
// it again from its edge. The note goes with the next write; the count
// stays with the attempt, so a second end without progress is final.
resume.retrying.set(key, parked.attempts ?? 1);
resume.ended.delete(key);
}
const startedAt = performance.now();
const pagesBefore = googleHealthSlicePagesRead();
const edgeFrom = resume.cursors.get(key)?.at ?? null;
@@ -999,11 +1088,13 @@ export async function runGoogleHealthCollection(
(hardFailStorage.getStore()?.failures.length ?? 0) !== hardBefore ||
(softSkipStorage.getStore()?.count ?? 0) !== softBefore;
const truncated = truncatedBefore || googleHealthOutcomeTruncated();
// A walk that ended short on purpose (the horizon, or no progress) has
// read all it is going to: finished, though the walk may report itself
// truncated.
const endedShort = resume.ended.has(key);
const clean = !failedHard && (endedShort || !truncated);
// A walk that ended short for good (the horizon, or no progress on its
// last attempt) has read all it is going to: finished, though the walk may
// report itself truncated. One that ended without progress for the first
// time is not: it keeps its edge and waits for its next attempt.
const end = resume.ended.get(key);
const endedFinal = end !== undefined && googleHealthBackfillEndIsFinal(end);
const clean = !failedHard && (endedFinal || (!truncated && !end));
let finished = false;
if (clean) {
// The collection counts as finished only once the note says so. The job
+8 -3
View File
@@ -327,15 +327,20 @@ async function syncMetricCollection(
? {
resumable: true,
// Google keeps handing out pages that reach no further back:
// the history ends where the walk got to, and the backfill
// goes on without it instead of meeting the same wall again.
// the walk is noted as ending where it got to. The first time
// it waits a day for another attempt; a second such end is
// where the history stays (`backfill-bounds.ts`).
onNoProgress: async () => {
// A page whose write failed fails the collection instead.
if (edgeHeld) return;
// Nothing written yet: there is no place the history could
// be said to end at, and "now" would be false. The walk
// stays unfinished and fails the slice like any stuck one.
const at = googleHealthBackfillCursor(resource.verb)?.edge?.at;
if (!at) return;
await recordGoogleHealthBackfillEnd(resource.verb, {
type: resource.type,
at: (at ?? new Date()).toISOString(),
at: at.toISOString(),
reason: "no_progress",
});
},
@@ -37,8 +37,12 @@ const {
/** Collections whose resume edge the mocked walk moves. */
advances: [] as string[],
collections: [] as Array<{ key: string; status: string }>,
/** Collections the mocked walk ends short (horizon / no progress). */
/** Collections the mocked walk ends short for good. */
ends: [] as string[],
/** Collections the mocked walk parks after a first end without progress. */
parks: [] as string[],
/** When the parked ones are due again (epoch ms). */
retryDueAt: 0,
},
}));
@@ -67,6 +71,14 @@ vi.mock("@/lib/google-health/sync-core", () => ({
readGoogleHealthBackfillEnded: () => new Map(),
googleHealthBackfillProgressAfterCompletion: (ended: Map<string, unknown>) =>
ended.size > 0 ? { ended: Object.fromEntries(ended) } : null,
googleHealthBackfillRetryPending: (
ended: Map<string, { reason: string; attempts?: number }>,
) => {
const keys = [...ended]
.filter(([, e]) => e.reason === "no_progress" && (e.attempts ?? 1) < 2)
.map(([key]) => key);
return keys.length > 0 ? { keys, dueAt: resumeState.retryDueAt } : null;
},
runWithGoogleHealthBackfillResume: async (
_userId: string,
done: Set<string>,
@@ -78,10 +90,21 @@ vi.mock("@/lib/google-health/sync-core", () => ({
const result = await fn();
for (const key of resumeState.finishes) done.add(key);
for (const key of resumeState.ends) {
// Final: finished the way `runGoogleHealthCollection` finishes it.
ended.set(key, {
type: "PULSE",
at: "2024-05-11T00:00:00.000Z",
reason: "no_progress",
attempts: 2,
});
done.add(key);
}
for (const key of resumeState.parks) {
ended.set(key, {
type: "PULSE",
at: "2024-05-11T00:00:00.000Z",
reason: "no_progress",
attempts: 1,
});
}
return {
@@ -104,6 +127,7 @@ import {
runGoogleHealthBackfillForUser,
} from "../google-health-backfill";
import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission";
import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys";
import {
GOOGLE_HEALTH_DATA_TYPES,
forEachDataPointPage,
@@ -121,6 +145,8 @@ beforeEach(() => {
resumeState.advances = [];
resumeState.collections = [];
resumeState.ends = [];
resumeState.parks = [];
resumeState.retryDueAt = 0;
});
describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
@@ -326,11 +352,64 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
type: "PULSE",
at: "2024-05-11T00:00:00.000Z",
reason: "no_progress",
attempts: 2,
},
},
});
});
it("continues a slice that only parked a collection for another try", async () => {
resumeState.parks = ["fetchHeartRate"];
resumeState.retryDueAt = Date.now() + 24 * 3_600_000;
syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: true });
await expect(
runGoogleHealthBackfillForUser("u1", undefined, 4),
).resolves.toEqual({ imported: 0, continued: true });
// Other collections are still to walk: the next slice comes as usual.
expect(bossSend.mock.calls[0]![2]).toMatchObject({
startAfter: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
});
expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
});
it("waits for the retry instead of stamping when only a parked collection is left", async () => {
resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter(
(k) => k !== "fetchHeartRate",
);
resumeState.parks = ["fetchHeartRate"];
resumeState.retryDueAt = Date.now() + 10 * 3_600_000;
// The parked collection was skipped as not yet due: a clean run.
syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false });
await expect(
runGoogleHealthBackfillForUser("u1", undefined, 7),
).resolves.toEqual({ imported: 0, continued: true });
expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
expect(bossSend).toHaveBeenCalledTimes(1);
const startAfter = (bossSend.mock.calls[0]![2] as { startAfter: number })
.startAfter;
expect(Math.abs(startAfter - 10 * 3600)).toBeLessThanOrEqual(2);
expect(bossSend.mock.calls[0]![1]).toMatchObject({
data: { slice: 8 },
});
});
it("does not stamp over a parked collection when the retry cannot be queued", async () => {
resumeState.doneBefore = GOOGLE_HEALTH_COLLECTION_KEYS.filter(
(k) => k !== "fetchHeartRate",
);
resumeState.parks = ["fetchHeartRate"];
resumeState.retryDueAt = Date.now() + 3_600_000;
bossSend.mockResolvedValue(null);
syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: false });
await expect(
runGoogleHealthBackfillForUser("u1", undefined, 7),
).resolves.toEqual({ imported: 0 });
expect(prismaMock.googleHealthConnection.update).not.toHaveBeenCalled();
});
it("continues a slice that only moved a collection's edge", async () => {
resumeState.advances = ["fetchHeartRate"];
syncUserGoogleHealthMock.mockResolvedValue({ imported: 900, failed: true });
+82 -24
View File
@@ -26,8 +26,10 @@ import {
readGoogleHealthBackfillDone,
readGoogleHealthBackfillEdgeFilterUnsupported,
readGoogleHealthBackfillEnded,
googleHealthBackfillRetryPending,
runWithGoogleHealthBackfillResume,
} from "@/lib/google-health/sync-core";
import { googleHealthBackfillEndIsFinal } from "@/lib/google-health/backfill-bounds";
import {
isGoogleHealthParked,
syncUserGoogleHealth,
@@ -204,11 +206,20 @@ export async function runGoogleHealthBackfillForUser(
const endedBefore = readGoogleHealthBackfillEnded(
connection?.backfillProgress,
);
// A collection that ended short (`endedBefore`) has read all it is going
// to, so it counts as finished alongside the ones read to the end.
// A collection that ended short for good has read all it is going to, so
// it counts as finished alongside the ones read to the end. One parked
// after a first end without progress does not: it waits for another try.
const finishedCount = (): number =>
new Set([...done, ...endedBefore.keys()]).size;
new Set([
...done,
...[...endedBefore]
.filter(([, e]) => googleHealthBackfillEndIsFinal(e))
.map(([key]) => key),
]).size;
const doneBefore = finishedCount();
const parkedBefore = new Set(
googleHealthBackfillRetryPending(endedBefore)?.keys ?? [],
);
const budget = googleHealthSliceBudget(
sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS,
@@ -274,28 +285,75 @@ export async function runGoogleHealthBackfillForUser(
);
}
if (failed) {
// Progress is what this slice put on disk: collections finished, and
// edges moved. Either is enough to go on; neither means stuck.
const progressed = finished > 0 || advanced.size > 0;
if (progressed && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) {
const boss = getGlobalBoss();
const sent = boss
? await enqueueIntegrationBackfillAdmission(
boss,
{
kind: "google-health-backfill",
data: {
userId,
enqueuedAt: new Date().toISOString(),
slice: slice + 1,
},
const sendNextSlice = async (delaySeconds: number): Promise<boolean> => {
if (slice + 1 > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) return false;
const boss = getGlobalBoss();
const sent = boss
? await enqueueIntegrationBackfillAdmission(
boss,
{
kind: "google-health-backfill",
data: {
userId,
enqueuedAt: new Date().toISOString(),
slice: slice + 1,
},
budget.stopReason === "rate_limited"
? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS
: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
)
: null;
},
delaySeconds,
)
: null;
return Boolean(sent);
};
// Collections parked after a first end without progress
// (`backfill-bounds.ts`). Once nothing else is left, the next slice waits
// until the earliest of them is due; the marker is never stamped over one.
const pending = googleHealthBackfillRetryPending(ended);
const onlyRetriesLeft =
pending !== null &&
GOOGLE_HEALTH_COLLECTION_KEYS.every(
(key) => done.has(key) || pending.keys.includes(key),
);
if (pending && (onlyRetriesLeft || !failed)) {
const delaySeconds = Math.max(
INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
Math.ceil((pending.dueAt - Date.now()) / 1000),
);
if (await sendNextSlice(delaySeconds)) {
annotate({
action: {
name: "google_health.backfill.retry_scheduled",
details: { slice, collections: pending.keys, delaySeconds },
},
});
return { imported, continued: true };
}
// Not queued (most often one is already waiting under the same slice
// key, sent by an earlier boot's slice). Nothing is lost: the marker is
// not stamped, so the next boot offers the backfill again.
annotate({
action: {
name: "google_health.backfill.retry_not_queued",
details: { slice, collections: pending.keys },
},
});
return { imported };
}
if (failed) {
// Progress is what this slice put on disk: collections finished, edges
// moved, or a collection parked for another attempt. Any is enough to go
// on; none means stuck.
const newlyParked = (pending?.keys ?? []).filter(
(key) => !parkedBefore.has(key),
).length;
const progressed = finished > 0 || advanced.size > 0 || newlyParked > 0;
if (progressed) {
const sent = await sendNextSlice(
budget.stopReason === "rate_limited"
? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS
: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
);
if (sent) {
annotate({
action: {
@@ -17,8 +17,11 @@
* - an account left at the old ceiling resumes with the next slice, from
* where it stopped, and completes;
* - a Google that keeps answering with the same page is stopped after a
* bounded number of requests, and the backfill still completes, with the
* data type noted as ending there because of it;
* bounded number of requests; the data type waits a day for a second
* attempt (a restart before then leaves it alone), and only when that
* one stalls too does the backfill complete with the type noted as
* ending there;
* - a walk that stalls before writing anything notes no end at all;
* - a history older than the horizon is imported up to the horizon, and the
* data type is noted as ending at it;
* - a history handed out oldest first is read whole, not ended at the
@@ -46,6 +49,8 @@ const fake = vi.hoisted(() => ({
perPage: 1000,
/** Answer every heart-rate request with the newest page and a next token. */
loop: false,
/** Answer every heart-rate request with no readings and a next token. */
empty: false,
/** Hand the history out oldest first instead. */
asc: false,
heartRateRequests: 0,
@@ -66,6 +71,9 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
return respond({});
}
fake.heartRateRequests += 1;
if (fake.empty) {
return respond({ dataPoints: [], nextPageToken: "1" });
}
const filter = u.searchParams.get("filter") ?? "";
const upTo = /physical_time <= "([^"]+)"/.exec(filter)?.[1];
const from = /physical_time >= "([^"]+)"/.exec(filter)?.[1];
@@ -109,13 +117,21 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
};
});
/** The next slices the backfill queues, recorded. */
const sent = vi.hoisted(() => ({ slices: [] as number[] }));
/** The next slices the backfill queues, recorded, with their delays. */
const sent = vi.hoisted(() => ({
slices: [] as number[],
delays: [] as number[],
}));
vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({
...(await importOriginal<typeof import("@/lib/jobs/boss-instance")>()),
getGlobalBoss: () => ({
send: async (_queue: string, data: { data?: { slice?: number } }) => {
send: async (
_queue: string,
data: { data?: { slice?: number } },
options?: { startAfter?: number },
) => {
sent.slices.push(data.data?.slice ?? -1);
sent.delays.push(options?.startAfter ?? 0);
return `job-${sent.slices.length}`;
},
}),
@@ -176,9 +192,11 @@ async function seed(opts: {
fake.spacing = opts.spacing;
fake.perPage = opts.perPage;
fake.loop = false;
fake.empty = false;
fake.asc = false;
fake.heartRateRequests = 0;
sent.slices = [];
sent.delays = [];
}
/** Pages a slice reads here: a fifth of production, so slices stay cheap. */
@@ -186,12 +204,13 @@ const SLICE_PAGES = 100;
/**
* Run slices the way the admission drain does, each one sending the next,
* until one ends the chain. Returns how many ran.
* until one ends the chain or sends the next one an hour or more out (a
* collection waiting for another attempt). Returns how many ran.
*/
async function runChain(maxSlices: number): Promise<number> {
async function runChain(maxSlices: number, from = 0): Promise<number> {
const { runGoogleHealthBackfillForUser } =
await import("@/lib/jobs/google-health-backfill");
let slice = 0;
let slice = from;
let ran = 0;
for (; ran < maxSlices; ran++) {
const before = sent.slices.length;
@@ -204,6 +223,7 @@ async function runChain(maxSlices: number): Promise<number> {
if (!outcome.continued) return ran + 1;
expect(sent.slices.length).toBe(before + 1);
slice = sent.slices.at(-1)!;
if (sent.delays.at(-1)! >= 3600) return ran + 1;
}
throw new Error(`backfill did not finish within ${maxSlices} slices`);
}
@@ -311,29 +331,78 @@ describe("Google Health backfill bounded by progress, not pages (#1220)", () =>
expect(fake.heartRateRequests).toBeLessThanOrEqual(10 + 2 * slices);
}, 600_000);
it("stops a walk Google keeps answering with the same page, and completes with the reason noted", async () => {
it("parks a walk Google keeps answering with the same page, tries it again a day later, and only then ends it", async () => {
await seed({ total: 5_000, spacing: 60_000, perPage: 1000 });
fake.loop = true;
await runChain(5);
const { GOOGLE_HEALTH_NO_PROGRESS_PAGES } =
await import("@/lib/google-health/backfill-bounds");
const { googleHealthBackfillEndedList } =
await import("@/lib/google-health/sync-core");
// The first page and the twenty that repeated it, then never again.
expect(fake.heartRateRequests).toBe(1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES);
const c = await connection();
const walk = 1 + GOOGLE_HEALTH_NO_PROGRESS_PAGES;
// First attempt: the first page and the twenty that repeat it. Not
// complete, nothing published, and the next slice a day out.
await runChain(5);
expect(fake.heartRateRequests).toBe(walk);
let c = await connection();
expect(c.backfillCompletedAt).toBeNull();
expect(googleHealthBackfillEndedList(c.backfillProgress)).toEqual([]);
expect(sent.delays.at(-1)).toBeGreaterThan(23 * 3600);
// A slice that comes early (a restart) leaves it alone and waits again.
await runChain(1, sent.slices.at(-1)!);
expect(fake.heartRateRequests).toBe(walk);
expect((await connection()).backfillCompletedAt).toBeNull();
// A day later the walk goes on from its edge, gets one page further,
// meets the same wall, and this time the history ends there.
const prisma = getPrismaClient();
const progress = c.backfillProgress as {
ended: Record<string, { endedAt: string }>;
};
progress.ended.fetchHeartRate!.endedAt = new Date(
Date.now() - 25 * HOUR,
).toISOString();
await prisma.googleHealthConnection.update({
where: { userId: USER },
data: { backfillProgress: progress },
});
await runChain(5, sent.slices.at(-1)!);
expect(fake.heartRateRequests).toBe(2 * walk);
c = await connection();
expect(c.backfillCompletedAt).toBeInstanceOf(Date);
expect(googleHealthBackfillEndedList(c.backfillProgress)).toEqual([
{
type: "PULSE",
// The oldest reading the one real page wrote.
endsAt: new Date(NOW - 999 * 60_000).toISOString(),
// The oldest reading the second attempt's one real page wrote.
endsAt: new Date(NOW - 1998 * 60_000).toISOString(),
reason: "no_progress",
},
]);
expect(await pulseRows()).toBe(1000);
expect(await pulseRows()).toBe(1999);
}, 600_000);
it("notes no end for a walk that wrote nothing before it stalled", async () => {
// Pages without a single reading, forever: there is no place to say the
// history ends at, and "today" would be false. The slice fails as stuck.
await seed({ total: 5_000, spacing: 60_000, perPage: 1000 });
fake.empty = true;
const { runGoogleHealthBackfillForUser } =
await import("@/lib/jobs/google-health-backfill");
const { GOOGLE_HEALTH_NO_PROGRESS_PAGES } =
await import("@/lib/google-health/backfill-bounds");
await expect(
runGoogleHealthBackfillForUser(USER, () => false, 0, {
slicePages: SLICE_PAGES,
}),
).rejects.toThrow(/no progress in slice 0/);
expect(fake.heartRateRequests).toBe(GOOGLE_HEALTH_NO_PROGRESS_PAGES);
const c = await connection();
expect(c.backfillCompletedAt).toBeNull();
expect(c.backfillProgress).not.toHaveProperty("ended");
expect(await pulseRows()).toBe(0);
}, 600_000);
it("reads back to the horizon and notes where the history ends", async () => {