mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 13:01:00 +00:00
Merge branch 'v1424/gh' into release/v1.42.4
This commit is contained in:
Marc-André Bombeck committed
commit
b96c114a45
23 files changed
+2046
-168
No files matched your search
@@ -122,22 +122,87 @@ The Google Cloud client id/secret are per-user (Settings), not env vars.
|
||||
|
||||
## Backfill, the hourly sync and Google's rate limit
|
||||
|
||||
Only one sync runs per account at a time. While the history backfill that
|
||||
follows a connect is running, the hourly sync leaves that account alone, and
|
||||
**Sync now** reports that a sync is already in progress and shows its progress
|
||||
instead of starting a second one.
|
||||
Only one sync runs per account at a time. **Sync now** reports that a sync is
|
||||
already in progress while the history backfill that follows a connect is
|
||||
running, and shows its progress instead of starting a second one.
|
||||
|
||||
The backfill reads the account's whole history, and for a watch that records
|
||||
heart rate every minute that is a thousand-odd pages per year. It runs in
|
||||
slices: a slice reads for up to five minutes or 500 pages, writes what it
|
||||
read, notes how far it got and lets go of the account. The next slice starts
|
||||
a minute later where the last one stopped. Two kinds of note are kept: a data
|
||||
type read to the end is not read again, and a data type read page by page
|
||||
(heart rate, weight, body fat, blood glucose, core temperature) keeps the
|
||||
time of the oldest reading it has written, so the next slice asks Google only
|
||||
for what is older. A restart or an update in the middle loses at most the page
|
||||
that was being read.
|
||||
|
||||
The hourly sync waits for the slice that is running (up to eight minutes)
|
||||
instead of skipping the account, so recent data keeps arriving while an old
|
||||
history is still coming in.
|
||||
|
||||
Google limits how many requests one account may make per minute. When a sync
|
||||
reaches that limit, Google answers `429`, and HealthLog waits and repeats the
|
||||
request; the connection is never marked as needing a reconnect because of it.
|
||||
Access tokens last an hour, and a long backfill renews its token as it goes.
|
||||
A backfill that stops part-way (a restart, the job's time budget) picks up
|
||||
with the data types it has not finished yet on its next attempt.
|
||||
reaches that limit Google answers `429`, and HealthLog waits and repeats the
|
||||
request. If Google asks for a longer wait than a minute, the slice ends there
|
||||
and the next one starts five minutes later. A server error (`500`, `502`,
|
||||
`503`, `504`), a request that times out and a dropped connection are repeated
|
||||
three times before they count as a failure. None of these mark the connection
|
||||
as needing a reconnect. Access tokens last an hour, and a long backfill renews
|
||||
its token as it goes.
|
||||
|
||||
A slice that moved nothing forward (no data type finished, no reading older
|
||||
than before) is a real failure: the job fails with the reason, pg-boss retries
|
||||
it a few times, and after that the next restart offers the backfill again.
|
||||
Everything already written stays.
|
||||
|
||||
If an older version left your Google Health card asking for a reconnect after
|
||||
a backfill, the first sync after updating checks the stored grant and resumes
|
||||
on its own when the grant is still valid.
|
||||
|
||||
### Reading the backfill in the logs
|
||||
|
||||
Every slice writes one log line when it ends, with
|
||||
`"background":{"task_name":"job.integration_backfill"}`. The parts worth
|
||||
reading:
|
||||
|
||||
- `meta.kind` is `google-health-backfill`, `meta.slice` the slice number
|
||||
(0 for the first), `meta.job_id` the pg-boss job.
|
||||
- `external_calls` lists the requests the slice made to Google, one entry per
|
||||
request: `method` names the data type and page (`fetchHeartRate(page=212)`),
|
||||
`status` the HTTP status, `duration_ms` how long it took. A request that was
|
||||
repeated carries a `note` (`rate_limited_retrying_in_ms=4000`,
|
||||
`server_error_retrying_in_ms=2000`, `no_response_retrying_in_ms=2000`,
|
||||
`token_rejected_retrying_with_fresh_token`); one that failed for good
|
||||
carries an `error`. Successful requests are listed up to 60 per line and
|
||||
everything up to 200; `external_calls_omitted` counts the rest.
|
||||
- `meta["googleHealth.backfill.slice"]` sums the slice up:
|
||||
- `pages`, `elapsedMs` and `imported`: what the slice read, how long it
|
||||
took, how many readings it wrote.
|
||||
- `stopReason`: why it ended early. `time_budget` and `page_budget` are the
|
||||
normal end of a slice, `rate_limited` means Google would not take more
|
||||
requests for now, `job_budget` that the job itself ran out of time.
|
||||
`null` means the slice read everything it started.
|
||||
- `collectionsDone` of `collectionsTotal`: data types read to the end so
|
||||
far.
|
||||
- `collections`: one entry per data type the slice touched, with its
|
||||
`status` (`done`, `stopped` short on a budget or a rate limit,
|
||||
`deferred` to the next slice, or `failed`), its `pages`, and for the
|
||||
page-by-page types
|
||||
`edgeFrom` and `edgeTo`, the oldest reading written before and after the
|
||||
slice.
|
||||
- `edges`: where every unfinished page-by-page data type stands now.
|
||||
- `action.name` is `google_health.backfill.continued` for a slice that made
|
||||
progress and queued the next one, `google_health.backfill.complete` for the
|
||||
one that finished the history, and `google_health.backfill.stuck` for one
|
||||
that moved nothing. A stuck slice also fails the job with a message that
|
||||
names each unfinished data type and its status, for example
|
||||
`no progress in slice 4, stopped on rate_limited; fetchHeartRate:stopped`.
|
||||
|
||||
While a slice runs, the usual `job.progress` line for the
|
||||
`integration-backfill-admission` queue reports
|
||||
`google_health_backfill_slice`, `google_health_backfill_pages` and
|
||||
`google_health_backfill_collections_done` every two minutes.
|
||||
|
||||
## The 7-day re-consent caveat
|
||||
|
||||
While the OAuth consent screen stays in **Testing** publishing mode, Google
|
||||
|
||||
@@ -4411,8 +4411,10 @@ model GoogleHealthConnection {
|
||||
/// pre-fix externalId (a re-scored night minted parallel copies). Null =
|
||||
/// repair pending (or the connection predates the fix).
|
||||
sleepRepairedAt DateTime? @map("sleep_repaired_at")
|
||||
/// The collections the in-flight history backfill has already walked to
|
||||
/// the end (`{ done: string[] }`), so a retry resumes with the rest
|
||||
/// How far the in-flight history backfill has got
|
||||
/// (`{ done: string[], cursors: { [collection]: { at, order, pages } } }`):
|
||||
/// the collections it walked to the end, and for each collection read page
|
||||
/// by page the edge it has written up to, so the next slice resumes there
|
||||
/// instead of starting over. Null once the backfill completes, and reset
|
||||
/// on every reconnect.
|
||||
backfillProgress Json? @map("backfill_progress")
|
||||
|
||||
@@ -20,9 +20,17 @@ vi.mock("@/lib/logging/context", () => ({
|
||||
}));
|
||||
|
||||
import {
|
||||
readGoogleHealthBackfillCursors,
|
||||
recordGoogleHealthBackfillPage,
|
||||
runGoogleHealthCollection,
|
||||
runWithGoogleHealthBackfillResume,
|
||||
type GoogleHealthBackfillCursor,
|
||||
} from "../sync-core";
|
||||
import {
|
||||
googleHealthSliceBudget,
|
||||
runWithGoogleHealthClientOutcome,
|
||||
runWithGoogleHealthStop,
|
||||
} from "../client";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
@@ -41,7 +49,9 @@ describe("runGoogleHealthCollection under a backfill", () => {
|
||||
expect([...done]).toEqual(["fetchSteps", "sleep"]);
|
||||
expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({
|
||||
where: { userId: "u1" },
|
||||
data: { backfillProgress: { done: ["fetchSteps", "sleep"] } },
|
||||
data: {
|
||||
backfillProgress: { done: ["fetchSteps", "sleep"], cursors: {} },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
@@ -63,3 +73,183 @@ describe("runGoogleHealthCollection under a backfill", () => {
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
/**
|
||||
* #1220 — a collection read page by page keeps its place inside the
|
||||
* collection, not only once it is finished. A heart-rate history of years
|
||||
* that one attempt could not finish used to start over from its newest page
|
||||
* on every attempt.
|
||||
*/
|
||||
describe("the resume edge of a page-by-page collection", () => {
|
||||
const at = (iso: string) => new Date(iso);
|
||||
|
||||
it("moves the edge to the oldest sample of each page and persists it", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
const cursors = new Map<string, GoogleHealthBackfillCursor>();
|
||||
|
||||
const { advanced } = await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(["sleep"]),
|
||||
async () => {
|
||||
await recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at("2026-10-01T12:00:00Z"),
|
||||
last: at("2026-10-01T00:00:00Z"),
|
||||
});
|
||||
await recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at("2026-09-30T23:59:00Z"),
|
||||
last: at("2026-09-30T08:00:00Z"),
|
||||
});
|
||||
},
|
||||
cursors,
|
||||
);
|
||||
|
||||
expect(cursors.get("fetchHeartRate")).toEqual({
|
||||
at: "2026-09-30T08:00:00.000Z",
|
||||
order: "desc",
|
||||
pages: 2,
|
||||
});
|
||||
expect(advanced.get("fetchHeartRate")).toBeNull();
|
||||
const last = prismaMock.googleHealthConnection.update.mock.calls.at(
|
||||
-1,
|
||||
)![0] as { data: { backfillProgress: unknown } };
|
||||
expect(last.data.backfillProgress).toEqual({
|
||||
done: ["sleep"],
|
||||
cursors: {
|
||||
fetchHeartRate: {
|
||||
at: "2026-09-30T08:00:00.000Z",
|
||||
order: "desc",
|
||||
pages: 2,
|
||||
},
|
||||
},
|
||||
});
|
||||
// And it reads back the way it was written.
|
||||
expect(
|
||||
readGoogleHealthBackfillCursors(last.data.backfillProgress).get(
|
||||
"fetchHeartRate",
|
||||
),
|
||||
).toEqual(cursors.get("fetchHeartRate"));
|
||||
});
|
||||
|
||||
it("follows a walk that runs oldest first", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
const cursors = new Map<string, GoogleHealthBackfillCursor>();
|
||||
await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(),
|
||||
() =>
|
||||
recordGoogleHealthBackfillPage("fetchWeight", {
|
||||
first: at("2024-01-01T00:00:00Z"),
|
||||
last: at("2024-03-01T00:00:00Z"),
|
||||
}),
|
||||
cursors,
|
||||
);
|
||||
expect(cursors.get("fetchWeight")).toMatchObject({
|
||||
at: "2024-03-01T00:00:00.000Z",
|
||||
order: "asc",
|
||||
});
|
||||
});
|
||||
|
||||
it("does not count a page that moved nothing as progress", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
const cursors = new Map<string, GoogleHealthBackfillCursor>([
|
||||
[
|
||||
"fetchHeartRate",
|
||||
{ at: "2026-01-01T00:00:00.000Z", order: "desc", pages: 7 },
|
||||
],
|
||||
]);
|
||||
const { advanced } = await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(),
|
||||
() =>
|
||||
// The edge sample itself, read again: no further than before.
|
||||
recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at("2026-01-01T00:00:00Z"),
|
||||
last: at("2026-01-01T00:00:00Z"),
|
||||
}),
|
||||
cursors,
|
||||
);
|
||||
expect(advanced.size).toBe(0);
|
||||
expect(cursors.get("fetchHeartRate")).toMatchObject({
|
||||
at: "2026-01-01T00:00:00.000Z",
|
||||
pages: 8,
|
||||
});
|
||||
});
|
||||
|
||||
it("counts no progress when the edge cannot be written", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockRejectedValue(
|
||||
new Error("connection reset"),
|
||||
);
|
||||
const cursors = new Map<string, GoogleHealthBackfillCursor>();
|
||||
const { advanced } = await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(),
|
||||
() =>
|
||||
recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at("2026-10-01T12:00:00Z"),
|
||||
last: at("2026-10-01T00:00:00Z"),
|
||||
}),
|
||||
cursors,
|
||||
);
|
||||
expect(advanced.size).toBe(0);
|
||||
expect(cursors.size).toBe(0);
|
||||
expect(warnings.some((w) => w.includes("progress write failed"))).toBe(
|
||||
true,
|
||||
);
|
||||
});
|
||||
|
||||
it("clears a finished collection's edge with the same write that marks it done", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
const cursors = new Map<string, GoogleHealthBackfillCursor>([
|
||||
[
|
||||
"fetchHeartRate",
|
||||
{ at: "2026-01-01T00:00:00.000Z", order: "desc", pages: 7 },
|
||||
],
|
||||
]);
|
||||
const done = new Set<string>();
|
||||
await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
done,
|
||||
() => runGoogleHealthCollection("fetchHeartRate", async () => 1),
|
||||
cursors,
|
||||
);
|
||||
expect([...done]).toEqual(["fetchHeartRate"]);
|
||||
expect(cursors.size).toBe(0);
|
||||
expect(prismaMock.googleHealthConnection.update).toHaveBeenCalledWith({
|
||||
where: { userId: "u1" },
|
||||
data: { backfillProgress: { done: ["fetchHeartRate"], cursors: {} } },
|
||||
});
|
||||
});
|
||||
|
||||
it("leaves a collection for the next slice once the slice has no budget left", async () => {
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
const walk = vi.fn(async () => 5);
|
||||
const budget = googleHealthSliceBudget(0, 100);
|
||||
const { result } = await runWithGoogleHealthClientOutcome(() =>
|
||||
runWithGoogleHealthBackfillResume("u1", new Set(), () =>
|
||||
runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => runGoogleHealthCollection("fetchHeartRate", walk),
|
||||
budget,
|
||||
),
|
||||
),
|
||||
);
|
||||
expect(walk).not.toHaveBeenCalled();
|
||||
expect(result.collections).toEqual([
|
||||
{ key: "fetchHeartRate", status: "deferred", pages: 0, durationMs: 0 },
|
||||
]);
|
||||
expect(budget.stopReason).toBe("time_budget");
|
||||
});
|
||||
|
||||
it("marks the cycle incomplete when it leaves a collection for later", async () => {
|
||||
const { outcome } = await runWithGoogleHealthClientOutcome(() =>
|
||||
runWithGoogleHealthBackfillResume("u1", new Set(), () =>
|
||||
runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => runGoogleHealthCollection("fetchWeight", async () => 0),
|
||||
googleHealthSliceBudget(0, 100),
|
||||
),
|
||||
),
|
||||
);
|
||||
expect(outcome.truncated).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -208,7 +208,8 @@ describe("fetchDailyRollUp — chunk walk + conservative fallback", () => {
|
||||
});
|
||||
|
||||
it("propagates a non-400 first-request failure without a fallback walk", async () => {
|
||||
safeFetchMock.mockResolvedValueOnce(jsonResponse(500, {}));
|
||||
// A 403 is neither the fallback's 400 nor a server error worth repeating.
|
||||
safeFetchMock.mockResolvedValueOnce(jsonResponse(403, {}));
|
||||
await expect(
|
||||
fetchDailyRollUp(GOOGLE_HEALTH_DATA_TYPES.steps, "token", "fetchSteps", {
|
||||
start: new Date("2026-06-01T00:00:00Z"),
|
||||
|
||||
@@ -32,6 +32,10 @@ vi.mock("../sync-core", () => ({
|
||||
(_userId: string, info: { accessToken: string }) => async () =>
|
||||
info.accessToken,
|
||||
runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
|
||||
// Outside a backfill: no resume edge, nothing to record.
|
||||
googleHealthBackfillCursor: () => null,
|
||||
googleHealthHardFailureCount: () => 0,
|
||||
recordGoogleHealthBackfillPage: vi.fn(async () => {}),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
|
||||
@@ -58,6 +58,10 @@ vi.mock("../sync-core", () => ({
|
||||
(_userId: string, info: { accessToken: string }) => async () =>
|
||||
info.accessToken,
|
||||
runGoogleHealthCollection: (_key: string, fn: () => Promise<number>) => fn(),
|
||||
// Outside a backfill: no resume edge, nothing to record.
|
||||
googleHealthBackfillCursor: () => null,
|
||||
googleHealthHardFailureCount: () => 0,
|
||||
recordGoogleHealthBackfillPage: vi.fn(async () => {}),
|
||||
upsertGoogleHealthMeasurements: vi.fn(async () => ({
|
||||
imported: 0,
|
||||
touched: [],
|
||||
|
||||
@@ -0,0 +1,192 @@
|
||||
/**
|
||||
* #1220 — the list walk under a backfill slice.
|
||||
*
|
||||
* - A server error or a request that got no answer is repeated a few times
|
||||
* instead of ending a walk of thousands of pages.
|
||||
* - A resumable walk asks only for the samples beyond the edge an earlier
|
||||
* attempt reached.
|
||||
* - Inside a slice, a resumable walk stops on the slice's budget and on a
|
||||
* rate limit it cannot wait out, and reports itself truncated rather than
|
||||
* failing; a walk that is not resumable runs on under the job's limit.
|
||||
*/
|
||||
import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { safeFetchMock } = vi.hoisted(() => ({ safeFetchMock: vi.fn() }));
|
||||
vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("@/lib/safe-fetch")>();
|
||||
return { ...actual, safeFetch: safeFetchMock };
|
||||
});
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
annotate: () => {},
|
||||
getEvent: () => null,
|
||||
}));
|
||||
|
||||
import { SafeFetchError } from "@/lib/safe-fetch";
|
||||
import {
|
||||
GOOGLE_HEALTH_DATA_TYPES,
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY,
|
||||
forEachDataPointPage,
|
||||
googleHealthSliceBudget,
|
||||
runWithGoogleHealthClientOutcome,
|
||||
runWithGoogleHealthStop,
|
||||
} from "../client";
|
||||
import { GoogleHealthApiError } from "../response-classifier";
|
||||
|
||||
function page(n: number, next: boolean, status = 200, headers = {}): Response {
|
||||
return new Response(
|
||||
JSON.stringify(
|
||||
status === 200
|
||||
? {
|
||||
dataPoints: [{ n }],
|
||||
...(next ? { nextPageToken: String(n + 1) } : {}),
|
||||
}
|
||||
: { error: { code: status } },
|
||||
),
|
||||
{ status, headers: { "content-type": "application/json", ...headers } },
|
||||
);
|
||||
}
|
||||
|
||||
const baseDelay = GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs;
|
||||
beforeEach(() => {
|
||||
safeFetchMock.mockReset();
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0;
|
||||
});
|
||||
afterAll(() => {
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = baseDelay;
|
||||
});
|
||||
|
||||
const walk = (query: Parameters<typeof forEachDataPointPage>[3] = {}) =>
|
||||
runWithGoogleHealthClientOutcome(() =>
|
||||
forEachDataPointPage(
|
||||
GOOGLE_HEALTH_DATA_TYPES.heartRate,
|
||||
"token",
|
||||
"fetchHeartRate",
|
||||
query,
|
||||
() => {},
|
||||
),
|
||||
);
|
||||
|
||||
describe("a server error or a lost request mid-walk", () => {
|
||||
it("is repeated and the walk goes on", async () => {
|
||||
safeFetchMock
|
||||
.mockResolvedValueOnce(page(0, true))
|
||||
.mockResolvedValueOnce(page(0, false, 503))
|
||||
.mockRejectedValueOnce(new SafeFetchError("aborted", "timeout"))
|
||||
.mockResolvedValueOnce(page(1, false));
|
||||
|
||||
const { outcome } = await walk();
|
||||
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(4);
|
||||
expect(outcome).toMatchObject({ pages: 2, truncated: false });
|
||||
});
|
||||
|
||||
it("fails the walk once the retries are spent", async () => {
|
||||
safeFetchMock.mockResolvedValue(page(0, false, 500));
|
||||
await expect(walk()).rejects.toBeInstanceOf(GoogleHealthApiError);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries + 1,
|
||||
);
|
||||
});
|
||||
|
||||
it("does not repeat an error that is not about the server", async () => {
|
||||
safeFetchMock.mockRejectedValue(new TypeError("bad request init"));
|
||||
await expect(walk()).rejects.toBeInstanceOf(TypeError);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a resumable walk", () => {
|
||||
it("asks only for the samples at or beyond the recorded edge", async () => {
|
||||
safeFetchMock.mockResolvedValue(page(0, false));
|
||||
await walk({
|
||||
resumable: true,
|
||||
edge: { at: new Date("2025-03-01T08:00:00Z"), order: "desc" },
|
||||
});
|
||||
const url = new URL(safeFetchMock.mock.calls[0]![0] as string);
|
||||
expect(url.searchParams.get("filter")).toBe(
|
||||
'heart_rate.sample_time.physical_time <= "2025-03-01T08:00:00.000Z"',
|
||||
);
|
||||
});
|
||||
|
||||
it("stops on the slice's page budget and reports itself truncated", async () => {
|
||||
safeFetchMock.mockImplementation(async (url: string) =>
|
||||
page(Number(new URL(url).searchParams.get("pageToken") ?? 0), true),
|
||||
);
|
||||
const budget = googleHealthSliceBudget(60_000, 3);
|
||||
const { outcome } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => walk({ resumable: true }),
|
||||
budget,
|
||||
);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(3);
|
||||
expect(outcome.truncated).toBe(true);
|
||||
expect(budget.stopReason).toBe("page_budget");
|
||||
});
|
||||
|
||||
it("starts the slice's clock at the first walk, not when the slice was set up", async () => {
|
||||
safeFetchMock.mockResolvedValue(page(0, false));
|
||||
let clock = 0;
|
||||
const budget = googleHealthSliceBudget(1_000, 500, () => clock);
|
||||
// The slice waited ten minutes for the account's lock before walking.
|
||||
clock = 600_000;
|
||||
const { outcome } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => walk({ resumable: true }),
|
||||
budget,
|
||||
);
|
||||
expect(outcome).toMatchObject({ pages: 1, truncated: false });
|
||||
expect(budget.deadline).toBe(601_000);
|
||||
});
|
||||
|
||||
it("ends the slice on a rate limit it cannot wait out, without failing", async () => {
|
||||
safeFetchMock
|
||||
.mockResolvedValueOnce(page(0, true))
|
||||
.mockResolvedValueOnce(page(1, true, 429, { "retry-after": "3600" }));
|
||||
const budget = googleHealthSliceBudget(60_000, 500);
|
||||
const { outcome } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => walk({ resumable: true }),
|
||||
budget,
|
||||
);
|
||||
expect(outcome).toMatchObject({ pages: 1, truncated: true });
|
||||
expect(budget.stopReason).toBe("rate_limited");
|
||||
});
|
||||
|
||||
it("still fails on a rate limit outside a slice", async () => {
|
||||
safeFetchMock.mockResolvedValue(
|
||||
page(0, true, 429, { "retry-after": "3600" }),
|
||||
);
|
||||
await expect(walk({ resumable: true })).rejects.toBeInstanceOf(
|
||||
GoogleHealthApiError,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a walk that is not resumable", () => {
|
||||
it("runs past the slice's budget to its end", async () => {
|
||||
safeFetchMock.mockImplementation(async (url: string) => {
|
||||
const n = Number(new URL(url).searchParams.get("pageToken") ?? 0);
|
||||
return page(n, n < 5);
|
||||
});
|
||||
const budget = googleHealthSliceBudget(0, 1);
|
||||
const { outcome } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() => walk(),
|
||||
budget,
|
||||
);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(6);
|
||||
expect(outcome.truncated).toBe(false);
|
||||
});
|
||||
|
||||
it("still stops on the job's own budget", async () => {
|
||||
safeFetchMock.mockImplementation(async () => page(0, true));
|
||||
let asked = 0;
|
||||
const { outcome } = await runWithGoogleHealthStop(
|
||||
() => ++asked > 2,
|
||||
() => walk(),
|
||||
googleHealthSliceBudget(60_000, 500),
|
||||
);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(outcome.truncated).toBe(true);
|
||||
});
|
||||
});
|
||||
+247
-20
@@ -36,7 +36,7 @@
|
||||
import { AsyncLocalStorage } from "node:async_hooks";
|
||||
import { createHash, randomBytes } from "node:crypto";
|
||||
import { annotate, getEvent } from "@/lib/logging/context";
|
||||
import { safeFetch } from "@/lib/safe-fetch";
|
||||
import { SafeFetchError, safeFetch } from "@/lib/safe-fetch";
|
||||
import { wallClockInTz } from "@/lib/tz/wall-clock";
|
||||
import {
|
||||
GoogleHealthApiError,
|
||||
@@ -134,18 +134,140 @@ export async function runWithGoogleHealthClientOutcome<T>(
|
||||
* reaching the page ceiling: the resource is incomplete, the cycle is not
|
||||
* stamped, and the reason shown is that only part of the history was
|
||||
* processed. Outside such a scope the walks never stop early.
|
||||
*
|
||||
* A backfill slice adds a second, tighter limit (`GoogleHealthSliceBudget`):
|
||||
* minutes and pages rather than the job's hours. Only a walk that keeps its
|
||||
* place across attempts (`resumable`) honours it, because only such a walk
|
||||
* loses nothing by stopping; a walk that would start over from its first page
|
||||
* runs on to the end under the job's limit alone.
|
||||
*/
|
||||
const stopStorage = new AsyncLocalStorage<() => boolean>();
|
||||
interface GoogleHealthWalkLimits {
|
||||
shouldStop: () => boolean;
|
||||
slice?: GoogleHealthSliceBudget;
|
||||
}
|
||||
|
||||
function stopRequested(): boolean {
|
||||
return stopStorage.getStore()?.() === true;
|
||||
/** Why a backfill slice stopped walking. */
|
||||
export type GoogleHealthSliceStopReason =
|
||||
"time_budget" | "page_budget" | "rate_limited" | "job_budget";
|
||||
|
||||
/**
|
||||
* One backfill slice's share of the work: a span of wall-clock time and a
|
||||
* page count, both well inside the job's own expiry. The clock starts the
|
||||
* first time a walk asks, which is after the slice has the account's lock: a
|
||||
* slice that waited for an hourly sync to finish does not start out with its
|
||||
* time already spent. `pages` counts every page the slice's walks have read;
|
||||
* `stopReason` is set the first time a limit is reached and stays set, so
|
||||
* every walk after it stops too.
|
||||
*/
|
||||
export interface GoogleHealthSliceBudget {
|
||||
ms: number;
|
||||
/** When the slice's time is up; null until the first walk asks. */
|
||||
deadline: number | null;
|
||||
maxPages: number;
|
||||
pages: number;
|
||||
stopReason: GoogleHealthSliceStopReason | null;
|
||||
now: () => number;
|
||||
}
|
||||
|
||||
/** A fresh slice budget of `ms` milliseconds and `maxPages` pages. */
|
||||
export function googleHealthSliceBudget(
|
||||
ms: number,
|
||||
maxPages: number,
|
||||
now: () => number = Date.now,
|
||||
): GoogleHealthSliceBudget {
|
||||
return {
|
||||
ms,
|
||||
deadline: null,
|
||||
maxPages,
|
||||
pages: 0,
|
||||
stopReason: null,
|
||||
now,
|
||||
};
|
||||
}
|
||||
|
||||
const stopStorage = new AsyncLocalStorage<GoogleHealthWalkLimits>();
|
||||
|
||||
/** Whether the slice in scope has used its budget (and why, once it has). */
|
||||
function sliceExhausted(slice: GoogleHealthSliceBudget): boolean {
|
||||
if (slice.stopReason) return true;
|
||||
slice.deadline ??= slice.now() + slice.ms;
|
||||
if (slice.now() >= slice.deadline) slice.stopReason = "time_budget";
|
||||
else if (slice.pages >= slice.maxPages) slice.stopReason = "page_budget";
|
||||
return slice.stopReason !== null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a walk should stop before its next request. Every walk stops on the
|
||||
* job's limit; a resumable one also stops on the slice's.
|
||||
*/
|
||||
function stopRequested(resumable = false): boolean {
|
||||
const limits = stopStorage.getStore();
|
||||
if (!limits) return false;
|
||||
if (limits.shouldStop()) {
|
||||
if (limits.slice && !limits.slice.stopReason) {
|
||||
limits.slice.stopReason = "job_budget";
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
resumable && limits.slice !== undefined && sliceExhausted(limits.slice)
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether the backfill slice in scope has used its budget. A collection that
|
||||
* has not started yet waits for the next slice rather than starting now.
|
||||
* False outside a slice.
|
||||
*/
|
||||
export function googleHealthSliceExhausted(): boolean {
|
||||
const limits = stopStorage.getStore();
|
||||
if (!limits) return false;
|
||||
if (limits.shouldStop()) {
|
||||
if (limits.slice && !limits.slice.stopReason) {
|
||||
limits.slice.stopReason = "job_budget";
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return limits.slice !== undefined && sliceExhausted(limits.slice);
|
||||
}
|
||||
|
||||
/** Pages the slice in scope has read so far (0 outside a slice). */
|
||||
export function googleHealthSlicePagesRead(): number {
|
||||
return stopStorage.getStore()?.slice?.pages ?? 0;
|
||||
}
|
||||
|
||||
/** Count one page read against the slice in scope, if any. */
|
||||
function noteSlicePage(): void {
|
||||
const slice = stopStorage.getStore()?.slice;
|
||||
if (slice) slice.pages += 1;
|
||||
}
|
||||
|
||||
/** End the slice in scope for `reason` (no-op outside a slice). */
|
||||
function endSlice(reason: GoogleHealthSliceStopReason): void {
|
||||
const slice = stopStorage.getStore()?.slice;
|
||||
if (slice && !slice.stopReason) slice.stopReason = reason;
|
||||
}
|
||||
|
||||
/** Whether a resumable walk can stop and keep its place here. */
|
||||
function inSlice(): boolean {
|
||||
return stopStorage.getStore()?.slice !== undefined;
|
||||
}
|
||||
|
||||
export async function runWithGoogleHealthStop<T>(
|
||||
shouldStop: () => boolean,
|
||||
fn: () => Promise<T>,
|
||||
slice?: GoogleHealthSliceBudget,
|
||||
): Promise<T> {
|
||||
return stopStorage.run(shouldStop, fn);
|
||||
return stopStorage.run({ shouldStop, slice }, fn);
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark the resource in scope truncated without a walk: a collection a slice
|
||||
* left for the next one. The cycle then reports itself incomplete, so the
|
||||
* backfill does not mistake a skipped collection for an empty one.
|
||||
*/
|
||||
export function noteGoogleHealthDeferred(): void {
|
||||
notePagination(0, 0, true);
|
||||
}
|
||||
|
||||
export const GOOGLE_HEALTH_API_BASE = "https://health.googleapis.com/v4";
|
||||
@@ -527,6 +649,7 @@ export function resolveGoogleHealthUserId(
|
||||
// camelCase nested objects (`sampleTime.physicalTime`,
|
||||
// `interval.startTime`, `civilStartTime.date`).
|
||||
//
|
||||
|
||||
// proto3 int64 fields arrive as JSON **strings** (`"12345"`) — every numeric
|
||||
// extractor coerces numeric strings before the finite check.
|
||||
|
||||
@@ -553,6 +676,9 @@ export const GOOGLE_HEALTH_ACTIVITY_PAGE_SIZE = 25;
|
||||
*/
|
||||
export const GOOGLE_HEALTH_DENSE_MAX_PAGES = 10_000;
|
||||
|
||||
/** Page ceiling for every other list walk. */
|
||||
export const GOOGLE_HEALTH_DEFAULT_MAX_PAGES = 1000;
|
||||
|
||||
/** `dataPoints.list` envelope: `{ dataPoints, nextPageToken }`. */
|
||||
interface GoogleHealthDataPointPage {
|
||||
dataPoints?: GoogleHealthDataPoint[];
|
||||
@@ -578,6 +704,20 @@ interface DataPointQuery {
|
||||
* surfaces it so a live account reports which grammar Google accepted.
|
||||
*/
|
||||
onDateFilterStyle?: (style: GoogleHealthDateFilterStyle) => void;
|
||||
/**
|
||||
* The walk keeps its place across attempts (a backfill's resume cursor), so
|
||||
* it may stop on the backfill slice's budget and on a rate limit it cannot
|
||||
* wait out, and report itself truncated instead of failing: the next slice
|
||||
* picks up from the edge it recorded.
|
||||
*/
|
||||
resumable?: boolean;
|
||||
/**
|
||||
* Where an earlier attempt of this walk got to: the oldest sample it wrote
|
||||
* when the walk ran newest first (`desc`), the newest when it ran oldest
|
||||
* first (`asc`). The walk asks only for the samples at or beyond it; the
|
||||
* one sample on the edge itself is read again and overwrites in place.
|
||||
*/
|
||||
edge?: { at: Date; order: "desc" | "asc" };
|
||||
}
|
||||
|
||||
const pad2 = (n: number): string => String(n).padStart(2, "0");
|
||||
@@ -797,6 +937,21 @@ export function googleHealthRetryDelayMs(
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* How a data request rides out a server that did not answer properly: a
|
||||
* 500, 502, 503 or 504, a timeout, or a broken connection. A short history
|
||||
* reads a handful of pages and rarely meets one; a full-history heart-rate
|
||||
* walk reads thousands, and a single one used to end the walk and throw away
|
||||
* everything it had read. Three tries more, 2, 4 and 8 s apart (or the
|
||||
* `Retry-After` a 503 names, up to a minute), before the request fails.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_TRANSIENT_RETRY = {
|
||||
maxRetries: 3,
|
||||
baseDelayMs: 2_000,
|
||||
};
|
||||
|
||||
const GOOGLE_HEALTH_TRANSIENT_STATUSES = new Set([500, 502, 503, 504]);
|
||||
|
||||
function waitMs(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
@@ -811,10 +966,12 @@ interface GoogleHealthDataResponse<T> {
|
||||
/**
|
||||
* Send one data request and answer with the response it finally got.
|
||||
*
|
||||
* Two answers are handled here rather than handed back:
|
||||
* Three answers are handled here rather than handed back:
|
||||
* - 429: the request waits (see `GOOGLE_HEALTH_RATE_LIMIT`) and is sent
|
||||
* again, up to `maxRetries` times, unless the caller's stop condition
|
||||
* says to end the walk.
|
||||
* - a server error or no answer at all (see
|
||||
* `GOOGLE_HEALTH_TRANSIENT_RETRY`): the same, with its own count.
|
||||
* - 401 with a token source: the token expired or was revoked between
|
||||
* the source's last check and this request. The source is asked once for
|
||||
* a fresh token and the request is repeated. A second 401, or a 401 on a
|
||||
@@ -830,6 +987,7 @@ async function googleHealthDataRequest<T>(
|
||||
isExpectedFallback: (status: number) => boolean = () => false,
|
||||
): Promise<GoogleHealthDataResponse<T>> {
|
||||
let rateLimited = 0;
|
||||
let transient = 0;
|
||||
let refreshed = false;
|
||||
let forceRefresh = false;
|
||||
for (;;) {
|
||||
@@ -837,7 +995,33 @@ async function googleHealthDataRequest<T>(
|
||||
typeof token === "string" ? token : await token({ forceRefresh });
|
||||
forceRefresh = false;
|
||||
const started = performance.now();
|
||||
const res = await send(accessToken);
|
||||
let res: Response;
|
||||
try {
|
||||
res = await send(accessToken);
|
||||
} catch (err) {
|
||||
// No answer at all: the request timed out or the connection broke.
|
||||
// On a walk of a thousand pages one of those is a matter of time, and
|
||||
// it cost the whole walk.
|
||||
const retry =
|
||||
err instanceof SafeFetchError &&
|
||||
(err.kind === "timeout" || err.kind === "network") &&
|
||||
transient < GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries &&
|
||||
!stopRequested();
|
||||
const delayMs =
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs * 2 ** transient;
|
||||
getEvent()?.addExternalCall({
|
||||
service: "google-health",
|
||||
method: label,
|
||||
duration_ms: Math.round(performance.now() - started),
|
||||
...(retry
|
||||
? { note: `no_response_retrying_in_ms=${delayMs}` }
|
||||
: { error: err instanceof Error ? err.message : String(err) }),
|
||||
});
|
||||
if (!retry) throw err;
|
||||
transient += 1;
|
||||
await waitMs(delayMs);
|
||||
continue;
|
||||
}
|
||||
const json = (await res.json().catch(() => null)) as T | null;
|
||||
const verdict = classifyGoogleHealthResponse(res.status);
|
||||
const apiErrorDetail =
|
||||
@@ -859,6 +1043,17 @@ async function googleHealthDataRequest<T>(
|
||||
!stopRequested();
|
||||
const willRefresh =
|
||||
res.status === 401 && typeof token !== "string" && !refreshed;
|
||||
const transientDelayMs = GOOGLE_HEALTH_TRANSIENT_STATUSES.has(res.status)
|
||||
? Math.min(
|
||||
googleHealthRetryDelayMs(res, json) ??
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs * 2 ** transient,
|
||||
GOOGLE_HEALTH_RATE_LIMIT.maxDelayMs,
|
||||
)
|
||||
: 0;
|
||||
const willRetryTransient =
|
||||
GOOGLE_HEALTH_TRANSIENT_STATUSES.has(res.status) &&
|
||||
transient < GOOGLE_HEALTH_TRANSIENT_RETRY.maxRetries &&
|
||||
!stopRequested();
|
||||
|
||||
getEvent()?.addExternalCall({
|
||||
service: "google-health",
|
||||
@@ -871,14 +1066,16 @@ async function googleHealthDataRequest<T>(
|
||||
? { note: `rate_limited_retrying_in_ms=${backoffMs}` }
|
||||
: willRefresh
|
||||
? { note: "token_rejected_retrying_with_fresh_token" }
|
||||
: {
|
||||
error:
|
||||
verdict.classification === "success"
|
||||
? undefined
|
||||
: apiErrorDetail
|
||||
? `${verdict.reason} ${apiErrorDetail}`
|
||||
: verdict.reason,
|
||||
}),
|
||||
: willRetryTransient
|
||||
? { note: `server_error_retrying_in_ms=${transientDelayMs}` }
|
||||
: {
|
||||
error:
|
||||
verdict.classification === "success"
|
||||
? undefined
|
||||
: apiErrorDetail
|
||||
? `${verdict.reason} ${apiErrorDetail}`
|
||||
: verdict.reason,
|
||||
}),
|
||||
});
|
||||
|
||||
if (willWait) {
|
||||
@@ -886,6 +1083,11 @@ async function googleHealthDataRequest<T>(
|
||||
await waitMs(backoffMs);
|
||||
continue;
|
||||
}
|
||||
if (willRetryTransient) {
|
||||
transient += 1;
|
||||
await waitMs(transientDelayMs);
|
||||
continue;
|
||||
}
|
||||
if (willRefresh) {
|
||||
refreshed = true;
|
||||
forceRefresh = true;
|
||||
@@ -947,8 +1149,9 @@ export async function forEachDataPointPage(
|
||||
query: DataPointQuery,
|
||||
onPage: (points: GoogleHealthDataPoint[]) => void | Promise<void>,
|
||||
): Promise<void> {
|
||||
const maxPages = query.maxPages ?? 1000;
|
||||
const maxPages = query.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES;
|
||||
const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE;
|
||||
const resumable = query.resumable === true;
|
||||
|
||||
let requestCount = 0;
|
||||
|
||||
@@ -961,11 +1164,12 @@ export async function forEachDataPointPage(
|
||||
let stopped = false;
|
||||
|
||||
do {
|
||||
if (stopRequested()) {
|
||||
if (stopRequested(resumable)) {
|
||||
stopped = true;
|
||||
break;
|
||||
}
|
||||
const params = new URLSearchParams({ pageSize: String(pageSize) });
|
||||
const filters: string[] = [];
|
||||
if (query.start) {
|
||||
const { field, bound } = incrementalFilter(
|
||||
dataType,
|
||||
@@ -973,8 +1177,20 @@ export async function forEachDataPointPage(
|
||||
query.tz,
|
||||
dateStyle,
|
||||
);
|
||||
params.set("filter", `${field} >= "${bound}"`);
|
||||
filters.push(`${field} >= "${bound}"`);
|
||||
}
|
||||
if (query.edge) {
|
||||
const { field, bound } = incrementalFilter(
|
||||
dataType,
|
||||
query.edge.at,
|
||||
query.tz,
|
||||
dateStyle,
|
||||
);
|
||||
filters.push(
|
||||
`${field} ${query.edge.order === "desc" ? "<=" : ">="} "${bound}"`,
|
||||
);
|
||||
}
|
||||
if (filters.length > 0) params.set("filter", filters.join(" AND "));
|
||||
if (pageToken) params.set("pageToken", pageToken);
|
||||
|
||||
requestCount += 1;
|
||||
@@ -985,7 +1201,7 @@ export async function forEachDataPointPage(
|
||||
// an error next to the 200 that followed it. It is recorded as a note
|
||||
// now; any other 400 still carries its error.
|
||||
const firstRequest = requestCount === 1;
|
||||
const { json, verdict, apiErrorDetail } =
|
||||
const { status, json, verdict, apiErrorDetail } =
|
||||
await googleHealthDataRequest<GoogleHealthDataPointPage>(
|
||||
accessToken,
|
||||
`${verb}(page=${pageCount})`,
|
||||
@@ -1003,6 +1219,15 @@ export async function forEachDataPointPage(
|
||||
firstRequest &&
|
||||
status === 400,
|
||||
);
|
||||
// A rate limit the request could not wait out ends a resumable walk
|
||||
// inside a backfill slice where it is: the pages before it are written
|
||||
// and their edge recorded, so this is a pause, not a failure. The slice
|
||||
// ends with it, since every other walk would meet the same quota.
|
||||
if (status === 429 && resumable && inSlice()) {
|
||||
endSlice("rate_limited");
|
||||
stopped = true;
|
||||
break;
|
||||
}
|
||||
if (verdict.classification !== "success") {
|
||||
throw new GoogleHealthApiError({
|
||||
verb,
|
||||
@@ -1017,6 +1242,7 @@ export async function forEachDataPointPage(
|
||||
pageToken = json?.nextPageToken ?? null;
|
||||
pageCount += 1;
|
||||
fetched += page.length;
|
||||
noteSlicePage();
|
||||
await onPage(page);
|
||||
} while (pageToken && pageCount < maxPages);
|
||||
|
||||
@@ -1245,7 +1471,7 @@ export async function fetchDailyRollUp(
|
||||
let totalPages = 0;
|
||||
let truncated = false;
|
||||
for (const chunk of chunkCivilRange(startCivil, endCivil, maxDays)) {
|
||||
if (stopRequested()) {
|
||||
if (stopRequested(false)) {
|
||||
truncated = true;
|
||||
break;
|
||||
}
|
||||
@@ -1295,6 +1521,7 @@ export async function fetchDailyRollUp(
|
||||
for (const p of json?.rollupDataPoints ?? []) points.push(p);
|
||||
pageToken = json?.nextPageToken ?? null;
|
||||
pageCount += 1;
|
||||
noteSlicePage();
|
||||
} while (pageToken && pageCount < 100);
|
||||
totalPages += pageCount;
|
||||
truncated ||= Boolean(pageToken);
|
||||
|
||||
@@ -387,6 +387,21 @@ function resolveMeasuredAt(
|
||||
return fallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* The sample instant of a spot-reading `DataPoint` (`sample` data types), or
|
||||
* null for any other shape or a point without a parseable time. A backfill
|
||||
* reads it to record how far a walk has got.
|
||||
*/
|
||||
export function googleHealthSampleInstant(
|
||||
point: GoogleHealthDataPoint,
|
||||
dataType: GoogleHealthDataType,
|
||||
tz?: string,
|
||||
): Date | null {
|
||||
if (dataType.timeField !== "sample") return null;
|
||||
const t = readPath(point, `${dataType.key}.sampleTime.physicalTime`);
|
||||
return typeof t === "string" ? parseLocalInstant(t, tz) : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stable anchor for a `DataPoint`'s externalId. A spot reading anchors on its
|
||||
* sample time; a daily summary on its civil date. Combined with a type-specific
|
||||
|
||||
@@ -37,6 +37,9 @@ import {
|
||||
import { invalidateStatusInsightsForTypes } from "@/lib/insights/comprehensive-generate";
|
||||
import {
|
||||
googleHealthOutcomeTruncated,
|
||||
googleHealthSliceExhausted,
|
||||
googleHealthSlicePagesRead,
|
||||
noteGoogleHealthDeferred,
|
||||
noteGoogleHealthMapped,
|
||||
noteGoogleHealthOutcomeFailure,
|
||||
noteGoogleHealthWritten,
|
||||
@@ -402,33 +405,75 @@ export async function runWithGoogleHealthSyncCycle<T>(
|
||||
}
|
||||
|
||||
/**
|
||||
* Collections a full-history backfill already walked to the end, carried from
|
||||
* one attempt of the backfill to the next.
|
||||
* How far a full-history backfill has got, carried from one attempt of the
|
||||
* backfill to the next.
|
||||
*
|
||||
* A backfill that ended early (its job budget ran out, a collection failed)
|
||||
* used to start the next attempt from the first page of the first collection,
|
||||
* so the dense heart-rate walk, thousands of pages, was paid for again on
|
||||
* every retry, and an account whose walk never fit into one attempt never
|
||||
* finished at all. Each collection that completes cleanly is now written to
|
||||
* `backfillProgress`, and the next attempt skips it and goes on with the
|
||||
* collections it did not reach. A collection is the unit because Google
|
||||
* offers no cursor that survives between attempts; its page tokens are not
|
||||
* documented to outlive the walk that received them.
|
||||
* Two grains. A collection walked to the end is noted in `done`, and the next
|
||||
* attempt skips it. A collection read page by page (the spot readings: heart
|
||||
* rate, weight, glucose, ...) also notes, after every page it writes, the
|
||||
* edge it has reached: the oldest sample written so far when the walk runs
|
||||
* newest first, as Google's list does. The next attempt asks Google only for
|
||||
* the samples at or beyond that edge, so a collection that does not fit into
|
||||
* one attempt is worked through over several instead of being read from its
|
||||
* first page every time.
|
||||
*
|
||||
* The edge is a sample time, not Google's page token: page tokens are not
|
||||
* documented to outlive the walk that received them, a sample time is a
|
||||
* filter Google documents. The one sample on the edge itself is read twice
|
||||
* and overwrites its own row.
|
||||
*
|
||||
* Before the edge existed only `done` was kept, and a heart-rate history of
|
||||
* years that one attempt could not finish (its budget, a rate limit, one
|
||||
* failed page among thousands) started again from the newest page on every
|
||||
* attempt and never reached the end (#1220).
|
||||
*/
|
||||
export interface GoogleHealthBackfillCursor {
|
||||
/** The edge: the sample instant (ISO 8601) the walk has written up to. */
|
||||
at: string;
|
||||
/** The order the walk returned samples in: newest first, or oldest first. */
|
||||
order: "desc" | "asc";
|
||||
/** Pages this collection has read across every attempt so far. */
|
||||
pages: number;
|
||||
}
|
||||
|
||||
/** What one collection came to in one backfill attempt. */
|
||||
export interface GoogleHealthBackfillCollectionReport {
|
||||
key: string;
|
||||
/**
|
||||
* `done`: walked to the end. `stopped`: ended short on a budget, a page
|
||||
* ceiling or a rate limit, with nothing failed. `deferred`: not started,
|
||||
* left for the next slice. `failed`: a request or a write failed, or the
|
||||
* collection answered 403.
|
||||
*/
|
||||
status: "done" | "stopped" | "deferred" | "failed";
|
||||
pages: number;
|
||||
durationMs: number;
|
||||
/** The edge before and after this attempt, for a resumable collection. */
|
||||
edgeFrom?: string | null;
|
||||
edgeTo?: string | null;
|
||||
}
|
||||
|
||||
interface BackfillResumeTracker {
|
||||
userId: string;
|
||||
done: Set<string>;
|
||||
cursors: Map<string, GoogleHealthBackfillCursor>;
|
||||
/** Collections this attempt skipped because an earlier one finished them. */
|
||||
skipped: string[];
|
||||
/** Collections whose edge this attempt moved, with the edge they started from. */
|
||||
advanced: Map<string, string | null>;
|
||||
collections: GoogleHealthBackfillCollectionReport[];
|
||||
}
|
||||
const backfillResumeStorage = new AsyncLocalStorage<BackfillResumeTracker>();
|
||||
|
||||
function progressRecord(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
}
|
||||
|
||||
/** Read the `backfillProgress` column into the set of finished collections. */
|
||||
export function readGoogleHealthBackfillDone(value: unknown): Set<string> {
|
||||
const done =
|
||||
value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as { done?: unknown }).done
|
||||
: undefined;
|
||||
const done = progressRecord(value).done;
|
||||
return new Set(
|
||||
Array.isArray(done)
|
||||
? done.filter(
|
||||
@@ -438,20 +483,188 @@ export function readGoogleHealthBackfillDone(value: unknown): Set<string> {
|
||||
);
|
||||
}
|
||||
|
||||
/** Read the `backfillProgress` column into the per-collection edges. */
|
||||
export function readGoogleHealthBackfillCursors(
|
||||
value: unknown,
|
||||
): Map<string, GoogleHealthBackfillCursor> {
|
||||
const out = new Map<string, GoogleHealthBackfillCursor>();
|
||||
const cursors = progressRecord(progressRecord(value).cursors);
|
||||
for (const [key, raw] of Object.entries(cursors)) {
|
||||
if (key.length > 64) continue;
|
||||
const c = progressRecord(raw);
|
||||
const at = typeof c.at === "string" ? c.at : null;
|
||||
if (!at || Number.isNaN(Date.parse(at))) continue;
|
||||
out.set(key, {
|
||||
at: new Date(at).toISOString(),
|
||||
order: c.order === "asc" ? "asc" : "desc",
|
||||
pages:
|
||||
typeof c.pages === "number" && Number.isFinite(c.pages) && c.pages > 0
|
||||
? Math.trunc(c.pages)
|
||||
: 0,
|
||||
});
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Write the tracker's progress with one change applied, and apply it to the
|
||||
* tracker only once the write has landed: progress that is not on disk is
|
||||
* progress the next attempt cannot see, so it must not count in this one.
|
||||
*/
|
||||
async function persistBackfillProgress(
|
||||
tracker: BackfillResumeTracker,
|
||||
next: {
|
||||
done: Set<string>;
|
||||
cursors: Map<string, GoogleHealthBackfillCursor>;
|
||||
},
|
||||
): Promise<void> {
|
||||
await prisma.googleHealthConnection.update({
|
||||
where: { userId: tracker.userId },
|
||||
data: {
|
||||
backfillProgress: {
|
||||
done: [...next.done],
|
||||
cursors: Object.fromEntries(
|
||||
[...next.cursors].map(([key, c]) => [
|
||||
key,
|
||||
{ at: c.at, order: c.order, pages: c.pages },
|
||||
]),
|
||||
),
|
||||
},
|
||||
},
|
||||
});
|
||||
// `next` may be the tracker's own collections: copy before clearing.
|
||||
const done = [...next.done];
|
||||
const cursors = [...next.cursors];
|
||||
tracker.done.clear();
|
||||
for (const key of done) tracker.done.add(key);
|
||||
tracker.cursors.clear();
|
||||
for (const [key, c] of cursors) tracker.cursors.set(key, c);
|
||||
}
|
||||
|
||||
/** What one backfill attempt did, as `runWithGoogleHealthBackfillResume` reports it. */
|
||||
export interface GoogleHealthBackfillAttempt<T> {
|
||||
result: T;
|
||||
skipped: string[];
|
||||
/** Collections whose edge moved, each with the edge it started from. */
|
||||
advanced: Map<string, string | null>;
|
||||
collections: GoogleHealthBackfillCollectionReport[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a full-history backfill attempt that resumes: collections in `done` are
|
||||
* skipped, and each collection finished cleanly during this attempt is added
|
||||
* to it and persisted at once, so an attempt that dies part-way keeps what it
|
||||
* finished.
|
||||
* skipped, a collection with an edge in `cursors` starts from it, and each
|
||||
* collection finished or page written during this attempt is persisted at
|
||||
* once, so an attempt that dies part-way keeps what it did.
|
||||
*/
|
||||
export async function runWithGoogleHealthBackfillResume<T>(
|
||||
userId: string,
|
||||
done: Set<string>,
|
||||
fn: () => Promise<T>,
|
||||
): Promise<{ result: T; skipped: string[] }> {
|
||||
const tracker: BackfillResumeTracker = { userId, done, skipped: [] };
|
||||
cursors: Map<string, GoogleHealthBackfillCursor> = new Map(),
|
||||
): Promise<GoogleHealthBackfillAttempt<T>> {
|
||||
const tracker: BackfillResumeTracker = {
|
||||
userId,
|
||||
done,
|
||||
cursors,
|
||||
skipped: [],
|
||||
advanced: new Map(),
|
||||
collections: [],
|
||||
};
|
||||
const result = await backfillResumeStorage.run(tracker, fn);
|
||||
return { result, skipped: tracker.skipped };
|
||||
return {
|
||||
result,
|
||||
skipped: tracker.skipped,
|
||||
advanced: tracker.advanced,
|
||||
collections: tracker.collections,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Where a resumable collection's walk starts in this backfill attempt: the
|
||||
* edge an earlier attempt recorded (if any) and the pages it has already
|
||||
* read. Null outside a backfill, where every walk reads its whole window.
|
||||
*/
|
||||
export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): {
|
||||
edge?: { at: Date; order: "desc" | "asc" };
|
||||
pagesSoFar: number;
|
||||
} | null {
|
||||
const resume = backfillResumeStorage.getStore();
|
||||
if (!resume) return null;
|
||||
const cursor = resume.cursors.get(key);
|
||||
return cursor
|
||||
? {
|
||||
edge: { at: new Date(cursor.at), order: cursor.order },
|
||||
pagesSoFar: cursor.pages,
|
||||
}
|
||||
: { pagesSoFar: 0 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Record that a resumable collection wrote one more page, holding samples
|
||||
* from `first` to `last` in the order Google returned them. Moves the
|
||||
* collection's edge to the far end of the page and persists it. Outside a
|
||||
* backfill this does nothing.
|
||||
*
|
||||
* The order is read off the first page whose two ends differ and kept for
|
||||
* the rest of the backfill; Google returns its lists newest first, which is
|
||||
* also what a page that cannot tell is taken to mean.
|
||||
*/
|
||||
export async function recordGoogleHealthBackfillPage(
|
||||
key: GoogleHealthCollectionKey,
|
||||
page: { first: Date | null; last: Date | null },
|
||||
): Promise<void> {
|
||||
const resume = backfillResumeStorage.getStore();
|
||||
if (!resume) return;
|
||||
const current = resume.cursors.get(key);
|
||||
const order: "desc" | "asc" =
|
||||
current?.order ??
|
||||
(page.first && page.last && page.first.getTime() < page.last.getTime()
|
||||
? "asc"
|
||||
: "desc");
|
||||
const ends = [page.first, page.last].filter((d): d is Date => d !== null);
|
||||
const far =
|
||||
ends.length === 0
|
||||
? null
|
||||
: new Date(
|
||||
order === "desc"
|
||||
? Math.min(...ends.map((d) => d.getTime()))
|
||||
: Math.max(...ends.map((d) => d.getTime())),
|
||||
);
|
||||
const currentAt = current ? Date.parse(current.at) : null;
|
||||
const moved =
|
||||
far !== null &&
|
||||
(currentAt === null ||
|
||||
(order === "desc"
|
||||
? far.getTime() < currentAt
|
||||
: far.getTime() > currentAt));
|
||||
if (!current && !moved) return;
|
||||
const nextCursor: GoogleHealthBackfillCursor = {
|
||||
at: moved ? far.toISOString() : current!.at,
|
||||
order,
|
||||
pages: (current?.pages ?? 0) + 1,
|
||||
};
|
||||
const cursors = new Map(resume.cursors);
|
||||
cursors.set(key, nextCursor);
|
||||
try {
|
||||
await persistBackfillProgress(resume, { done: resume.done, cursors });
|
||||
if (moved && !resume.advanced.has(key)) {
|
||||
resume.advanced.set(key, current?.at ?? null);
|
||||
}
|
||||
} catch (err) {
|
||||
// The page's rows are written; only the note of how far the walk got is
|
||||
// missing, so the next attempt reads this page again. Not a failure.
|
||||
getEvent()?.addWarning(
|
||||
`google-health: backfill progress write failed for ${key}: ${err}`,
|
||||
);
|
||||
annotate({
|
||||
meta: { "googleHealth.backfill.progressWriteFailed": key },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** Hard failures the cycle in scope has recorded so far. */
|
||||
export function googleHealthHardFailureCount(): number {
|
||||
return hardFailStorage.getStore()?.failures.length ?? 0;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -459,10 +672,12 @@ export async function runWithGoogleHealthBackfillResume<T>(
|
||||
* one in scope. Outside a backfill this is just `fn()`.
|
||||
*
|
||||
* Inside one, a collection already walked to the end by an earlier attempt is
|
||||
* skipped. Otherwise it runs, and it counts as finished only when nothing
|
||||
* went wrong while it ran: no hard failure on the cycle's ledger, no 403
|
||||
* soft-skip, and no walk that ended short (its page ceiling, or the job's
|
||||
* budget asking it to stop). Anything less is walked again next attempt.
|
||||
* skipped, and so is one the slice has no budget left to start: it is left
|
||||
* for the next slice, and the cycle reports itself incomplete. Otherwise it
|
||||
* runs, and it counts as finished only when nothing went wrong while it ran:
|
||||
* no hard failure on the cycle's ledger, no 403 soft-skip, and no walk that
|
||||
* ended short (its page ceiling, or a budget asking it to stop). Anything
|
||||
* less is walked again next attempt, from its edge when it has one.
|
||||
*/
|
||||
export async function runGoogleHealthCollection(
|
||||
key: GoogleHealthCollectionKey,
|
||||
@@ -474,15 +689,29 @@ export async function runGoogleHealthCollection(
|
||||
resume.skipped.push(key);
|
||||
return 0;
|
||||
}
|
||||
if (googleHealthSliceExhausted()) {
|
||||
noteGoogleHealthDeferred();
|
||||
resume.collections.push({
|
||||
key,
|
||||
status: "deferred",
|
||||
pages: 0,
|
||||
durationMs: 0,
|
||||
});
|
||||
return 0;
|
||||
}
|
||||
const startedAt = performance.now();
|
||||
const pagesBefore = googleHealthSlicePagesRead();
|
||||
const edgeFrom = resume.cursors.get(key)?.at ?? null;
|
||||
const hardBefore = hardFailStorage.getStore()?.failures.length ?? 0;
|
||||
const softBefore = softSkipStorage.getStore()?.count ?? 0;
|
||||
const truncatedBefore = googleHealthOutcomeTruncated();
|
||||
const imported = await fn();
|
||||
const clean =
|
||||
(hardFailStorage.getStore()?.failures.length ?? 0) === hardBefore &&
|
||||
(softSkipStorage.getStore()?.count ?? 0) === softBefore &&
|
||||
!truncatedBefore &&
|
||||
!googleHealthOutcomeTruncated();
|
||||
const failedHard =
|
||||
(hardFailStorage.getStore()?.failures.length ?? 0) !== hardBefore ||
|
||||
(softSkipStorage.getStore()?.count ?? 0) !== softBefore;
|
||||
const truncated = truncatedBefore || googleHealthOutcomeTruncated();
|
||||
const clean = !failedHard && !truncated;
|
||||
let finished = false;
|
||||
if (clean) {
|
||||
// The collection counts as finished only once the note says so. The job
|
||||
// reads the size of `done` as this attempt's progress and sends another
|
||||
@@ -490,11 +719,13 @@ export async function runGoogleHealthCollection(
|
||||
// progress nobody can see next attempt, and a chain of slices could keep
|
||||
// "progressing" over the same collection forever.
|
||||
try {
|
||||
await prisma.googleHealthConnection.update({
|
||||
where: { userId: resume.userId },
|
||||
data: { backfillProgress: { done: [...resume.done, key] } },
|
||||
const cursors = new Map(resume.cursors);
|
||||
cursors.delete(key);
|
||||
await persistBackfillProgress(resume, {
|
||||
done: new Set([...resume.done, key]),
|
||||
cursors,
|
||||
});
|
||||
resume.done.add(key);
|
||||
finished = true;
|
||||
} catch (err) {
|
||||
// Not finished, then: the collection is walked again next attempt. It
|
||||
// must not fail a collection whose rows are written.
|
||||
@@ -506,6 +737,14 @@ export async function runGoogleHealthCollection(
|
||||
});
|
||||
}
|
||||
}
|
||||
const edgeTo = finished ? null : (resume.cursors.get(key)?.at ?? null);
|
||||
resume.collections.push({
|
||||
key,
|
||||
status: finished ? "done" : failedHard ? "failed" : "stopped",
|
||||
pages: googleHealthSlicePagesRead() - pagesBefore,
|
||||
durationMs: Math.round(performance.now() - startedAt),
|
||||
...(edgeFrom !== null || edgeTo !== null ? { edgeFrom, edgeTo } : {}),
|
||||
});
|
||||
return imported;
|
||||
}
|
||||
|
||||
|
||||
@@ -16,7 +16,9 @@
|
||||
* the hourly poll skips the account for this tick, the manual trigger answers
|
||||
* that a sync is already running, and the backfill and the repair wait a
|
||||
* bounded time for a short run to finish before handing their job back to
|
||||
* pg-boss to retry.
|
||||
* pg-boss to retry. The hourly poll waits too, for about one backfill slice:
|
||||
* a backfill lets go of the account between its slices, so the hourly sync
|
||||
* gets its turn instead of being skipped for as long as the import runs.
|
||||
*/
|
||||
import { withJobLock, type GuardedRun } from "@/lib/jobs/job-lock";
|
||||
|
||||
@@ -32,6 +34,16 @@ export function googleHealthSyncLockKey(userId: string): string {
|
||||
*/
|
||||
export const GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS = 10 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* How long the hourly poll waits for an account another run holds. A backfill
|
||||
* slice walks for five minutes and then lets go
|
||||
* (`GOOGLE_HEALTH_BACKFILL_SLICE_MS`); the page in flight, a rate-limit wait
|
||||
* and the slice's own bookkeeping can add a minute or two, and the next slice
|
||||
* starts a minute after this one ends. Eight minutes covers that, and leaves
|
||||
* the hourly job well inside its fifteen-minute expiry.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_POLL_LOCK_WAIT_MS = 8 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* Run `run` while holding the account's Google Health lock. Resolves
|
||||
* `{ ran: false }` without calling `run` when another run holds it and does
|
||||
|
||||
@@ -27,12 +27,14 @@
|
||||
*/
|
||||
import {
|
||||
GOOGLE_HEALTH_DATA_TYPES,
|
||||
GOOGLE_HEALTH_DEFAULT_MAX_PAGES,
|
||||
GOOGLE_HEALTH_DENSE_MAX_PAGES,
|
||||
type GoogleHealthDataType,
|
||||
type GoogleHealthMappedMeasurement,
|
||||
type GoogleHealthTokenSource,
|
||||
fetchDataPoints,
|
||||
forEachDataPointPage,
|
||||
googleHealthSampleInstant,
|
||||
mapBloodGlucose,
|
||||
mapBodyFat,
|
||||
mapCoreBodyTemperature,
|
||||
@@ -44,11 +46,15 @@ import {
|
||||
mapRestingHeartRate,
|
||||
mapWeight,
|
||||
mapWristTemperature,
|
||||
noteGoogleHealthDeferred,
|
||||
} from "./client";
|
||||
import {
|
||||
getValidToken,
|
||||
googleHealthBackfillCursor,
|
||||
googleHealthHardFailureCount,
|
||||
googleHealthTokenSource,
|
||||
handleCollectionFetchError,
|
||||
recordGoogleHealthBackfillPage,
|
||||
runGoogleHealthCollection,
|
||||
upsertGoogleHealthMeasurements,
|
||||
type GoogleHealthMeasurementUpsert,
|
||||
@@ -108,13 +114,6 @@ const METRIC_RESOURCES: MetricResource[] = [
|
||||
map: mapRespiratoryRate,
|
||||
verb: "fetchRespiratoryRate",
|
||||
},
|
||||
{
|
||||
dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate,
|
||||
map: mapHeartRate,
|
||||
verb: "fetchHeartRate",
|
||||
maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES,
|
||||
intraday: true,
|
||||
},
|
||||
{
|
||||
dataType: GOOGLE_HEALTH_DATA_TYPES.bloodGlucose,
|
||||
map: mapBloodGlucose,
|
||||
@@ -133,8 +132,34 @@ const METRIC_RESOURCES: MetricResource[] = [
|
||||
map: mapWristTemperature,
|
||||
verb: "fetchSleepTemperature",
|
||||
},
|
||||
// Last: the dense stream. A backfill finishes every light collection in
|
||||
// its first slice and then works through this one slice by slice; ahead
|
||||
// of them it held them back until the whole stream was in.
|
||||
{
|
||||
dataType: GOOGLE_HEALTH_DATA_TYPES.heartRate,
|
||||
map: mapHeartRate,
|
||||
verb: "fetchHeartRate",
|
||||
maxPages: GOOGLE_HEALTH_DENSE_MAX_PAGES,
|
||||
intraday: true,
|
||||
},
|
||||
];
|
||||
|
||||
/**
|
||||
* Whether a metric is walked resumably by a backfill: written page by page,
|
||||
* and a spot reading whose sample time can mark how far the walk got.
|
||||
*/
|
||||
function isResumable(resource: MetricResource): boolean {
|
||||
return resource.dataType.timeField === "sample";
|
||||
}
|
||||
|
||||
/**
|
||||
* The page ceilings of every resumable collection, summed: the most pages a
|
||||
* backfill can ever spend on them across all of its attempts.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING = METRIC_RESOURCES.filter(
|
||||
isResumable,
|
||||
).reduce((sum, r) => sum + (r.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES), 0);
|
||||
|
||||
export async function syncUserMetrics(
|
||||
userId: string,
|
||||
opts: GoogleHealthResourceSyncOptions = {},
|
||||
@@ -245,6 +270,27 @@ async function syncMetricCollection(
|
||||
opts: GoogleHealthResourceSyncOptions,
|
||||
): Promise<number> {
|
||||
let imported = 0;
|
||||
// Inside a backfill a spot-reading collection keeps its place: it starts
|
||||
// from the edge an earlier attempt reached and records its new edge after
|
||||
// every page it writes (see `GoogleHealthBackfillCursor`). Its page ceiling
|
||||
// then counts across attempts, so a cursor that never ends is still bounded.
|
||||
const cursor = isResumable(resource)
|
||||
? googleHealthBackfillCursor(resource.verb)
|
||||
: null;
|
||||
const maxPages =
|
||||
(resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES) -
|
||||
(cursor?.pagesSoFar ?? 0);
|
||||
if (cursor && maxPages <= 0) {
|
||||
getEvent()?.addWarning(
|
||||
`google-health ${resource.verb}: backfill reached its page ceiling of ${resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES} pages for ${userId}`,
|
||||
);
|
||||
noteGoogleHealthDeferred();
|
||||
return 0;
|
||||
}
|
||||
// Once one page's write fails, the edge stays where it is for the rest of
|
||||
// the walk: moving it past the failed page would leave that page's rows
|
||||
// behind it for good.
|
||||
let edgeHeld = false;
|
||||
try {
|
||||
await forEachDataPointPage(
|
||||
resource.dataType,
|
||||
@@ -254,7 +300,8 @@ async function syncMetricCollection(
|
||||
start: resource.intraday
|
||||
? (opts.intradayStart ?? opts.start)
|
||||
: opts.start,
|
||||
maxPages: resource.maxPages,
|
||||
maxPages,
|
||||
...(cursor ? { resumable: true, edge: cursor.edge } : {}),
|
||||
},
|
||||
async (points) => {
|
||||
const readings: GoogleHealthMeasurementUpsert[] = [];
|
||||
@@ -269,11 +316,23 @@ async function syncMetricCollection(
|
||||
});
|
||||
}
|
||||
}
|
||||
const failuresBefore = googleHealthHardFailureCount();
|
||||
imported += (
|
||||
await upsertGoogleHealthMeasurements(userId, readings, {
|
||||
deferRollup: opts.deferRollup,
|
||||
})
|
||||
).imported;
|
||||
if (!cursor || points.length === 0) return;
|
||||
if (googleHealthHardFailureCount() !== failuresBefore) edgeHeld = true;
|
||||
if (edgeHeld) return;
|
||||
await recordGoogleHealthBackfillPage(resource.verb, {
|
||||
first: googleHealthSampleInstant(points[0]!, resource.dataType, tz),
|
||||
last: googleHealthSampleInstant(
|
||||
points[points.length - 1]!,
|
||||
resource.dataType,
|
||||
tz,
|
||||
),
|
||||
});
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
|
||||
@@ -31,7 +31,13 @@ const {
|
||||
bossSend: vi.fn(),
|
||||
// Collections the mocked walk "finishes" in this attempt, added to the
|
||||
// resume set the way `runGoogleHealthCollection` adds them.
|
||||
resumeState: { finishes: [] as string[], doneBefore: [] as string[] },
|
||||
resumeState: {
|
||||
finishes: [] as string[],
|
||||
doneBefore: [] as string[],
|
||||
/** Collections whose resume edge the mocked walk moves. */
|
||||
advances: [] as string[],
|
||||
collections: [] as Array<{ key: string; status: string }>,
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db", () => ({ prisma: prismaMock }));
|
||||
@@ -54,6 +60,7 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
vi.mock("@/lib/google-health/sync-core", () => ({
|
||||
GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
|
||||
readGoogleHealthBackfillDone: () => new Set<string>(resumeState.doneBefore),
|
||||
readGoogleHealthBackfillCursors: () => new Map(),
|
||||
runWithGoogleHealthBackfillResume: async (
|
||||
_userId: string,
|
||||
done: Set<string>,
|
||||
@@ -61,7 +68,12 @@ vi.mock("@/lib/google-health/sync-core", () => ({
|
||||
) => {
|
||||
const result = await fn();
|
||||
for (const key of resumeState.finishes) done.add(key);
|
||||
return { result, skipped: [] };
|
||||
return {
|
||||
result,
|
||||
skipped: [],
|
||||
advanced: new Map(resumeState.advances.map((key) => [key, null])),
|
||||
collections: resumeState.collections,
|
||||
};
|
||||
},
|
||||
}));
|
||||
vi.mock("@/lib/google-health/sync", () => ({
|
||||
@@ -71,8 +83,11 @@ vi.mock("@/lib/google-health/sync", () => ({
|
||||
|
||||
import {
|
||||
GOOGLE_HEALTH_BACKFILL_MAX_SLICE,
|
||||
GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS,
|
||||
runGoogleHealthBackfillForUser,
|
||||
} from "../google-health-backfill";
|
||||
import { GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING } from "@/lib/google-health/sync-metrics";
|
||||
import { INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS } from "@/lib/jobs/integration-backfill-admission";
|
||||
import {
|
||||
GOOGLE_HEALTH_DATA_TYPES,
|
||||
forEachDataPointPage,
|
||||
@@ -87,6 +102,8 @@ beforeEach(() => {
|
||||
bossSend.mockResolvedValue("job-next");
|
||||
resumeState.finishes = [];
|
||||
resumeState.doneBefore = [];
|
||||
resumeState.advances = [];
|
||||
resumeState.collections = [];
|
||||
});
|
||||
|
||||
describe("runGoogleHealthBackfillForUser — verdict-gated marker", () => {
|
||||
@@ -261,8 +278,71 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("caps the chain at the number of collections plus two", () => {
|
||||
expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(17 + 2);
|
||||
it("derives the chain's ceiling from the work a backfill can do", () => {
|
||||
// Every collection once, plus every page the resumable collections may
|
||||
// ever read (heart rate's 10 000 and four others' 1 000), plus slack.
|
||||
expect(GOOGLE_HEALTH_BACKFILL_MAX_SLICE).toBe(
|
||||
17 + GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING + 2,
|
||||
);
|
||||
expect(GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING).toBe(10_000 + 4 * 1000);
|
||||
});
|
||||
|
||||
it("continues a slice that only moved a collection's edge", async () => {
|
||||
resumeState.advances = ["fetchHeartRate"];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 900, failed: true });
|
||||
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser("u1", undefined, 7),
|
||||
).resolves.toEqual({ imported: 900, continued: true });
|
||||
expect(bossSend).toHaveBeenCalledTimes(1);
|
||||
expect(bossSend.mock.calls[0]![1]).toMatchObject({
|
||||
data: { userId: "u1", slice: 8 },
|
||||
});
|
||||
expect(bossSend.mock.calls[0]![2]).toMatchObject({
|
||||
startAfter: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
});
|
||||
});
|
||||
|
||||
it("waits longer before the next slice when this one ended on a rate limit", async () => {
|
||||
resumeState.advances = ["fetchHeartRate"];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 10, failed: true });
|
||||
// The slice ended the way a resumable walk ends it when Google keeps
|
||||
// answering 429 past every wait.
|
||||
const client = await import("@/lib/google-health/client");
|
||||
const spy = vi
|
||||
.spyOn(client, "googleHealthSliceBudget")
|
||||
.mockImplementation(() => ({
|
||||
ms: 300_000,
|
||||
deadline: null,
|
||||
maxPages: 500,
|
||||
pages: 3,
|
||||
stopReason: "rate_limited",
|
||||
now: Date.now,
|
||||
}));
|
||||
try {
|
||||
await runGoogleHealthBackfillForUser("u1");
|
||||
} finally {
|
||||
spy.mockRestore();
|
||||
}
|
||||
expect(bossSend.mock.calls[0]![2]).toMatchObject({
|
||||
startAfter: GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS,
|
||||
});
|
||||
});
|
||||
|
||||
it("names where it is stuck when a slice moved nothing", async () => {
|
||||
resumeState.collections = [
|
||||
{ key: "fetchWeight", status: "done" },
|
||||
{ key: "fetchHeartRate", status: "failed" },
|
||||
{ key: "fetchBloodGlucose", status: "deferred" },
|
||||
];
|
||||
syncUserGoogleHealthMock.mockResolvedValue({ imported: 0, failed: true });
|
||||
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser("u1", undefined, 3),
|
||||
).rejects.toThrow(
|
||||
/marker not stamped \(no progress in slice 3; fetchHeartRate:failed,fetchBloodGlucose:deferred\)/,
|
||||
);
|
||||
expect(bossSend).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("stamps the marker on the slice that finishes the history", async () => {
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
/**
|
||||
* #1220 — every admitted full-history import runs inside its own background
|
||||
* event, so what the provider walk records on the ambient event (each
|
||||
* request with its status, every warning, every annotation) reaches the log.
|
||||
* Before, the drain ran the import with no event in scope, every one of those
|
||||
* calls was a no-op, and a failing backfill left nothing but its error
|
||||
* message behind.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Job } from "pg-boss";
|
||||
|
||||
import type { WideEvent } from "@/lib/logging/types";
|
||||
|
||||
const { emitted, runGoogleHealthBackfillMock } = vi.hoisted(() => ({
|
||||
emitted: [] as WideEvent[],
|
||||
runGoogleHealthBackfillMock: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/logging/transports", () => ({
|
||||
emitIfSampled: (event: WideEvent) => emitted.push(event),
|
||||
}));
|
||||
vi.mock("@/lib/jobs/integration-backfill-lock", () => ({
|
||||
withIntegrationBackfillLock: async (_id: string, run: () => unknown) => ({
|
||||
ran: true,
|
||||
result: await run(),
|
||||
}),
|
||||
}));
|
||||
vi.mock("@/lib/jobs/google-health-backfill", () => ({
|
||||
runGoogleHealthBackfillForUser: runGoogleHealthBackfillMock,
|
||||
}));
|
||||
vi.mock("@/lib/jobs/reminder/shared", () => ({ workerLog: () => {} }));
|
||||
|
||||
import { drainIntegrationBackfillAdmission } from "../integration-backfill-drain";
|
||||
import type { IntegrationBackfillAdmissionPayload } from "../integration-backfill-admission";
|
||||
import { annotate, getEvent } from "@/lib/logging/context";
|
||||
|
||||
function job(
|
||||
data: IntegrationBackfillAdmissionPayload,
|
||||
): Job<IntegrationBackfillAdmissionPayload> {
|
||||
return {
|
||||
id: "job-1",
|
||||
data,
|
||||
expireInSeconds: 3600,
|
||||
} as unknown as Job<IntegrationBackfillAdmissionPayload>;
|
||||
}
|
||||
|
||||
const payload: IntegrationBackfillAdmissionPayload = {
|
||||
kind: "google-health-backfill",
|
||||
data: { userId: "u1", enqueuedAt: "2026-10-14T11:00:00Z", slice: 2 },
|
||||
};
|
||||
|
||||
beforeEach(() => {
|
||||
emitted.length = 0;
|
||||
runGoogleHealthBackfillMock.mockReset();
|
||||
});
|
||||
|
||||
describe("drainIntegrationBackfillAdmission — one event per admitted run", () => {
|
||||
it("emits what the import recorded, with the job it belongs to", async () => {
|
||||
runGoogleHealthBackfillMock.mockImplementation(async () => {
|
||||
getEvent()?.addExternalCall({
|
||||
service: "google-health",
|
||||
method: "fetchHeartRate(page=0)",
|
||||
duration_ms: 120,
|
||||
status: 200,
|
||||
});
|
||||
annotate({ meta: { "googleHealth.backfill.slice": { pages: 1 } } });
|
||||
return { imported: 1000, continued: true };
|
||||
});
|
||||
|
||||
await drainIntegrationBackfillAdmission([job(payload)]);
|
||||
|
||||
expect(emitted).toHaveLength(1);
|
||||
const [event] = emitted;
|
||||
expect(event!.background).toMatchObject({
|
||||
task_name: "job.integration_backfill",
|
||||
result: { kind: "google-health-backfill", imported: 1000, continued: 1 },
|
||||
});
|
||||
expect(event!.meta).toMatchObject({
|
||||
kind: "google-health-backfill",
|
||||
identity: "google-health-backfill|u1",
|
||||
job_id: "job-1",
|
||||
slice: 2,
|
||||
"googleHealth.backfill.slice": { pages: 1 },
|
||||
});
|
||||
expect(event!.external_calls).toEqual([
|
||||
expect.objectContaining({
|
||||
method: "fetchHeartRate(page=0)",
|
||||
status: 200,
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("emits the calls of a run that throws, with its error", async () => {
|
||||
runGoogleHealthBackfillMock.mockImplementation(async () => {
|
||||
getEvent()?.addExternalCall({
|
||||
service: "google-health",
|
||||
method: "fetchHeartRate(page=812)",
|
||||
duration_ms: 15000,
|
||||
status: 503,
|
||||
error: "transient HTTP 503",
|
||||
});
|
||||
throw new Error("google-health backfill incomplete for user u1");
|
||||
});
|
||||
|
||||
await expect(
|
||||
drainIntegrationBackfillAdmission([job(payload)]),
|
||||
).rejects.toThrow(/incomplete/);
|
||||
|
||||
expect(emitted).toHaveLength(1);
|
||||
expect(emitted[0]!.level).toBe("error");
|
||||
expect(emitted[0]!.error?.message).toMatch(/incomplete/);
|
||||
expect(emitted[0]!.external_calls?.[0]).toMatchObject({
|
||||
method: "fetchHeartRate(page=812)",
|
||||
status: 503,
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
integrationBackfillSourceOptions,
|
||||
} from "@/lib/jobs/integration-backfill-admission";
|
||||
import {
|
||||
readGoogleHealthBackfillCursors,
|
||||
readGoogleHealthBackfillDone,
|
||||
runWithGoogleHealthBackfillResume,
|
||||
} from "@/lib/google-health/sync-core";
|
||||
@@ -29,7 +30,12 @@ import {
|
||||
syncUserGoogleHealth,
|
||||
} from "@/lib/google-health/sync";
|
||||
import { GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock";
|
||||
import { runWithGoogleHealthStop } from "@/lib/google-health/client";
|
||||
import {
|
||||
googleHealthSliceBudget,
|
||||
runWithGoogleHealthStop,
|
||||
} from "@/lib/google-health/client";
|
||||
import { GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING } from "@/lib/google-health/sync-metrics";
|
||||
import { reportJobProgress } from "@/lib/jobs/job-observer";
|
||||
import { GOOGLE_HEALTH_COLLECTION_KEYS } from "@/lib/google-health/collection-keys";
|
||||
|
||||
export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
|
||||
@@ -42,15 +48,52 @@ export const GOOGLE_HEALTH_BACKFILL_QUEUE = "google-health-backfill";
|
||||
export const GOOGLE_HEALTH_BACKFILL_CONCURRENCY = 1;
|
||||
|
||||
/**
|
||||
* The highest slice a backfill chain may reach. Every continued slice has
|
||||
* finished at least one more collection than the one before it, so a healthy
|
||||
* chain is at most one slice per collection long; the two extra slices are
|
||||
* slack for the first one and the one that stamps the marker. A slice past
|
||||
* this is a chain that counts progress it does not make, and it fails rather
|
||||
* than queueing another.
|
||||
* One slice of a backfill: how long it walks and how many pages it reads
|
||||
* before it lets go of the account and hands the rest to the next slice.
|
||||
*
|
||||
* Far inside the admission job's expiry (sixteen hours) on purpose. A slice
|
||||
* holds the account's sync lock while it walks, and the hourly sync of that
|
||||
* account waits for the lock (`GOOGLE_HEALTH_POLL_LOCK_WAIT_MS`), so a slice
|
||||
* has to be short enough for the hourly sync to get its turn: before slices,
|
||||
* one attempt held the lock for as long as the import ran and every hourly
|
||||
* sync of the account in that time was skipped. The page budget keeps a
|
||||
* slice on a fast link from spending a whole quota window in one go.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_BACKFILL_SLICE_MS = 5 * 60 * 1000;
|
||||
export const GOOGLE_HEALTH_BACKFILL_SLICE_PAGES = 500;
|
||||
|
||||
/**
|
||||
* The pause before the next slice when this one ended on a rate limit it
|
||||
* could not wait out: long enough for Google's per-minute window to refill
|
||||
* several times over, so the next slice does not open on the same 429.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS = 5 * 60;
|
||||
|
||||
/**
|
||||
* The highest slice a backfill chain may reach, derived from the work a
|
||||
* backfill can do rather than guessed.
|
||||
*
|
||||
* A slice is continued only when it made progress that is on disk: it
|
||||
* finished a collection, or it moved a resumable collection's edge after
|
||||
* writing at least one page. Collections finish once each, and a resumable
|
||||
* collection's pages are capped across all attempts by its page ceiling
|
||||
* (`GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING`, the sum of them). So no chain of
|
||||
* progressing slices is longer than the collections plus those pages; the two
|
||||
* extra are slack for the first slice and the one that stamps the marker. In
|
||||
* practice a chain is a handful of slices per year of dense history. A slice
|
||||
* past this is counting progress it does not make, and fails rather than
|
||||
* queueing another.
|
||||
*/
|
||||
export const GOOGLE_HEALTH_BACKFILL_MAX_SLICE =
|
||||
GOOGLE_HEALTH_COLLECTION_KEYS.length + 2;
|
||||
GOOGLE_HEALTH_COLLECTION_KEYS.length +
|
||||
GOOGLE_HEALTH_RESUMABLE_PAGE_CEILING +
|
||||
2;
|
||||
|
||||
/** Test seam: a slice's budget, when not the production one. */
|
||||
export interface GoogleHealthBackfillSliceOptions {
|
||||
sliceMs?: number;
|
||||
slicePages?: number;
|
||||
}
|
||||
|
||||
export interface GoogleHealthBackfillPayload {
|
||||
userId: string;
|
||||
@@ -58,42 +101,40 @@ export interface GoogleHealthBackfillPayload {
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-user backfill handler. Runs a full-history sync for one account and
|
||||
* stamps `backfillCompletedAt` ONLY on a clean run, so the discovery query
|
||||
* drops it. Idempotent: the per-resource upserts are key-stable, so a re-run
|
||||
* (e.g. a reboot mid-walk) overwrites rather than duplicating. Mirrors
|
||||
* `runFitbitBackfillForUser`.
|
||||
* Per-user backfill handler. Runs one slice of a full-history sync for one
|
||||
* account and stamps `backfillCompletedAt` ONLY on a clean run, so the
|
||||
* discovery query drops it. Idempotent: the per-resource upserts are
|
||||
* key-stable, so a re-run (e.g. a reboot mid-walk) overwrites rather than
|
||||
* duplicating. Mirrors `runFitbitBackfillForUser`.
|
||||
*
|
||||
* Verdict-gated: a partial hard failure (one collection 400ing, a write
|
||||
* failing) THROWS so pg-boss retries the job — stamping the marker over a
|
||||
* partial walk would freeze the gap in forever. A connection parked at
|
||||
* `error_reauth` is NOT clean either, but throwing against a dead grant would
|
||||
* just burn the retry budget on the same no-op: return WITHOUT stamping — the
|
||||
* boot discovery re-enqueues the account on the next boot, and the stamp lands
|
||||
* once the user reconnects and a clean walk completes.
|
||||
* failing) is not complete, and stamping the marker over a partial walk would
|
||||
* freeze the gap in forever. A connection parked at `error_reauth` is NOT
|
||||
* clean either, but throwing against a dead grant would just burn the retry
|
||||
* budget on the same no-op: return WITHOUT stamping — the boot discovery
|
||||
* re-enqueues the account on the next boot, and the stamp lands once the user
|
||||
* reconnects and a clean walk completes.
|
||||
*
|
||||
* `shouldStop` is the admission job's budget (`jobBudget`). A per-minute
|
||||
* heart-rate history runs to thousands of pages, and the walk asks it before
|
||||
* every page: once it says stop, the walk ends where it is, the cycle reports
|
||||
* itself truncated, and this throws like any other incomplete run, without
|
||||
* the stamp. The rows already written stay.
|
||||
* Resumable at two grains (`GoogleHealthBackfillCursor`): every collection the
|
||||
* walk finishes cleanly is noted in `backfillProgress.done` and skipped from
|
||||
* then on, and every spot-reading collection notes the edge it has written up
|
||||
* to after each page, so the next attempt starts there instead of at the
|
||||
* newest page. The note is cleared once the stamp lands, and on every
|
||||
* reconnect.
|
||||
*
|
||||
* Resumable: every collection the walk finishes cleanly is noted in
|
||||
* `backfillProgress`, and the next attempt (the pg-boss retry, or the next
|
||||
* boot's discovery) skips those and walks only what is left
|
||||
* (`runWithGoogleHealthBackfillResume`). The note is cleared once the stamp
|
||||
* lands, and on every reconnect.
|
||||
* Runs in slices. A slice walks until its own budget is spent
|
||||
* (`GOOGLE_HEALTH_BACKFILL_SLICE_MS` / `_PAGES`), Google throttles it past
|
||||
* what a wait covers, or the job's budget (`shouldStop`) runs out, and then
|
||||
* lets go of the account. A slice that made progress on disk (a collection
|
||||
* finished, an edge moved) is not a failure: it sends the next slice
|
||||
* (`slice + 1`) and returns `continued`, without the error line or the
|
||||
* queue's failure count. Only a slice that moved nothing throws, with what
|
||||
* each collection came to in the message, because that one is stuck: the
|
||||
* pg-boss retry and, after it, the next boot take it from there. The chain is
|
||||
* capped at `GOOGLE_HEALTH_BACKFILL_MAX_SLICE`.
|
||||
*
|
||||
* Runs in slices: an attempt that ends before the whole history is in but
|
||||
* finished at least one more collection than it started with made progress.
|
||||
* It is not a failure: it sends the next slice (`slice + 1`) and returns
|
||||
* `continued`, without the error line or the queue's failure count. Only an
|
||||
* attempt that finished nothing new throws, because that one is stuck: the
|
||||
* pg-boss retry and, after it, the next boot take it from there. Progress is
|
||||
* counted only once it is persisted (`runGoogleHealthCollection` adds a
|
||||
* collection to `done` after its note is written), and the chain is capped at
|
||||
* `GOOGLE_HEALTH_BACKFILL_MAX_SLICE`: a slice past it throws instead of
|
||||
* running or sending another.
|
||||
* What the slice did is annotated on the run's event as
|
||||
* `googleHealth.backfill.slice` (see `docs/integrations/google-health.md`).
|
||||
*
|
||||
* Never beside another run of the same account: the sync takes the account's
|
||||
* lock, waiting a bounded time for an hourly or manual run to finish. If the
|
||||
@@ -104,6 +145,7 @@ export async function runGoogleHealthBackfillForUser(
|
||||
userId: string,
|
||||
shouldStop: () => boolean = () => false,
|
||||
slice: number = 0,
|
||||
sliceOptions: GoogleHealthBackfillSliceOptions = {},
|
||||
): Promise<{ imported: number; continued?: boolean }> {
|
||||
if (slice > GOOGLE_HEALTH_BACKFILL_MAX_SLICE) {
|
||||
annotate({
|
||||
@@ -142,22 +184,56 @@ export async function runGoogleHealthBackfillForUser(
|
||||
return { imported: 0 };
|
||||
}
|
||||
const done = readGoogleHealthBackfillDone(connection?.backfillProgress);
|
||||
const cursors = readGoogleHealthBackfillCursors(connection?.backfillProgress);
|
||||
const doneBefore = done.size;
|
||||
|
||||
const budget = googleHealthSliceBudget(
|
||||
sliceOptions.sliceMs ?? GOOGLE_HEALTH_BACKFILL_SLICE_MS,
|
||||
sliceOptions.slicePages ?? GOOGLE_HEALTH_BACKFILL_SLICE_PAGES,
|
||||
);
|
||||
const startedAt = Date.now();
|
||||
const {
|
||||
result: { imported, failed, busy },
|
||||
skipped,
|
||||
} = await runWithGoogleHealthBackfillResume(userId, done, () =>
|
||||
runWithGoogleHealthStop(shouldStop, () =>
|
||||
syncUserGoogleHealth(userId, {
|
||||
fullSync: true,
|
||||
waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS,
|
||||
}),
|
||||
),
|
||||
advanced,
|
||||
collections,
|
||||
} = await runWithGoogleHealthBackfillResume(
|
||||
userId,
|
||||
done,
|
||||
() =>
|
||||
runWithGoogleHealthStop(
|
||||
shouldStop,
|
||||
() =>
|
||||
syncUserGoogleHealth(userId, {
|
||||
fullSync: true,
|
||||
waitForLockMs: GOOGLE_HEALTH_SYNC_LOCK_WAIT_MS,
|
||||
}),
|
||||
budget,
|
||||
),
|
||||
cursors,
|
||||
);
|
||||
if (skipped.length > 0) {
|
||||
annotate({ meta: { "googleHealth.backfill.resumedPast": skipped } });
|
||||
}
|
||||
|
||||
const finished = done.size - doneBefore;
|
||||
const sliceReport = {
|
||||
slice,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
pages: budget.pages,
|
||||
stopReason: budget.stopReason,
|
||||
imported,
|
||||
collectionsDone: done.size,
|
||||
collectionsTotal: GOOGLE_HEALTH_COLLECTION_KEYS.length,
|
||||
collectionsFinished: finished,
|
||||
resumedPast: skipped.length,
|
||||
collections,
|
||||
// Where each unfinished resumable collection has got to, after this slice.
|
||||
edges: Object.fromEntries([...cursors].map(([key, c]) => [key, c.at])),
|
||||
};
|
||||
annotate({ meta: { "googleHealth.backfill.slice": sliceReport } });
|
||||
reportJobProgress({
|
||||
google_health_backfill_slice: slice,
|
||||
google_health_backfill_pages: budget.pages,
|
||||
google_health_backfill_collections_done: done.size,
|
||||
});
|
||||
|
||||
if (busy) {
|
||||
throw new Error(
|
||||
@@ -166,10 +242,10 @@ export async function runGoogleHealthBackfillForUser(
|
||||
}
|
||||
|
||||
if (failed) {
|
||||
// `done` is the resume set this attempt added to as it finished
|
||||
// collections; a larger set than it started with is progress.
|
||||
const finished = done.size - doneBefore;
|
||||
if (finished > 0 && slice + 1 <= GOOGLE_HEALTH_BACKFILL_MAX_SLICE) {
|
||||
// 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(
|
||||
@@ -182,7 +258,9 @@ export async function runGoogleHealthBackfillForUser(
|
||||
slice: slice + 1,
|
||||
},
|
||||
},
|
||||
INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
budget.stopReason === "rate_limited"
|
||||
? GOOGLE_HEALTH_BACKFILL_THROTTLED_DELAY_SECONDS
|
||||
: INTEGRATION_BACKFILL_CONTINUATION_DELAY_SECONDS,
|
||||
)
|
||||
: null;
|
||||
if (sent) {
|
||||
@@ -192,8 +270,10 @@ export async function runGoogleHealthBackfillForUser(
|
||||
details: {
|
||||
imported,
|
||||
slice,
|
||||
stopReason: budget.stopReason,
|
||||
collectionsFinished: finished,
|
||||
collectionsDone: done.size,
|
||||
edgesMoved: [...advanced.keys()],
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -202,12 +282,23 @@ export async function runGoogleHealthBackfillForUser(
|
||||
// No follow-up could be queued; fall through to the retry path so the
|
||||
// work is not dropped.
|
||||
}
|
||||
// No collection finished in this attempt: the backfill is stuck, not
|
||||
// slow. Surface through the pg-boss retry path (retryLimit 3, backoff).
|
||||
// If the failure parked the connection at error_reauth meanwhile, the
|
||||
// next attempt takes the reauth return above instead of failing again.
|
||||
// Nothing moved in this slice: the backfill is stuck, not slow. Surface
|
||||
// through the pg-boss retry path (retryLimit 3, backoff) with what each
|
||||
// collection came to, so the log line says where it is stuck. If the
|
||||
// failure parked the connection at error_reauth meanwhile, the next
|
||||
// attempt takes the reauth return above instead of failing again.
|
||||
annotate({
|
||||
action: {
|
||||
name: "google_health.backfill.stuck",
|
||||
details: { slice, stopReason: budget.stopReason },
|
||||
},
|
||||
});
|
||||
const where = collections
|
||||
.filter((c) => c.status !== "done")
|
||||
.map((c) => `${c.key}:${c.status}`)
|
||||
.join(",");
|
||||
throw new Error(
|
||||
`google-health backfill incomplete for user ${userId} — marker not stamped`,
|
||||
`google-health backfill incomplete for user ${userId} — marker not stamped (no progress in slice ${slice}${budget.stopReason ? `, stopped on ${budget.stopReason}` : ""}${where ? `; ${where}` : ""})`,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -219,7 +310,7 @@ export async function runGoogleHealthBackfillForUser(
|
||||
annotate({
|
||||
action: {
|
||||
name: "google_health.backfill.complete",
|
||||
details: { imported },
|
||||
details: { imported, slice },
|
||||
},
|
||||
});
|
||||
return { imported };
|
||||
|
||||
@@ -32,6 +32,7 @@ import { invalidateStatusInsightsForTypes } from "@/lib/insights/status-invalida
|
||||
import { jobDone, jobFailed, type JobOutcome } from "@/lib/jobs/job-outcome";
|
||||
import { trackBackgroundTask } from "@/lib/logging/background-tasks";
|
||||
import { annotate } from "@/lib/logging/context";
|
||||
import { withBackgroundEvent } from "@/lib/logging/background";
|
||||
import {
|
||||
recomputeUserRollups,
|
||||
ROLLUP_FOLD_WINDOW_MS,
|
||||
@@ -102,7 +103,18 @@ export async function handleHealthConnectImport(
|
||||
): Promise<JobOutcome> {
|
||||
let failure: JobOutcome | null = null;
|
||||
for (const job of jobs) {
|
||||
const outcome = await runHealthConnectImport(job.data);
|
||||
// Its own event per import, so the `annotate` calls below and anything
|
||||
// the import records on the ambient event reach the log instead of
|
||||
// vanishing for want of one.
|
||||
const outcome = await withBackgroundEvent(
|
||||
"job.health_connect_import",
|
||||
async (event) => {
|
||||
event.addMeta("import_job_id", job.data.importJobId);
|
||||
const result = await runHealthConnectImport(job.data);
|
||||
if (!result.ok) event.setError(result.cause ?? result.reason);
|
||||
return result;
|
||||
},
|
||||
);
|
||||
if (!outcome.ok && failure === null) failure = outcome;
|
||||
}
|
||||
return failure ?? jobDone({ jobs: jobs.length });
|
||||
|
||||
@@ -143,6 +143,49 @@ async function runAdmitted(
|
||||
}
|
||||
}
|
||||
|
||||
/** The task name of the event each admitted run writes. */
|
||||
export const INTEGRATION_BACKFILL_RUN_TASK = "job.integration_backfill";
|
||||
|
||||
/**
|
||||
* Run one admitted job inside its own background event.
|
||||
*
|
||||
* The runners write what they do through the ambient event: every request a
|
||||
* provider walk makes (`addExternalCall`, with its status, duration and the
|
||||
* page it asked for), every warning, every `annotate`. Without an event in
|
||||
* scope each of those is a no-op, so a full-history import that failed left
|
||||
* nothing behind but the job's own error message: not the collection it was
|
||||
* on, not the status the provider answered, not how far it got (#1220). The
|
||||
* hourly syncs always ran inside one; the imports did not. Every kind gets
|
||||
* one here, emitted when the run settles, thrown error included.
|
||||
*/
|
||||
async function runObserved(
|
||||
job: Job<IntegrationBackfillAdmissionPayload>,
|
||||
identity: string,
|
||||
shouldStop: () => boolean,
|
||||
totals: RunCounts,
|
||||
): Promise<void> {
|
||||
await withBackgroundEvent(INTEGRATION_BACKFILL_RUN_TASK, async (event) => {
|
||||
event.addMeta("kind", job.data.kind);
|
||||
event.addMeta("identity", identity);
|
||||
event.addMeta("job_id", job.id);
|
||||
event.addMeta("slice", job.data.data.slice ?? 0);
|
||||
const before = { ...totals };
|
||||
await runAdmitted(job.data, shouldStop, totals);
|
||||
event.setBackground({
|
||||
task_name: INTEGRATION_BACKFILL_RUN_TASK,
|
||||
result: {
|
||||
kind: job.data.kind,
|
||||
imported: totals.imported - before.imported,
|
||||
continued: totals.continued - before.continued,
|
||||
removed: totals.removed - before.removed,
|
||||
deleted: totals.deleted - before.deleted,
|
||||
markers: totals.markers - before.markers,
|
||||
linked: totals.linked - before.linked,
|
||||
},
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Drain a batch of admission jobs. There is no per-job error isolation: a
|
||||
* runner that throws fails the job and the admission retry policy applies. So
|
||||
@@ -166,7 +209,7 @@ export async function drainIntegrationBackfillAdmission(
|
||||
for (const job of jobs) {
|
||||
const identity = integrationBackfillIdentity(job.data);
|
||||
const run = await withIntegrationBackfillLock(identity, () =>
|
||||
runAdmitted(job.data, shouldStop, totals),
|
||||
runObserved(job, identity, shouldStop, totals),
|
||||
);
|
||||
if (!run.ran) {
|
||||
alreadyRunning += 1;
|
||||
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
isGoogleHealthParked,
|
||||
syncUserGoogleHealth,
|
||||
} from "@/lib/google-health/sync";
|
||||
import { GOOGLE_HEALTH_POLL_LOCK_WAIT_MS } from "@/lib/google-health/sync-lock";
|
||||
import { enqueueReminderSatisfy } from "@/lib/jobs/reminder-satisfy";
|
||||
import { cleanupExpiredGoogleHealthOAuthStates } from "@/lib/jobs/google-health-oauth-state-cleanup";
|
||||
import {
|
||||
@@ -69,9 +70,15 @@ export async function handleGoogleHealthSync(
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await syncUserGoogleHealth(userId);
|
||||
// A backfill holds the account for one slice at a time, so
|
||||
// the tick waits for the slice to end rather than skipping the
|
||||
// account for as long as the whole import runs.
|
||||
const result = await syncUserGoogleHealth(userId, {
|
||||
waitForLockMs: GOOGLE_HEALTH_POLL_LOCK_WAIT_MS,
|
||||
});
|
||||
// Another run of this account (its backfill, a manual sync)
|
||||
// holds the account; this tick leaves it to that run.
|
||||
// still held the account after the wait; this tick leaves it
|
||||
// to that run.
|
||||
if (result.busy) return { status: "skipped", imported: 0 };
|
||||
const parked =
|
||||
result.failed &&
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { WideEventBuilder } from "../event-builder";
|
||||
import {
|
||||
EXTERNAL_CALLS_MAX,
|
||||
EXTERNAL_CALLS_PLAIN_MAX,
|
||||
WideEventBuilder,
|
||||
} from "../event-builder";
|
||||
|
||||
describe("WideEventBuilder.setHttp", () => {
|
||||
it("stores plain http metadata as-is", () => {
|
||||
@@ -140,3 +144,46 @@ describe("WideEventBuilder.capStatusLevel", () => {
|
||||
expect(evt.toJSON().level).toBe("warn");
|
||||
});
|
||||
});
|
||||
|
||||
describe("WideEventBuilder — bounded external calls", () => {
|
||||
it("lists failures beyond the share of plain successes, and counts the rest", () => {
|
||||
const evt = new WideEventBuilder("background");
|
||||
for (let i = 0; i < 500; i++) {
|
||||
evt.addExternalCall({
|
||||
service: "google-health",
|
||||
method: `page=${i}`,
|
||||
duration_ms: 1,
|
||||
status: 200,
|
||||
});
|
||||
}
|
||||
evt.addExternalCall({
|
||||
service: "google-health",
|
||||
method: "page=500",
|
||||
duration_ms: 1,
|
||||
status: 503,
|
||||
error: "transient HTTP 503",
|
||||
});
|
||||
const event = evt.toJSON();
|
||||
expect(event.external_calls).toHaveLength(EXTERNAL_CALLS_PLAIN_MAX + 1);
|
||||
expect(event.external_calls!.at(-1)).toMatchObject({ status: 503 });
|
||||
expect(event.external_calls_omitted).toBe(500 - EXTERNAL_CALLS_PLAIN_MAX);
|
||||
|
||||
for (let i = 0; i < 1000; i++) {
|
||||
evt.addExternalCall({
|
||||
service: "google-health",
|
||||
method: `retry=${i}`,
|
||||
duration_ms: 1,
|
||||
status: 429,
|
||||
note: "rate_limited_retrying_in_ms=2000",
|
||||
});
|
||||
}
|
||||
const after = evt.toJSON();
|
||||
expect(after.external_calls).toHaveLength(EXTERNAL_CALLS_MAX);
|
||||
expect(after.external_calls_omitted).toBe(
|
||||
500 -
|
||||
EXTERNAL_CALLS_PLAIN_MAX +
|
||||
1000 -
|
||||
(EXTERNAL_CALLS_MAX - EXTERNAL_CALLS_PLAIN_MAX - 1),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -35,6 +35,12 @@ function redactDeep<T>(value: T, depth = 0): T {
|
||||
return value;
|
||||
}
|
||||
|
||||
/** The most external calls one event lists. */
|
||||
export const EXTERNAL_CALLS_MAX = 200;
|
||||
|
||||
/** The most calls without an error or a note one event lists. */
|
||||
export const EXTERNAL_CALLS_PLAIN_MAX = 60;
|
||||
|
||||
/**
|
||||
* Baut ein Wide Event Schritt fuer Schritt auf.
|
||||
* Wird am Anfang eines Requests/Tasks erstellt und am Ende emittiert.
|
||||
@@ -44,6 +50,7 @@ export class WideEventBuilder {
|
||||
private startTime: number;
|
||||
private highestLevel: LogLevel = "info";
|
||||
private statusLevelCeiling: LogLevel = "error";
|
||||
private plainExternalCalls = 0;
|
||||
|
||||
constructor(kind: EventKind = "http") {
|
||||
this.startTime = performance.now();
|
||||
@@ -208,6 +215,23 @@ export class WideEventBuilder {
|
||||
|
||||
addExternalCall(call: NonNullable<WideEvent["external_calls"]>[0]): this {
|
||||
if (!this.event.external_calls) this.event.external_calls = [];
|
||||
// Bounded. A long background run (a full-history backfill walks hundreds
|
||||
// of pages in one event) would otherwise hold every call in memory and
|
||||
// write them all as one log line. Calls that failed or carry a note are
|
||||
// what a reader looks for, so they keep the larger share; plain successes
|
||||
// stop being listed sooner. Whatever is left out is counted, never lost
|
||||
// silently.
|
||||
const calls = this.event.external_calls;
|
||||
const notable = call.error !== undefined || call.note !== undefined;
|
||||
if (
|
||||
calls.length >= EXTERNAL_CALLS_MAX ||
|
||||
(!notable && this.plainExternalCalls >= EXTERNAL_CALLS_PLAIN_MAX)
|
||||
) {
|
||||
this.event.external_calls_omitted =
|
||||
(this.event.external_calls_omitted ?? 0) + 1;
|
||||
return this;
|
||||
}
|
||||
if (!notable) this.plainExternalCalls += 1;
|
||||
// An outbound failure carries whatever the remote said, and for several
|
||||
// integrations that is the request URL — which for Telegram embeds the bot
|
||||
// token. `setError` and `setHttp` scrub on the way in; this entry point did
|
||||
|
||||
@@ -123,6 +123,11 @@ export interface WideEvent {
|
||||
*/
|
||||
note?: string;
|
||||
}>;
|
||||
/**
|
||||
* How many external calls the event did not list because it had already
|
||||
* listed its share (`EXTERNAL_CALLS_MAX`, `EXTERNAL_CALLS_PLAIN_MAX`).
|
||||
*/
|
||||
external_calls_omitted?: number;
|
||||
|
||||
// Hintergrundaufgabe (nur bei kind === "background")
|
||||
background?: {
|
||||
|
||||
@@ -0,0 +1,396 @@
|
||||
/**
|
||||
* #1220 — a Google Health history backfill over three years of heart rate,
|
||||
* driven through the admission drain slice by slice until it completes,
|
||||
* against Postgres.
|
||||
*
|
||||
* What went wrong. A backfill noted its progress only once a whole collection
|
||||
* was walked to the end. Three years of heart rate at one reading a minute is
|
||||
* some 1 600 pages, and an attempt that did not get through all of them (its
|
||||
* budget, a rate limit it could not wait out, one failed request among
|
||||
* thousands) started the next attempt from the newest page again. The account
|
||||
* never got past its first unfinished collection, every attempt failed with
|
||||
* "marker not stamped", and none of them left a line saying where it stopped,
|
||||
* because the drain ran the import outside any log event.
|
||||
*
|
||||
* What this pins, end to end:
|
||||
* - the backfill completes, through as many slices as the history needs,
|
||||
* and every slice after the first starts where the one before it stopped;
|
||||
* - every reading is imported exactly once: the rows match the history, the
|
||||
* slices' import counts add up to it, and no row was ever rewritten;
|
||||
* - a server error, a lost request and a rate limit Google will not lift
|
||||
* for an hour each cost a retry or a slice boundary, never the walk;
|
||||
* - the hourly sync runs between slices (and beside one, by waiting for
|
||||
* it) instead of being skipped for as long as the import runs;
|
||||
* - every slice writes a log event naming its requests and where it got to;
|
||||
* - the heap does not grow with the history (run it under the production
|
||||
* ceiling: `NODE_OPTIONS=--max-old-space-size=524`).
|
||||
*
|
||||
* Google is replaced at the `safeFetch` seam by a generator that builds each
|
||||
* page on request from the filter and page token it is asked with, so the
|
||||
* fixture never sits in memory.
|
||||
*/
|
||||
import v8 from "node:v8";
|
||||
import vm from "node:vm";
|
||||
|
||||
import type { Job } from "pg-boss";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import type { WideEvent } from "@/lib/logging/types";
|
||||
import { getPrismaClient, truncateAllTables } from "./setup";
|
||||
|
||||
const USER = "gh-multi-year-owner";
|
||||
const MINUTE = 60_000;
|
||||
/** The newest reading: now, so the hourly sync's window lands on real data. */
|
||||
const NOW = Math.floor(Date.now() / MINUTE) * MINUTE;
|
||||
/** Three years of heart rate, one reading a minute. */
|
||||
const POINTS = 3 * 365 * 24 * 60;
|
||||
|
||||
/**
|
||||
* How much the collected heap may grow over the second half of the history:
|
||||
* from the first sample taken inside a page request once the second slice is
|
||||
* over, to the most any later sample reaches. Every sample is taken mid-walk,
|
||||
* so each carries the same in-flight working set (one page, the ORM's and the
|
||||
* driver's buffers, some 35 MB here) and only growth is compared. The first
|
||||
* two slices are left out because the first hourly sync and the first
|
||||
* rate-limited slice load code paths once. Measured, not chosen: the line
|
||||
* stays within half a MB of flat. A backfill that kept what it read would
|
||||
* grow by hundreds of MB over the same stretch.
|
||||
*/
|
||||
const GROWTH_BUDGET_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
const fake = vi.hoisted(() => ({
|
||||
now: 0,
|
||||
total: 0,
|
||||
/** Heart-rate requests seen, the backfill's among them, and readings served. */
|
||||
heartRateRequests: 0,
|
||||
backfillRequests: 0,
|
||||
servedPoints: 0,
|
||||
/** Faults to inject, keyed by the backfill request number they hit. */
|
||||
faults: new Map<number, "503" | "timeout" | "429-long">(),
|
||||
onRequest: null as (() => void) | null,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("@/lib/safe-fetch")>();
|
||||
const respond = (body: unknown, status = 200, headers = {}) =>
|
||||
new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { "content-type": "application/json", ...headers },
|
||||
});
|
||||
return {
|
||||
...actual,
|
||||
// A plain function, not `vi.fn`: a mock records every result, and here
|
||||
// each result is a page of a thousand readings.
|
||||
safeFetch: async (url: string) => {
|
||||
const u = new URL(url);
|
||||
if (!u.pathname.endsWith("/dataTypes/heart-rate/dataPoints")) {
|
||||
// Every other collection is empty for this account.
|
||||
return respond({});
|
||||
}
|
||||
fake.heartRateRequests += 1;
|
||||
fake.onRequest?.();
|
||||
// The hourly sync always asks from a lower bound; the backfill never.
|
||||
const filter = u.searchParams.get("filter") ?? "";
|
||||
const fromBackfill = !filter.includes(">=");
|
||||
if (fromBackfill) fake.backfillRequests += 1;
|
||||
const fault = fromBackfill
|
||||
? fake.faults.get(fake.backfillRequests)
|
||||
: undefined;
|
||||
if (fault) {
|
||||
fake.faults.delete(fake.backfillRequests);
|
||||
if (fault === "timeout") {
|
||||
throw new actual.SafeFetchError("safeFetch aborted", "timeout");
|
||||
}
|
||||
if (fault === "503") {
|
||||
return respond({ error: { code: 503, status: "UNAVAILABLE" } }, 503);
|
||||
}
|
||||
return respond(
|
||||
{ error: { code: 429, status: "RESOURCE_EXHAUSTED" } },
|
||||
429,
|
||||
{ "retry-after": "3600" },
|
||||
);
|
||||
}
|
||||
|
||||
// Newest first, one reading a minute: reading `g` is `g` minutes old.
|
||||
// The hourly sync asks from a lower bound (`>=`), a resumed backfill
|
||||
// up to its edge (`<=`).
|
||||
const from = /physical_time >= "([^"]+)"/.exec(filter)?.[1];
|
||||
const upTo = /physical_time <= "([^"]+)"/.exec(filter)?.[1];
|
||||
const newest = upTo
|
||||
? Math.max(0, Math.ceil((fake.now - Date.parse(upTo)) / MINUTE))
|
||||
: 0;
|
||||
const oldest = from
|
||||
? Math.min(
|
||||
fake.total - 1,
|
||||
Math.floor((fake.now - Date.parse(from)) / MINUTE),
|
||||
)
|
||||
: fake.total - 1;
|
||||
const pageSize = Number(u.searchParams.get("pageSize") ?? 1000);
|
||||
const page = Number(u.searchParams.get("pageToken") ?? 0);
|
||||
const start = newest + page * pageSize;
|
||||
const end = Math.min(oldest + 1, start + pageSize);
|
||||
const dataPoints = [];
|
||||
for (let g = start; g < end; g++) {
|
||||
dataPoints.push({
|
||||
name: `users/me/dataTypes/heart-rate/dataPoints/${7_000_000_000 + g}`,
|
||||
dataSource: { platform: "FITBIT" },
|
||||
heartRate: {
|
||||
beatsPerMinute: String(52 + (g % 61)),
|
||||
sampleTime: {
|
||||
physicalTime: new Date(fake.now - g * MINUTE).toISOString(),
|
||||
},
|
||||
},
|
||||
});
|
||||
}
|
||||
fake.servedPoints += dataPoints.length;
|
||||
return respond({
|
||||
dataPoints,
|
||||
...(end <= oldest ? { nextPageToken: String(page + 1) } : {}),
|
||||
});
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
/** Jobs the backfill and the hourly sync queue, recorded. */
|
||||
const sent = vi.hoisted(() => ({
|
||||
jobs: [] as Array<{ queue: string; data: unknown }>,
|
||||
}));
|
||||
vi.mock("@/lib/jobs/boss-instance", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("@/lib/jobs/boss-instance")>()),
|
||||
getGlobalBoss: () => ({
|
||||
send: async (queue: string, data: unknown) => {
|
||||
sent.jobs.push({ queue, data });
|
||||
return `job-${sent.jobs.length}`;
|
||||
},
|
||||
}),
|
||||
}));
|
||||
|
||||
/** The log events the runs write, recorded instead of printed. */
|
||||
const logged = vi.hoisted(() => ({ events: [] as WideEvent[] }));
|
||||
vi.mock("@/lib/logging/transports", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("@/lib/logging/transports")>()),
|
||||
emitIfSampled: (event: WideEvent) => {
|
||||
logged.events.push(event);
|
||||
},
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/db-compat", () => ({
|
||||
ensureDbCompatibility: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
const forceGc = ((): (() => void) => {
|
||||
v8.setFlagsFromString("--expose-gc");
|
||||
const gc = vm.runInNewContext("gc") as () => void;
|
||||
v8.setFlagsFromString("--no-expose-gc");
|
||||
return gc;
|
||||
})();
|
||||
|
||||
function liveHeapBytes(): number {
|
||||
forceGc();
|
||||
forceGc();
|
||||
return process.memoryUsage().heapUsed;
|
||||
}
|
||||
|
||||
const mb = (bytes: number): string => (bytes / 1024 / 1024).toFixed(1);
|
||||
|
||||
function report(line: string): void {
|
||||
process.stderr.write(`[gh-multi-year] ${line}\n`);
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
const { encrypt } = await import("@/lib/crypto");
|
||||
await prisma.user.create({
|
||||
data: { id: USER, username: USER, timezone: "Europe/Berlin" },
|
||||
});
|
||||
await prisma.googleHealthConnection.create({
|
||||
data: {
|
||||
userId: USER,
|
||||
googleUserId: `google-${USER}`,
|
||||
accessToken: encrypt("access"),
|
||||
refreshToken: encrypt("refresh"),
|
||||
tokenExpiresAt: new Date(Date.now() + 7 * 86_400_000),
|
||||
},
|
||||
});
|
||||
fake.now = NOW;
|
||||
fake.total = POINTS;
|
||||
}, 120_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await truncateAllTables(getPrismaClient());
|
||||
});
|
||||
|
||||
type AdmissionPayload =
|
||||
import("@/lib/jobs/integration-backfill-admission").IntegrationBackfillAdmissionPayload;
|
||||
|
||||
function admissionJob(
|
||||
payload: AdmissionPayload,
|
||||
n: number,
|
||||
): Job<AdmissionPayload> {
|
||||
return {
|
||||
id: `admission-${n}`,
|
||||
name: "integration-backfill-admission",
|
||||
data: payload,
|
||||
// The production expiry: sixteen hours.
|
||||
expireInSeconds: 16 * 60 * 60,
|
||||
} as unknown as Job<AdmissionPayload>;
|
||||
}
|
||||
|
||||
function nextSlice(): AdmissionPayload | null {
|
||||
const next = sent.jobs.find(
|
||||
(j) => j.queue === "integration-backfill-admission",
|
||||
);
|
||||
return next ? (next.data as AdmissionPayload) : null;
|
||||
}
|
||||
|
||||
describe("Google Health backfill over three years of heart rate (#1220)", () => {
|
||||
it("completes slice by slice, imports every reading once, and keeps the hourly sync running", async () => {
|
||||
const { drainIntegrationBackfillAdmission } =
|
||||
await import("@/lib/jobs/integration-backfill-drain");
|
||||
const { handleGoogleHealthSync } =
|
||||
await import("@/lib/jobs/reminder/google-health-sync");
|
||||
const { GOOGLE_HEALTH_TRANSIENT_RETRY } =
|
||||
await import("@/lib/google-health/client");
|
||||
const { GOOGLE_HEALTH_BACKFILL_SLICE_PAGES } =
|
||||
await import("@/lib/jobs/google-health-backfill");
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0;
|
||||
|
||||
// A server error and a lost request in the first slice, and a rate limit
|
||||
// Google will not lift for an hour in the second.
|
||||
fake.faults.set(40, "503");
|
||||
fake.faults.set(80, "timeout");
|
||||
fake.faults.set(GOOGLE_HEALTH_BACKFILL_SLICE_PAGES + 120, "429-long");
|
||||
|
||||
// Heap samples from inside the page requests, between two pages of the
|
||||
// work being measured, once the second slice is over (see the budget).
|
||||
let measuring = false;
|
||||
let anchor: number | null = null;
|
||||
let peak = 0;
|
||||
let samples = 0;
|
||||
fake.onRequest = () => {
|
||||
if (!measuring || fake.heartRateRequests % 50 !== 0) return;
|
||||
const live = liveHeapBytes();
|
||||
if (anchor === null) {
|
||||
anchor = peak = live;
|
||||
return;
|
||||
}
|
||||
samples++;
|
||||
peak = Math.max(peak, live);
|
||||
};
|
||||
|
||||
const prisma = getPrismaClient();
|
||||
let payload: AdmissionPayload | null = {
|
||||
kind: "google-health-backfill",
|
||||
data: { userId: USER, enqueuedAt: new Date().toISOString() },
|
||||
};
|
||||
const slices: Array<{
|
||||
slice: number;
|
||||
imported: number;
|
||||
continued: number;
|
||||
}> = [];
|
||||
let hourlyBeside: Promise<unknown> | null = null;
|
||||
|
||||
for (let n = 0; payload && n < 40; n++) {
|
||||
sent.jobs = [];
|
||||
logged.events = [];
|
||||
const drained = drainIntegrationBackfillAdmission([
|
||||
admissionJob(payload, n),
|
||||
]);
|
||||
// Once, start the hourly sync while a slice is running: it waits for
|
||||
// the slice instead of skipping the account.
|
||||
if (n === 2) {
|
||||
hourlyBeside = handleGoogleHealthSync([] as never);
|
||||
}
|
||||
const outcome = await drained;
|
||||
expect(outcome.ok).toBe(true);
|
||||
const did = (outcome as { did: Record<string, number> }).did;
|
||||
slices.push({
|
||||
slice: payload.data.slice ?? 0,
|
||||
imported: did.imported,
|
||||
continued: did.continued,
|
||||
});
|
||||
|
||||
// The slice wrote one log event, with its requests and where it got to.
|
||||
const runEvents = logged.events.filter(
|
||||
(e) => e.background?.task_name === "job.integration_backfill",
|
||||
);
|
||||
expect(runEvents).toHaveLength(1);
|
||||
const sliceMeta = runEvents[0]!.meta?.["googleHealth.backfill.slice"] as
|
||||
{ pages: number; stopReason: string | null; edges: object } | undefined;
|
||||
expect(sliceMeta).toBeDefined();
|
||||
expect(
|
||||
runEvents[0]!.external_calls?.some((c) =>
|
||||
c.method.startsWith("fetchHeartRate(page="),
|
||||
),
|
||||
).toBe(true);
|
||||
report(
|
||||
`slice ${payload.data.slice ?? 0}: pages=${sliceMeta!.pages} ` +
|
||||
`stop=${sliceMeta!.stopReason} imported=${did.imported} ` +
|
||||
`edges=${JSON.stringify(sliceMeta!.edges)}`,
|
||||
);
|
||||
|
||||
if (n === 1) measuring = true;
|
||||
payload = nextSlice();
|
||||
|
||||
// Between slices the account is free, and the hourly sync runs.
|
||||
if (payload) {
|
||||
const hourly = await handleGoogleHealthSync([] as never);
|
||||
expect(hourly).toMatchObject({
|
||||
ok: true,
|
||||
did: { users_skipped: 0, users_failed: 0, users_parked: 0 },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
expect(await hourlyBeside).toMatchObject({
|
||||
ok: true,
|
||||
did: { users_skipped: 0 },
|
||||
});
|
||||
|
||||
// Completed, and the note of how far it got is gone with it.
|
||||
const connection = await prisma.googleHealthConnection.findUniqueOrThrow({
|
||||
where: { userId: USER },
|
||||
select: { backfillCompletedAt: true, backfillProgress: true },
|
||||
});
|
||||
expect(connection.backfillCompletedAt).toBeInstanceOf(Date);
|
||||
expect(connection.backfillProgress).toBeNull();
|
||||
expect(payload).toBeNull();
|
||||
|
||||
// More than one slice, every one before the last continued, and the
|
||||
// rate-limited one among them.
|
||||
expect(slices.length).toBeGreaterThan(2);
|
||||
expect(slices.slice(0, -1).every((s) => s.continued === 1)).toBe(true);
|
||||
expect(fake.faults.size).toBe(0);
|
||||
expect(slices.map((s) => s.slice)).toEqual(slices.map((_, i) => i));
|
||||
|
||||
// Every reading exactly once.
|
||||
const rows = await prisma.measurement.count({
|
||||
where: { userId: USER, type: "PULSE", source: "GOOGLE_HEALTH" },
|
||||
});
|
||||
expect(rows).toBe(POINTS);
|
||||
expect(slices.reduce((sum, s) => sum + s.imported, 0)).toBe(POINTS);
|
||||
const versions = await prisma.measurement.aggregate({
|
||||
where: { userId: USER, type: "PULSE", source: "GOOGLE_HEALTH" },
|
||||
_max: { syncVersion: true },
|
||||
_min: { syncVersion: true },
|
||||
});
|
||||
expect(versions._max.syncVersion).toBe(versions._min.syncVersion);
|
||||
// Each slice re-reads only the reading on its edge (plus the hourly
|
||||
// sync's own window), not the history before it.
|
||||
report(
|
||||
`requests=${fake.heartRateRequests} served=${fake.servedPoints} ` +
|
||||
`points=${POINTS} slices=${slices.length}`,
|
||||
);
|
||||
|
||||
// The heap did not grow with the history.
|
||||
const growth = peak - (anchor ?? 0);
|
||||
report(
|
||||
`heap growth after the second slice: ${mb(growth)} MB over ${samples} ` +
|
||||
`samples (budget ${mb(GROWTH_BUDGET_BYTES)} MB, ceiling ` +
|
||||
`${mb(v8.getHeapStatistics().heap_size_limit)} MB, peak ${mb(peak)} MB)`,
|
||||
);
|
||||
expect(samples).toBeGreaterThan(5);
|
||||
expect(growth).toBeLessThan(GROWTH_BUDGET_BYTES);
|
||||
}, 3_600_000);
|
||||
});
|
||||
@@ -57,6 +57,8 @@ const fake = vi.hoisted(() => ({
|
||||
throttledPages: new Set<number>(),
|
||||
/** Revoke the current access token when this heart-rate page is asked. */
|
||||
revokeAtPage: -1,
|
||||
/** Answer every heart-rate request with a 500. */
|
||||
failHeartRate: false,
|
||||
/** Hold the request for this heart-rate page until `release` is called. */
|
||||
holdAtPage: -1,
|
||||
release: null as null | (() => void),
|
||||
@@ -138,13 +140,24 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
{ "retry-after": "0" },
|
||||
);
|
||||
}
|
||||
if (fake.failHeartRate) {
|
||||
return json({ error: { code: 500, status: "INTERNAL" } }, 500);
|
||||
}
|
||||
if (page === fake.holdAtPage) {
|
||||
fake.holdAtPage = -1;
|
||||
await fake.held;
|
||||
}
|
||||
const end = Math.min(fake.heartRatePoints, (page + 1) * 10);
|
||||
// Newest first, one point a minute; a resumed walk's `<=` edge
|
||||
// filter starts the list at the edge.
|
||||
const edge = /physical_time <= "([^"]+)"/.exec(
|
||||
u.searchParams.get("filter") ?? "",
|
||||
)?.[1];
|
||||
const first = edge
|
||||
? Math.ceil((fake.now - Date.parse(edge)) / 60_000)
|
||||
: 0;
|
||||
const end = Math.min(fake.heartRatePoints, first + (page + 1) * 10);
|
||||
const dataPoints = [];
|
||||
for (let g = page * 10; g < end; g++) {
|
||||
for (let g = first + page * 10; g < end; g++) {
|
||||
dataPoints.push({
|
||||
name: `users/me/dataTypes/heart-rate/dataPoints/${1_000_000 + g}`,
|
||||
heartRate: {
|
||||
@@ -221,6 +234,7 @@ beforeEach(async () => {
|
||||
fake.throttledPages = new Set();
|
||||
fake.revokeAtPage = -1;
|
||||
fake.holdAtPage = -1;
|
||||
fake.failHeartRate = false;
|
||||
fake.release = null;
|
||||
fake.held = null;
|
||||
fake.requestsByPath = new Map();
|
||||
@@ -398,8 +412,19 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
|
||||
);
|
||||
expect(done).not.toContain("fetchHeartRate");
|
||||
|
||||
// And the heart-rate stream keeps the edge it reached: the oldest of
|
||||
// the thirty samples it wrote.
|
||||
expect(
|
||||
(afterFirst?.backfillProgress as { cursors: Record<string, unknown> })
|
||||
.cursors.fetchHeartRate,
|
||||
).toMatchObject({
|
||||
at: new Date(fake.now - 29 * 60_000).toISOString(),
|
||||
pages: 3,
|
||||
});
|
||||
|
||||
// Second attempt, the queued slice: the finished collections are not
|
||||
// read again.
|
||||
// read again, and the stream starts at its edge, not at its newest page:
|
||||
// the 31 samples from the edge on are four pages, not six.
|
||||
fake.requestsByPath = new Map();
|
||||
await runGoogleHealthBackfillForUser(USER, undefined, 1);
|
||||
|
||||
@@ -408,7 +433,7 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
|
||||
expect(
|
||||
fake.requestsByPath.get("steps/dataPoints:dailyRollUp"),
|
||||
).toBeUndefined();
|
||||
expect(fake.requestsByPath.get("heart-rate/dataPoints")).toBe(6);
|
||||
expect(fake.requestsByPath.get("heart-rate/dataPoints")).toBe(4);
|
||||
expect(await pulseRows()).toBe(60);
|
||||
const afterSecond =
|
||||
await getPrismaClient().googleHealthConnection.findUnique({
|
||||
@@ -418,7 +443,7 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
|
||||
expect(afterSecond?.backfillProgress).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps failing a backfill whose slice finished nothing new", async () => {
|
||||
it("continues a slice that only moved the stream's edge, and fails one that moved nothing", async () => {
|
||||
const { runGoogleHealthBackfillForUser } =
|
||||
await import("@/lib/jobs/google-health-backfill");
|
||||
|
||||
@@ -428,13 +453,29 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
|
||||
await runGoogleHealthBackfillForUser(USER, stopAtThirdPage);
|
||||
expect(sent.jobs).toHaveLength(1);
|
||||
|
||||
// The next slice cannot get through the heart-rate stream either: no
|
||||
// collection finishes, so this one is stuck and says so.
|
||||
// The next slice finishes no collection either, but it moves the edge
|
||||
// thirty samples further back: progress, so it continues.
|
||||
fake.requestsByPath = new Map();
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser(USER, stopAtThirdPage, 1),
|
||||
).rejects.toThrow(/incomplete/);
|
||||
expect(sent.jobs).toHaveLength(1);
|
||||
).resolves.toMatchObject({ continued: true });
|
||||
expect(sent.jobs).toHaveLength(2);
|
||||
|
||||
// A slice whose every heart-rate request fails moves nothing: it is
|
||||
// stuck, says where, and queues nothing.
|
||||
const { GOOGLE_HEALTH_TRANSIENT_RETRY } =
|
||||
await import("@/lib/google-health/client");
|
||||
const baseDelay = GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs;
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0;
|
||||
fake.failHeartRate = true;
|
||||
try {
|
||||
await expect(
|
||||
runGoogleHealthBackfillForUser(USER, undefined, 2),
|
||||
).rejects.toThrow(/no progress in slice 2; fetchHeartRate:failed/);
|
||||
} finally {
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = baseDelay;
|
||||
}
|
||||
expect(sent.jobs).toHaveLength(2);
|
||||
const row = await getPrismaClient().googleHealthConnection.findUnique({
|
||||
where: { userId: USER },
|
||||
});
|
||||
@@ -459,27 +500,32 @@ describe("Google Health backfill under rate limits and token expiry (#1194)", ()
|
||||
}
|
||||
expect(fake.holdAtPage).toBe(-1);
|
||||
|
||||
const requestsBefore = [...fake.requestsByPath.values()].reduce(
|
||||
(a, b) => a + b,
|
||||
0,
|
||||
);
|
||||
const total = () =>
|
||||
[...fake.requestsByPath.values()].reduce((a, b) => a + b, 0);
|
||||
const requestsBefore = total();
|
||||
const manual = await syncUserGoogleHealth(USER);
|
||||
expect(manual.busy).toBe(true);
|
||||
|
||||
const hourly = await handleGoogleHealthSync([] as never);
|
||||
expect(hourly).toMatchObject({
|
||||
ok: true,
|
||||
did: { users_skipped: 1, users_parked: 0, users_failed: 0 },
|
||||
// The hourly poll waits for the account instead of skipping it: it does
|
||||
// nothing while the backfill holds the lock ...
|
||||
let hourlySettled = false;
|
||||
const hourly = handleGoogleHealthSync([] as never).finally(() => {
|
||||
hourlySettled = true;
|
||||
});
|
||||
const requestsAfter = [...fake.requestsByPath.values()].reduce(
|
||||
(a, b) => a + b,
|
||||
0,
|
||||
);
|
||||
expect(requestsAfter).toBe(requestsBefore);
|
||||
await new Promise((r) => setTimeout(r, 2_500));
|
||||
expect(hourlySettled).toBe(false);
|
||||
expect(total()).toBe(requestsBefore);
|
||||
|
||||
// ... and runs once the backfill lets go.
|
||||
fake.release?.();
|
||||
await backfill;
|
||||
const requestsAfterBackfill = total();
|
||||
expect(await hourly).toMatchObject({
|
||||
ok: true,
|
||||
did: { users_skipped: 0, users_parked: 0, users_failed: 0 },
|
||||
});
|
||||
expect(total()).toBeGreaterThan(requestsAfterBackfill);
|
||||
expect(await pulseRows()).toBe(60);
|
||||
expect(await ledgerState()).toBe("connected");
|
||||
});
|
||||
}, 60_000);
|
||||
});
|
||||
Reference in new issue
Block a user