mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-10 05:12:14 +00:00
Merge branch 'v1424/ghfallback' into release/v1.42.4
This commit is contained in:
commit
a4942e631f
8 files changed
+927
-104
No files matched your search
@@ -137,6 +137,19 @@ 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.
|
||||
|
||||
Asking for "what is older" uses an upper bound on the sample time, which
|
||||
Google's filter grammar allows but which a given account may still refuse. If
|
||||
Google answers that first resumed request with `400`, the slice reads the data
|
||||
type again without the bound, passes over the readings newer than its note
|
||||
without writing them, and carries on writing from there. The refusal is noted
|
||||
per data type, so later slices read without the bound straight away. The pages
|
||||
passed over are not counted against the slice's five minutes or 500 pages:
|
||||
otherwise a history deeper than one slice would spend every slice passing over
|
||||
what earlier slices wrote and never write anything new. Only the job's own
|
||||
time limit stops a slice while it is passing over, so on such an account each
|
||||
slice takes longer the further back the backfill has got. A `400` that names
|
||||
something else (the page token, the page size) is still a failure.
|
||||
|
||||
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.
|
||||
@@ -172,12 +185,15 @@ reading:
|
||||
`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
|
||||
`token_rejected_retrying_with_fresh_token`,
|
||||
`edge_filter_rejected_reading_without_it`); 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.
|
||||
took, how many readings it wrote. `skippedPages` is the part of `pages`
|
||||
read only to pass over readings already written, on a data type whose
|
||||
upper-bound filter Google refused.
|
||||
- `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.
|
||||
@@ -191,6 +207,9 @@ reading:
|
||||
`edgeFrom` and `edgeTo`, the oldest reading written before and after the
|
||||
slice.
|
||||
- `edges`: where every unfinished page-by-page data type stands now.
|
||||
- `meta["googleHealth.backfill.edgeFilterUnsupported"]` appears on the slice
|
||||
where Google refused the upper-bound filter, listing the data types noted
|
||||
as refusing it.
|
||||
- `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
|
||||
|
||||
@@ -0,0 +1,421 @@
|
||||
/**
|
||||
* A resumed backfill walk asks Google for the samples up to its edge with an
|
||||
* upper bound (`sample_time.physical_time <= "<edge>"`). That bound is valid
|
||||
* filter grammar but has not been seen working against the live API. If
|
||||
* Google refuses it, every resumed slice would fail on its first request and
|
||||
* the backfill would be stuck again.
|
||||
*
|
||||
* - A 400 on the first resumed request is answered by reading the
|
||||
* collection again without the bound and passing over the samples beyond
|
||||
* the edge here; the rows handed on are the ones the bound would have
|
||||
* returned.
|
||||
* - The refusal is noted per collection in the backfill's progress, and the
|
||||
* next slice reads without the bound from the start.
|
||||
* - A 400 that is about something else still fails the walk.
|
||||
* - The pages passed over are not charged against the slice's budget, so a
|
||||
* skip deeper than one slice's pages cannot keep a backfill from moving.
|
||||
*/
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const { safeFetchMock, prismaMock } = vi.hoisted(() => ({
|
||||
safeFetchMock: vi.fn(),
|
||||
prismaMock: { googleHealthConnection: { update: 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/db", () => ({ prisma: prismaMock }));
|
||||
vi.mock("@/lib/logging/context", () => ({
|
||||
annotate: () => {},
|
||||
getEvent: () => null,
|
||||
}));
|
||||
|
||||
import {
|
||||
GOOGLE_HEALTH_DATA_TYPES,
|
||||
type GoogleHealthDataPoint,
|
||||
forEachDataPointPage,
|
||||
googleHealthSliceBudget,
|
||||
refusesEdgeFilter,
|
||||
runWithGoogleHealthClientOutcome,
|
||||
runWithGoogleHealthStop,
|
||||
} from "../client";
|
||||
import { GoogleHealthApiError } from "../response-classifier";
|
||||
import {
|
||||
googleHealthBackfillCursor,
|
||||
noteGoogleHealthEdgeFilterUnsupported,
|
||||
readGoogleHealthBackfillCursors,
|
||||
readGoogleHealthBackfillEdgeFilterUnsupported,
|
||||
recordGoogleHealthBackfillPage,
|
||||
runWithGoogleHealthBackfillResume,
|
||||
} from "../sync-core";
|
||||
|
||||
const MINUTE = 60_000;
|
||||
const NOW = Date.parse("2026-01-01T00:00:00Z");
|
||||
/** Reading `g` is `g` minutes old; the list runs newest first. */
|
||||
const TOTAL = 40;
|
||||
const PAGE = 5;
|
||||
const at = (g: number) => new Date(NOW - g * MINUTE);
|
||||
|
||||
const server = { refuseUpperBound: false, requests: [] as string[] };
|
||||
|
||||
/**
|
||||
* A heart-rate list built from the filter and the page token it is asked
|
||||
* with, like the multi-year integration fixture. With `refuseUpperBound`,
|
||||
* any filter carrying `<=` is refused the way Google refuses a filter.
|
||||
*/
|
||||
function respond(url: string): Response {
|
||||
const u = new URL(url);
|
||||
const filter = u.searchParams.get("filter") ?? "";
|
||||
server.requests.push(filter);
|
||||
if (server.refuseUpperBound && filter.includes("<=")) {
|
||||
return json(
|
||||
{
|
||||
error: {
|
||||
code: 400,
|
||||
status: "INVALID_ARGUMENT",
|
||||
message: "Invalid filter expression.",
|
||||
},
|
||||
},
|
||||
400,
|
||||
);
|
||||
}
|
||||
const upTo = /physical_time <= "([^"]+)"/.exec(filter)?.[1];
|
||||
const newest = upTo
|
||||
? Math.max(0, Math.ceil((NOW - Date.parse(upTo)) / MINUTE))
|
||||
: 0;
|
||||
const size = Number(u.searchParams.get("pageSize"));
|
||||
const token = Number(u.searchParams.get("pageToken") ?? 0);
|
||||
const start = newest + token * size;
|
||||
const end = Math.min(TOTAL, start + size);
|
||||
const dataPoints = [];
|
||||
for (let g = start; g < end; g++) {
|
||||
dataPoints.push({
|
||||
heartRate: {
|
||||
beatsPerMinute: String(60 + g),
|
||||
sampleTime: { physicalTime: at(g).toISOString() },
|
||||
},
|
||||
});
|
||||
}
|
||||
return json({
|
||||
dataPoints,
|
||||
...(end < TOTAL ? { nextPageToken: String(token + 1) } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
function json(body: unknown, status = 200): Response {
|
||||
return new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
type Query = Parameters<typeof forEachDataPointPage>[3];
|
||||
|
||||
/** Walk and collect what the walk hands on, page by page. */
|
||||
async function walk(query: Query) {
|
||||
const pages: GoogleHealthDataPoint[][] = [];
|
||||
const { outcome } = await runWithGoogleHealthClientOutcome(() =>
|
||||
forEachDataPointPage(
|
||||
GOOGLE_HEALTH_DATA_TYPES.heartRate,
|
||||
"token",
|
||||
"fetchHeartRate",
|
||||
{ pageSize: PAGE, ...query },
|
||||
(points) => {
|
||||
pages.push(points);
|
||||
},
|
||||
),
|
||||
);
|
||||
return { pages, rows: pages.flat(), outcome };
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
safeFetchMock.mockReset();
|
||||
safeFetchMock.mockImplementation(async (url: string) => respond(url));
|
||||
prismaMock.googleHealthConnection.update.mockReset();
|
||||
prismaMock.googleHealthConnection.update.mockResolvedValue({});
|
||||
server.refuseUpperBound = false;
|
||||
server.requests = [];
|
||||
});
|
||||
|
||||
// The edge sits mid-page so the page that crosses it is split.
|
||||
const EDGE = { at: at(17), order: "desc" as const };
|
||||
|
||||
describe("a resumed walk whose edge filter Google refuses", () => {
|
||||
it("reads again without the bound and hands on exactly the rows the bound would have", async () => {
|
||||
const filtered = await walk({ resumable: true, edge: EDGE });
|
||||
expect(server.requests[0]).toContain("<=");
|
||||
expect(filtered.rows).toHaveLength(TOTAL - 17);
|
||||
|
||||
server.requests = [];
|
||||
server.refuseUpperBound = true;
|
||||
const onFilterUnsupported = vi.fn();
|
||||
const fallback = await walk({
|
||||
resumable: true,
|
||||
edge: { ...EDGE, onFilterUnsupported },
|
||||
});
|
||||
|
||||
expect(onFilterUnsupported).toHaveBeenCalledTimes(1);
|
||||
// One refused request, then the whole walk without any filter.
|
||||
expect(server.requests[0]).toContain("<=");
|
||||
expect(server.requests.slice(1).every((f) => f === "")).toBe(true);
|
||||
expect(fallback.rows).toEqual(filtered.rows);
|
||||
expect(fallback.outcome.truncated).toBe(false);
|
||||
});
|
||||
|
||||
it("reads without the bound from the start once the collection is known to refuse it", async () => {
|
||||
const filtered = await walk({ resumable: true, edge: EDGE });
|
||||
server.requests = [];
|
||||
server.refuseUpperBound = true;
|
||||
const onFilterUnsupported = vi.fn();
|
||||
|
||||
const skipped = await walk({
|
||||
resumable: true,
|
||||
edge: { ...EDGE, filter: false, onFilterUnsupported },
|
||||
});
|
||||
|
||||
expect(server.requests.some((f) => f.includes("<="))).toBe(false);
|
||||
expect(onFilterUnsupported).not.toHaveBeenCalled();
|
||||
expect(skipped.rows).toEqual(filtered.rows);
|
||||
});
|
||||
|
||||
it("follows a walk that runs oldest first", async () => {
|
||||
// An ascending walk keeps the samples at or after its edge.
|
||||
safeFetchMock.mockImplementation(async () =>
|
||||
json({
|
||||
dataPoints: [5, 4, 3, 2, 1, 0].map((g) => ({
|
||||
heartRate: {
|
||||
beatsPerMinute: "60",
|
||||
sampleTime: { physicalTime: at(g).toISOString() },
|
||||
},
|
||||
})),
|
||||
}),
|
||||
);
|
||||
const { rows } = await walk({
|
||||
resumable: true,
|
||||
edge: { at: at(2), order: "asc", filter: false },
|
||||
});
|
||||
expect(rows).toHaveLength(3);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a 400 that is not about the edge filter", () => {
|
||||
it("fails the walk when the first request names something else", async () => {
|
||||
safeFetchMock.mockImplementation(async () =>
|
||||
json(
|
||||
{
|
||||
error: {
|
||||
code: 400,
|
||||
status: "INVALID_ARGUMENT",
|
||||
message: "Invalid page token.",
|
||||
},
|
||||
},
|
||||
400,
|
||||
),
|
||||
);
|
||||
const onFilterUnsupported = vi.fn();
|
||||
await expect(
|
||||
walk({ resumable: true, edge: { ...EDGE, onFilterUnsupported } }),
|
||||
).rejects.toBeInstanceOf(GoogleHealthApiError);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(onFilterUnsupported).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("fails the walk on a 400 past the first page, where the filter was already taken", async () => {
|
||||
let n = 0;
|
||||
safeFetchMock.mockImplementation(async (url: string) =>
|
||||
++n === 2
|
||||
? json({ error: { code: 400, status: "INVALID_ARGUMENT" } }, 400)
|
||||
: respond(url),
|
||||
);
|
||||
const onFilterUnsupported = vi.fn();
|
||||
await expect(
|
||||
walk({ resumable: true, edge: { ...EDGE, onFilterUnsupported } }),
|
||||
).rejects.toBeInstanceOf(GoogleHealthApiError);
|
||||
expect(onFilterUnsupported).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("fails a walk without an edge on a 400 as before", async () => {
|
||||
safeFetchMock.mockImplementation(async () =>
|
||||
json({ error: { code: 400, status: "INVALID_ARGUMENT" } }, 400),
|
||||
);
|
||||
await expect(walk({ resumable: true })).rejects.toBeInstanceOf(
|
||||
GoogleHealthApiError,
|
||||
);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("is told apart by what Google says", () => {
|
||||
expect(refusesEdgeFilter(undefined)).toBe(true);
|
||||
expect(refusesEdgeFilter("INVALID_ARGUMENT: Invalid filter.")).toBe(true);
|
||||
expect(refusesEdgeFilter("INVALID_ARGUMENT")).toBe(true);
|
||||
expect(refusesEdgeFilter("FAILED_PRECONDITION: filter on field")).toBe(
|
||||
true,
|
||||
);
|
||||
expect(refusesEdgeFilter("INVALID_ARGUMENT: Invalid page token.")).toBe(
|
||||
false,
|
||||
);
|
||||
expect(refusesEdgeFilter("INVALID_ARGUMENT: page_size too large")).toBe(
|
||||
false,
|
||||
);
|
||||
expect(refusesEdgeFilter("FAILED_PRECONDITION: account not set up")).toBe(
|
||||
false,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe("passing over the samples beyond the edge inside a slice", () => {
|
||||
it("is not charged against the slice's pages, so the slice still writes its share", async () => {
|
||||
// The edge is three pages deep; the slice may read two.
|
||||
const budget = googleHealthSliceBudget(60_000, 2);
|
||||
const { pages, outcome } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() =>
|
||||
walk({
|
||||
resumable: true,
|
||||
edge: { at: at(15), order: "desc", filter: false },
|
||||
}),
|
||||
budget,
|
||||
);
|
||||
expect(pages).toHaveLength(2);
|
||||
expect(budget.skippedPages).toBe(3);
|
||||
expect(budget.pages).toBe(5);
|
||||
expect(budget.stopReason).toBe("page_budget");
|
||||
expect(outcome.truncated).toBe(true);
|
||||
});
|
||||
|
||||
it("does not stop on the slice's clock while it is still passing over", async () => {
|
||||
let clock = 0;
|
||||
safeFetchMock.mockImplementation(async (url: string) => {
|
||||
clock += 10_000;
|
||||
return respond(url);
|
||||
});
|
||||
// Each request takes ten seconds and the slice has fifteen, counted from
|
||||
// when the collection started; passing over the first three pages takes
|
||||
// forty.
|
||||
const budget = googleHealthSliceBudget(15_000, 500, () => clock);
|
||||
budget.deadline = 15_000;
|
||||
const { pages } = await runWithGoogleHealthStop(
|
||||
() => false,
|
||||
() =>
|
||||
walk({
|
||||
resumable: true,
|
||||
edge: { at: at(15), order: "desc", filter: false },
|
||||
}),
|
||||
budget,
|
||||
);
|
||||
// The crossing page and the two after it, inside the fifteen seconds the
|
||||
// slice still had once the skip was given back.
|
||||
expect(pages).toHaveLength(3);
|
||||
expect(budget.deadline).toBe(55_000);
|
||||
expect(budget.stopReason).toBe("time_budget");
|
||||
});
|
||||
|
||||
it("still stops on the job's own budget", async () => {
|
||||
let asked = 0;
|
||||
const { pages, outcome } = await runWithGoogleHealthStop(
|
||||
() => ++asked > 2,
|
||||
() =>
|
||||
walk({
|
||||
resumable: true,
|
||||
edge: { at: at(30), order: "desc", filter: false },
|
||||
}),
|
||||
googleHealthSliceBudget(60_000, 500),
|
||||
);
|
||||
expect(pages).toHaveLength(0);
|
||||
expect(outcome.truncated).toBe(true);
|
||||
});
|
||||
|
||||
it("gives up past its own ceiling instead of reading without end", async () => {
|
||||
const { pages, outcome } = await walk({
|
||||
resumable: true,
|
||||
edge: { at: at(30), order: "desc", filter: false, maxSkipPages: 2 },
|
||||
});
|
||||
expect(pages).toHaveLength(0);
|
||||
expect(safeFetchMock).toHaveBeenCalledTimes(2);
|
||||
expect(outcome.truncated).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("the note that a collection refuses the edge filter", () => {
|
||||
/** The progress the last write left in the column. */
|
||||
const lastProgress = () =>
|
||||
prismaMock.googleHealthConnection.update.mock.calls.at(-1)![0].data
|
||||
.backfillProgress as unknown;
|
||||
|
||||
it("is written once Google refuses, kept by later writes, and honoured by the next slice", async () => {
|
||||
server.refuseUpperBound = true;
|
||||
const cursors = new Map([
|
||||
[
|
||||
"fetchHeartRate",
|
||||
{ at: EDGE.at.toISOString(), order: "desc" as const, pages: 4 },
|
||||
],
|
||||
]);
|
||||
|
||||
// Slice one: resumes with the bound, is refused, falls back, writes a page.
|
||||
await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(),
|
||||
async () => {
|
||||
const cursor = googleHealthBackfillCursor("fetchHeartRate")!;
|
||||
expect(cursor.edge?.filter).toBe(true);
|
||||
await walk({
|
||||
resumable: true,
|
||||
maxPages: 1,
|
||||
edge: {
|
||||
...cursor.edge!,
|
||||
onFilterUnsupported: () =>
|
||||
noteGoogleHealthEdgeFilterUnsupported("fetchHeartRate"),
|
||||
},
|
||||
});
|
||||
await recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at(17),
|
||||
last: at(19),
|
||||
});
|
||||
},
|
||||
new Map(cursors),
|
||||
);
|
||||
expect(server.requests.filter((f) => f.includes("<="))).toHaveLength(1);
|
||||
const progress = lastProgress();
|
||||
expect(progress).toMatchObject({
|
||||
edgeFilterUnsupported: ["fetchHeartRate"],
|
||||
cursors: { fetchHeartRate: { at: at(19).toISOString() } },
|
||||
});
|
||||
|
||||
// Slice two: starts from what slice one left on disk.
|
||||
server.requests = [];
|
||||
const unsupported = readGoogleHealthBackfillEdgeFilterUnsupported(progress);
|
||||
await runWithGoogleHealthBackfillResume(
|
||||
"u1",
|
||||
new Set(),
|
||||
async () => {
|
||||
const cursor = googleHealthBackfillCursor("fetchHeartRate")!;
|
||||
expect(cursor.edge?.filter).toBe(false);
|
||||
const { rows } = await walk({ resumable: true, edge: cursor.edge });
|
||||
expect(rows).toHaveLength(TOTAL - 19);
|
||||
},
|
||||
readGoogleHealthBackfillCursors(progress),
|
||||
unsupported,
|
||||
);
|
||||
expect(server.requests.some((f) => f.includes("<="))).toBe(false);
|
||||
});
|
||||
|
||||
it("leaves the progress in its old shape for a backfill that never met the refusal", async () => {
|
||||
await runWithGoogleHealthBackfillResume("u1", new Set(), async () => {
|
||||
await recordGoogleHealthBackfillPage("fetchHeartRate", {
|
||||
first: at(0),
|
||||
last: at(4),
|
||||
});
|
||||
});
|
||||
expect(lastProgress()).not.toHaveProperty("edgeFilterUnsupported");
|
||||
});
|
||||
|
||||
it("reads only well-formed keys", () => {
|
||||
expect([
|
||||
...readGoogleHealthBackfillEdgeFilterUnsupported({
|
||||
edgeFilterUnsupported: ["fetchHeartRate", 3, "x".repeat(65)],
|
||||
}),
|
||||
]).toEqual(["fetchHeartRate"]);
|
||||
expect(readGoogleHealthBackfillEdgeFilterUnsupported(null).size).toBe(0);
|
||||
});
|
||||
});
|
||||
+235
-91
@@ -43,10 +43,11 @@ import {
|
||||
classifyGoogleHealthResponse,
|
||||
type GoogleHealthClassification,
|
||||
} from "./response-classifier";
|
||||
import type {
|
||||
GoogleHealthDataPoint,
|
||||
GoogleHealthDataType,
|
||||
GoogleHealthRollupPoint,
|
||||
import {
|
||||
googleHealthSampleInstant,
|
||||
type GoogleHealthDataPoint,
|
||||
type GoogleHealthDataType,
|
||||
type GoogleHealthRollupPoint,
|
||||
} from "./mappers";
|
||||
import { envValue } from "@/lib/env";
|
||||
|
||||
@@ -165,6 +166,12 @@ export interface GoogleHealthSliceBudget {
|
||||
deadline: number | null;
|
||||
maxPages: number;
|
||||
pages: number;
|
||||
/**
|
||||
* Of `pages`, those a walk only passed over on its way to its edge, because
|
||||
* Google refused the edge filter (see `forEachDataPointPage`). They are not
|
||||
* charged against `maxPages`, and the time they took moves `deadline` on.
|
||||
*/
|
||||
skippedPages: number;
|
||||
stopReason: GoogleHealthSliceStopReason | null;
|
||||
now: () => number;
|
||||
}
|
||||
@@ -180,6 +187,7 @@ export function googleHealthSliceBudget(
|
||||
deadline: null,
|
||||
maxPages,
|
||||
pages: 0,
|
||||
skippedPages: 0,
|
||||
stopReason: null,
|
||||
now,
|
||||
};
|
||||
@@ -192,7 +200,9 @@ 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";
|
||||
else if (slice.pages - slice.skippedPages >= slice.maxPages) {
|
||||
slice.stopReason = "page_budget";
|
||||
}
|
||||
return slice.stopReason !== null;
|
||||
}
|
||||
|
||||
@@ -242,6 +252,22 @@ function noteSlicePage(): void {
|
||||
if (slice) slice.pages += 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Give back to the slice in scope the pages and time a walk spent passing
|
||||
* over samples beyond its edge (no-op outside a slice).
|
||||
*/
|
||||
function refundSliceSkip(pages: number, ms: number): void {
|
||||
const slice = stopStorage.getStore()?.slice;
|
||||
if (!slice) return;
|
||||
slice.skippedPages += pages;
|
||||
if (slice.deadline !== null) slice.deadline += ms;
|
||||
}
|
||||
|
||||
/** The slice clock in scope, or the wall clock outside a slice. */
|
||||
function sliceNow(): number {
|
||||
return stopStorage.getStore()?.slice?.now() ?? Date.now();
|
||||
}
|
||||
|
||||
/** End the slice in scope for `reason` (no-op outside a slice). */
|
||||
function endSlice(reason: GoogleHealthSliceStopReason): void {
|
||||
const slice = stopStorage.getStore()?.slice;
|
||||
@@ -716,8 +742,22 @@ interface DataPointQuery {
|
||||
* 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.
|
||||
*
|
||||
* `filter: false` reads without the edge bound and passes over the samples
|
||||
* beyond the edge here instead (see `forEachDataPointPage`); the walk falls
|
||||
* back to that on its own when Google refuses the bound, and calls
|
||||
* `onFilterUnsupported` once when it does. `tz` reads an offset-less sample
|
||||
* time the way the edge was recorded. `maxSkipPages` bounds how many pages
|
||||
* the walk may pass over before it reaches the edge.
|
||||
*/
|
||||
edge?: { at: Date; order: "desc" | "asc" };
|
||||
edge?: {
|
||||
at: Date;
|
||||
order: "desc" | "asc";
|
||||
filter?: boolean;
|
||||
tz?: string;
|
||||
onFilterUnsupported?: () => void | Promise<void>;
|
||||
maxSkipPages?: number;
|
||||
};
|
||||
}
|
||||
|
||||
const pad2 = (n: number): string => String(n).padStart(2, "0");
|
||||
@@ -984,7 +1024,10 @@ async function googleHealthDataRequest<T>(
|
||||
token: GoogleHealthAccessToken,
|
||||
label: string,
|
||||
send: (accessToken: string) => Promise<Response>,
|
||||
isExpectedFallback: (status: number) => boolean = () => false,
|
||||
expectedFallbackNote: (
|
||||
status: number,
|
||||
apiErrorDetail: string | undefined,
|
||||
) => string | undefined = () => undefined,
|
||||
): Promise<GoogleHealthDataResponse<T>> {
|
||||
let rateLimited = 0;
|
||||
let transient = 0;
|
||||
@@ -1060,8 +1103,8 @@ async function googleHealthDataRequest<T>(
|
||||
method: label,
|
||||
duration_ms: Math.round(performance.now() - started),
|
||||
status: res.status,
|
||||
...(isExpectedFallback(res.status)
|
||||
? { note: "date_filter_rejected_retrying_snake_case" }
|
||||
...(expectedFallbackNote(res.status, apiErrorDetail)
|
||||
? { note: expectedFallbackNote(res.status, apiErrorDetail) }
|
||||
: willWait
|
||||
? { note: `rate_limited_retrying_in_ms=${backoffMs}` }
|
||||
: willRefresh
|
||||
@@ -1152,99 +1195,163 @@ export async function forEachDataPointPage(
|
||||
const maxPages = query.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES;
|
||||
const pageSize = query.pageSize ?? GOOGLE_HEALTH_PAGE_SIZE;
|
||||
const resumable = query.resumable === true;
|
||||
const edge = query.edge;
|
||||
|
||||
let requestCount = 0;
|
||||
|
||||
const walk = async (
|
||||
dateStyle: GoogleHealthDateFilterStyle,
|
||||
edgeBy: "filter" | "skip",
|
||||
): Promise<void> => {
|
||||
let fetched = 0;
|
||||
let pageToken: string | null | undefined;
|
||||
let pageCount = 0;
|
||||
let stopped = false;
|
||||
// Reading without the edge bound: the pages before the edge are passed
|
||||
// over, not written, until the first sample at or beyond it.
|
||||
const skipTo = edge && edgeBy === "skip" ? edge : undefined;
|
||||
let crossed = skipTo === undefined;
|
||||
let skipped = 0;
|
||||
const skipStartedAt = skipTo ? sliceNow() : 0;
|
||||
const maxSkip = skipTo?.maxSkipPages ?? GOOGLE_HEALTH_DENSE_MAX_PAGES;
|
||||
const beyondEdge = (point: GoogleHealthDataPoint): boolean | null => {
|
||||
const t = googleHealthSampleInstant(point, dataType, skipTo?.tz);
|
||||
if (t === null || !skipTo) return null;
|
||||
return skipTo.order === "desc"
|
||||
? t.getTime() > skipTo.at.getTime()
|
||||
: t.getTime() < skipTo.at.getTime();
|
||||
};
|
||||
const refundSkip = () => {
|
||||
if (skipped > 0) refundSliceSkip(skipped, sliceNow() - skipStartedAt);
|
||||
skipped = 0;
|
||||
};
|
||||
|
||||
do {
|
||||
if (stopRequested(resumable)) {
|
||||
stopped = true;
|
||||
break;
|
||||
}
|
||||
const params = new URLSearchParams({ pageSize: String(pageSize) });
|
||||
const filters: string[] = [];
|
||||
if (query.start) {
|
||||
const { field, bound } = incrementalFilter(
|
||||
dataType,
|
||||
query.start,
|
||||
query.tz,
|
||||
dateStyle,
|
||||
);
|
||||
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);
|
||||
try {
|
||||
do {
|
||||
// A walk still passing over the samples beyond its edge does not stop
|
||||
// on the slice's budget, only on the job's: the skip writes nothing,
|
||||
// so a slice that stopped inside it would have nothing to show, and
|
||||
// the next slice would start the same skip over (see
|
||||
// `GoogleHealthSliceBudget.skippedPages`).
|
||||
if (crossed ? stopRequested(resumable) : stopRequested(false)) {
|
||||
stopped = true;
|
||||
break;
|
||||
}
|
||||
if (!crossed && skipped >= maxSkip) {
|
||||
getEvent()?.addWarning(
|
||||
`google-health ${verb}: passed over ${skipped} pages without reaching the resume edge`,
|
||||
);
|
||||
stopped = true;
|
||||
break;
|
||||
}
|
||||
const params = new URLSearchParams({ pageSize: String(pageSize) });
|
||||
const filters: string[] = [];
|
||||
if (query.start) {
|
||||
const { field, bound } = incrementalFilter(
|
||||
dataType,
|
||||
query.start,
|
||||
query.tz,
|
||||
dateStyle,
|
||||
);
|
||||
filters.push(`${field} >= "${bound}"`);
|
||||
}
|
||||
const edgeFiltered = edge !== undefined && edgeBy === "filter";
|
||||
if (edgeFiltered) {
|
||||
const { field, bound } = incrementalFilter(
|
||||
dataType,
|
||||
edge.at,
|
||||
query.tz,
|
||||
dateStyle,
|
||||
);
|
||||
filters.push(
|
||||
`${field} ${edge.order === "desc" ? "<=" : ">="} "${bound}"`,
|
||||
);
|
||||
}
|
||||
if (filters.length > 0) params.set("filter", filters.join(" AND "));
|
||||
if (pageToken) params.set("pageToken", pageToken);
|
||||
|
||||
requestCount += 1;
|
||||
// v1.42 (#1023) — the first filtered daily-summary read answering 400
|
||||
// is the expected trigger for the snake_case retry below, not a
|
||||
// failure: some accounts reject the documented camelCase prefix on
|
||||
// some types, and the very next request answers 200. It was logged as
|
||||
// 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 { status, json, verdict, apiErrorDetail } =
|
||||
await googleHealthDataRequest<GoogleHealthDataPointPage>(
|
||||
accessToken,
|
||||
`${verb}(page=${pageCount})`,
|
||||
(token) =>
|
||||
safeFetch(
|
||||
`${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
|
||||
{
|
||||
method: "GET",
|
||||
headers: { Authorization: `Bearer ${token}` },
|
||||
},
|
||||
),
|
||||
(status) =>
|
||||
canFallBack &&
|
||||
dateStyle === "camel" &&
|
||||
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,
|
||||
classification: verdict.classification,
|
||||
httpStatus: verdict.httpStatus,
|
||||
reason: verdict.reason,
|
||||
upstreamError: apiErrorDetail,
|
||||
});
|
||||
}
|
||||
requestCount += 1;
|
||||
// v1.42 (#1023) — the first filtered daily-summary read answering 400
|
||||
// is the expected trigger for the snake_case retry below, not a
|
||||
// failure: some accounts reject the documented camelCase prefix on
|
||||
// some types, and the very next request answers 200. It was logged as
|
||||
// an error next to the 200 that followed it. It is recorded as a note
|
||||
// now; any other 400 still carries its error. The same holds for the
|
||||
// first resumed request refusing the edge bound.
|
||||
const firstRequest = requestCount === 1;
|
||||
const firstOfWalk = pageCount === 0;
|
||||
const edgeRefused = (status: number, detail: string | undefined) =>
|
||||
edgeFiltered &&
|
||||
firstOfWalk &&
|
||||
status === 400 &&
|
||||
refusesEdgeFilter(detail);
|
||||
const { status, json, verdict, apiErrorDetail } =
|
||||
await googleHealthDataRequest<GoogleHealthDataPointPage>(
|
||||
accessToken,
|
||||
`${verb}(page=${pageCount})`,
|
||||
(token) =>
|
||||
safeFetch(
|
||||
`${GOOGLE_HEALTH_API_BASE}/users/me/dataTypes/${dataType.path}/dataPoints?${params}`,
|
||||
{
|
||||
method: "GET",
|
||||
headers: { Authorization: `Bearer ${token}` },
|
||||
},
|
||||
),
|
||||
(status, detail) =>
|
||||
canFallBack &&
|
||||
dateStyle === "camel" &&
|
||||
firstRequest &&
|
||||
status === 400
|
||||
? "date_filter_rejected_retrying_snake_case"
|
||||
: edgeRefused(status, detail)
|
||||
? "edge_filter_rejected_reading_without_it"
|
||||
: undefined,
|
||||
);
|
||||
if (edgeRefused(status, apiErrorDetail)) {
|
||||
throw new EdgeFilterRefused();
|
||||
}
|
||||
// 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,
|
||||
classification: verdict.classification,
|
||||
httpStatus: verdict.httpStatus,
|
||||
reason: verdict.reason,
|
||||
upstreamError: apiErrorDetail,
|
||||
});
|
||||
}
|
||||
|
||||
const page = json?.dataPoints ?? [];
|
||||
pageToken = json?.nextPageToken ?? null;
|
||||
pageCount += 1;
|
||||
fetched += page.length;
|
||||
noteSlicePage();
|
||||
await onPage(page);
|
||||
} while (pageToken && pageCount < maxPages);
|
||||
let page = json?.dataPoints ?? [];
|
||||
pageToken = json?.nextPageToken ?? null;
|
||||
pageCount += 1;
|
||||
fetched += page.length;
|
||||
noteSlicePage();
|
||||
if (!crossed) {
|
||||
// Placed samples beyond the edge were written by an earlier slice;
|
||||
// one without a sample time cannot be placed and is passed over
|
||||
// with them until the walk is past the edge.
|
||||
const kept = page.filter((p) => beyondEdge(p) === false);
|
||||
if (kept.length === 0) {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
crossed = true;
|
||||
refundSkip();
|
||||
page = page.filter((p) => beyondEdge(p) !== true);
|
||||
}
|
||||
await onPage(page);
|
||||
} while (pageToken && pageCount - skipped < maxPages);
|
||||
} finally {
|
||||
refundSkip();
|
||||
}
|
||||
|
||||
notePagination(pageCount, fetched, stopped || Boolean(pageToken));
|
||||
};
|
||||
@@ -1255,8 +1362,20 @@ export async function forEachDataPointPage(
|
||||
dataType.timeField === "date" && query.start !== undefined;
|
||||
|
||||
let style: GoogleHealthDateFilterStyle = "camel";
|
||||
let edgeBy: "filter" | "skip" = edge?.filter === false ? "skip" : "filter";
|
||||
const run = async (dateStyle: GoogleHealthDateFilterStyle) => {
|
||||
try {
|
||||
await walk(dateStyle, edgeBy);
|
||||
} catch (err) {
|
||||
if (!(err instanceof EdgeFilterRefused)) throw err;
|
||||
// Once: the walk without the bound sends no edge filter to refuse.
|
||||
edgeBy = "skip";
|
||||
await edge?.onFilterUnsupported?.();
|
||||
await walk(dateStyle, edgeBy);
|
||||
}
|
||||
};
|
||||
try {
|
||||
await walk("camel");
|
||||
await run("camel");
|
||||
} catch (err) {
|
||||
const firstRequestRejected =
|
||||
canFallBack &&
|
||||
@@ -1265,7 +1384,7 @@ export async function forEachDataPointPage(
|
||||
err.httpStatus === 400;
|
||||
if (!firstRequestRejected) throw err;
|
||||
style = "snake";
|
||||
await walk("snake");
|
||||
await run("snake");
|
||||
}
|
||||
if (canFallBack) {
|
||||
annotate({ meta: { "googleHealth.dateFilter.style": style } });
|
||||
@@ -1273,6 +1392,31 @@ export async function forEachDataPointPage(
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown inside a walk whose first request Google refused because of the
|
||||
* resume edge's bound; `forEachDataPointPage` reads the collection again
|
||||
* without it.
|
||||
*/
|
||||
class EdgeFilterRefused extends Error {
|
||||
constructor() {
|
||||
super("google-health: edge filter refused");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a 400 on the first resumed request is Google refusing the edge
|
||||
* bound, rather than something else about the request. The request differs
|
||||
* from one Google answers every hour only by that bound, so a bare 400 or an
|
||||
* INVALID_ARGUMENT counts, and so does any message naming the filter; one
|
||||
* naming the page token, the page size or the access token does not.
|
||||
*/
|
||||
export function refusesEdgeFilter(detail: string | undefined): boolean {
|
||||
if (detail === undefined) return true;
|
||||
if (/filter/i.test(detail)) return true;
|
||||
if (/token|page.?size/i.test(detail)) return false;
|
||||
return /INVALID_ARGUMENT/.test(detail);
|
||||
}
|
||||
|
||||
// ─── Daily roll-up reads (`POST …/dataPoints:dailyRollUp`) ─────────────────
|
||||
|
||||
/**
|
||||
|
||||
@@ -426,6 +426,13 @@ export async function runWithGoogleHealthSyncCycle<T>(
|
||||
* 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).
|
||||
*
|
||||
* The upper bound (`<=`) a newest-first walk resumes with is valid filter
|
||||
* grammar but has not been seen working against the live API. A collection
|
||||
* whose resumed request Google refuses with a 400 is noted in
|
||||
* `edgeFilterUnsupported`, and its walks from then on read without the bound
|
||||
* and pass over the samples beyond the edge themselves (see
|
||||
* `forEachDataPointPage`).
|
||||
*/
|
||||
export interface GoogleHealthBackfillCursor {
|
||||
/** The edge: the sample instant (ISO 8601) the walk has written up to. */
|
||||
@@ -462,6 +469,8 @@ interface BackfillResumeTracker {
|
||||
/** Collections whose edge this attempt moved, with the edge they started from. */
|
||||
advanced: Map<string, string | null>;
|
||||
collections: GoogleHealthBackfillCollectionReport[];
|
||||
/** Collections whose resumed request Google refused the edge filter on. */
|
||||
edgeFilterUnsupported: Set<string>;
|
||||
}
|
||||
const backfillResumeStorage = new AsyncLocalStorage<BackfillResumeTracker>();
|
||||
|
||||
@@ -506,6 +515,23 @@ export function readGoogleHealthBackfillCursors(
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the `backfillProgress` column into the collections Google refused the
|
||||
* edge filter on.
|
||||
*/
|
||||
export function readGoogleHealthBackfillEdgeFilterUnsupported(
|
||||
value: unknown,
|
||||
): Set<string> {
|
||||
const keys = progressRecord(value).edgeFilterUnsupported;
|
||||
return new Set(
|
||||
Array.isArray(keys)
|
||||
? keys.filter(
|
||||
(key): key is string => typeof key === "string" && key.length <= 64,
|
||||
)
|
||||
: [],
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
@@ -516,8 +542,11 @@ async function persistBackfillProgress(
|
||||
next: {
|
||||
done: Set<string>;
|
||||
cursors: Map<string, GoogleHealthBackfillCursor>;
|
||||
edgeFilterUnsupported?: Set<string>;
|
||||
},
|
||||
): Promise<void> {
|
||||
const unsupported =
|
||||
next.edgeFilterUnsupported ?? tracker.edgeFilterUnsupported;
|
||||
await prisma.googleHealthConnection.update({
|
||||
where: { userId: tracker.userId },
|
||||
data: {
|
||||
@@ -529,9 +558,17 @@ async function persistBackfillProgress(
|
||||
{ at: c.at, order: c.order, pages: c.pages },
|
||||
]),
|
||||
),
|
||||
// Only once there is something to say, so a backfill that never met
|
||||
// the refusal keeps the shape it always had.
|
||||
...(unsupported.size > 0
|
||||
? { edgeFilterUnsupported: [...unsupported] }
|
||||
: {}),
|
||||
},
|
||||
},
|
||||
});
|
||||
if (unsupported !== tracker.edgeFilterUnsupported) {
|
||||
for (const key of unsupported) tracker.edgeFilterUnsupported.add(key);
|
||||
}
|
||||
// `next` may be the tracker's own collections: copy before clearing.
|
||||
const done = [...next.done];
|
||||
const cursors = [...next.cursors];
|
||||
@@ -561,6 +598,7 @@ export async function runWithGoogleHealthBackfillResume<T>(
|
||||
done: Set<string>,
|
||||
fn: () => Promise<T>,
|
||||
cursors: Map<string, GoogleHealthBackfillCursor> = new Map(),
|
||||
edgeFilterUnsupported: Set<string> = new Set(),
|
||||
): Promise<GoogleHealthBackfillAttempt<T>> {
|
||||
const tracker: BackfillResumeTracker = {
|
||||
userId,
|
||||
@@ -569,6 +607,7 @@ export async function runWithGoogleHealthBackfillResume<T>(
|
||||
skipped: [],
|
||||
advanced: new Map(),
|
||||
collections: [],
|
||||
edgeFilterUnsupported,
|
||||
};
|
||||
const result = await backfillResumeStorage.run(tracker, fn);
|
||||
return {
|
||||
@@ -581,11 +620,13 @@ export async function runWithGoogleHealthBackfillResume<T>(
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* edge an earlier attempt recorded (if any), whether Google may be asked for
|
||||
* it with a filter (false once Google refused that filter for the
|
||||
* collection), 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" };
|
||||
edge?: { at: Date; order: "desc" | "asc"; filter: boolean };
|
||||
pagesSoFar: number;
|
||||
} | null {
|
||||
const resume = backfillResumeStorage.getStore();
|
||||
@@ -593,12 +634,47 @@ export function googleHealthBackfillCursor(key: GoogleHealthCollectionKey): {
|
||||
const cursor = resume.cursors.get(key);
|
||||
return cursor
|
||||
? {
|
||||
edge: { at: new Date(cursor.at), order: cursor.order },
|
||||
edge: {
|
||||
at: new Date(cursor.at),
|
||||
order: cursor.order,
|
||||
filter: !resume.edgeFilterUnsupported.has(key),
|
||||
},
|
||||
pagesSoFar: cursor.pages,
|
||||
}
|
||||
: { pagesSoFar: 0 };
|
||||
}
|
||||
|
||||
/**
|
||||
* Note that Google refused the edge filter on a resumed request of this
|
||||
* collection, so every later slice reads without it from the start instead
|
||||
* of paying for the refusal again. Annotated as
|
||||
* `googleHealth.backfill.edgeFilterUnsupported` (the collections noted so
|
||||
* far) on the run that met the refusal. A note that cannot be written is a
|
||||
* warning, not a failure: the next slice meets the refusal once more and
|
||||
* falls back again. Outside a backfill this does nothing.
|
||||
*/
|
||||
export async function noteGoogleHealthEdgeFilterUnsupported(
|
||||
key: GoogleHealthCollectionKey,
|
||||
): Promise<void> {
|
||||
const resume = backfillResumeStorage.getStore();
|
||||
if (!resume || resume.edgeFilterUnsupported.has(key)) return;
|
||||
const unsupported = new Set([...resume.edgeFilterUnsupported, key]);
|
||||
annotate({
|
||||
meta: { "googleHealth.backfill.edgeFilterUnsupported": [...unsupported] },
|
||||
});
|
||||
try {
|
||||
await persistBackfillProgress(resume, {
|
||||
done: resume.done,
|
||||
cursors: resume.cursors,
|
||||
edgeFilterUnsupported: unsupported,
|
||||
});
|
||||
} catch (err) {
|
||||
getEvent()?.addWarning(
|
||||
`google-health: backfill progress write failed for ${key}: ${err}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record that a resumable collection wrote one more page, holding samples
|
||||
* from `first` to `last` in the order Google returned them. Moves the
|
||||
|
||||
@@ -54,6 +54,7 @@ import {
|
||||
googleHealthHardFailureCount,
|
||||
googleHealthTokenSource,
|
||||
handleCollectionFetchError,
|
||||
noteGoogleHealthEdgeFilterUnsupported,
|
||||
recordGoogleHealthBackfillPage,
|
||||
runGoogleHealthCollection,
|
||||
upsertGoogleHealthMeasurements,
|
||||
@@ -277,9 +278,8 @@ async function syncMetricCollection(
|
||||
const cursor = isResumable(resource)
|
||||
? googleHealthBackfillCursor(resource.verb)
|
||||
: null;
|
||||
const maxPages =
|
||||
(resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES) -
|
||||
(cursor?.pagesSoFar ?? 0);
|
||||
const ceiling = resource.maxPages ?? GOOGLE_HEALTH_DEFAULT_MAX_PAGES;
|
||||
const maxPages = ceiling - (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}`,
|
||||
@@ -301,7 +301,20 @@ async function syncMetricCollection(
|
||||
? (opts.intradayStart ?? opts.start)
|
||||
: opts.start,
|
||||
maxPages,
|
||||
...(cursor ? { resumable: true, edge: cursor.edge } : {}),
|
||||
...(cursor
|
||||
? {
|
||||
resumable: true,
|
||||
edge: cursor.edge && {
|
||||
...cursor.edge,
|
||||
tz,
|
||||
// Without the bound the walk passes over what earlier slices
|
||||
// wrote, and they wrote no more than the collection's ceiling.
|
||||
maxSkipPages: ceiling,
|
||||
onFilterUnsupported: () =>
|
||||
noteGoogleHealthEdgeFilterUnsupported(resource.verb),
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
async (points) => {
|
||||
const readings: GoogleHealthMeasurementUpsert[] = [];
|
||||
|
||||
@@ -61,6 +61,7 @@ vi.mock("@/lib/google-health/sync-core", () => ({
|
||||
GOOGLE_HEALTH_INTEGRATION_KEY: "google-health",
|
||||
readGoogleHealthBackfillDone: () => new Set<string>(resumeState.doneBefore),
|
||||
readGoogleHealthBackfillCursors: () => new Map(),
|
||||
readGoogleHealthBackfillEdgeFilterUnsupported: () => new Set<string>(),
|
||||
runWithGoogleHealthBackfillResume: async (
|
||||
_userId: string,
|
||||
done: Set<string>,
|
||||
@@ -316,6 +317,7 @@ describe("runGoogleHealthBackfillForUser — slices", () => {
|
||||
deadline: null,
|
||||
maxPages: 500,
|
||||
pages: 3,
|
||||
skippedPages: 0,
|
||||
stopReason: "rate_limited",
|
||||
now: Date.now,
|
||||
}));
|
||||
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
import {
|
||||
readGoogleHealthBackfillCursors,
|
||||
readGoogleHealthBackfillDone,
|
||||
readGoogleHealthBackfillEdgeFilterUnsupported,
|
||||
runWithGoogleHealthBackfillResume,
|
||||
} from "@/lib/google-health/sync-core";
|
||||
import {
|
||||
@@ -185,6 +186,9 @@ export async function runGoogleHealthBackfillForUser(
|
||||
}
|
||||
const done = readGoogleHealthBackfillDone(connection?.backfillProgress);
|
||||
const cursors = readGoogleHealthBackfillCursors(connection?.backfillProgress);
|
||||
const edgeFilterUnsupported = readGoogleHealthBackfillEdgeFilterUnsupported(
|
||||
connection?.backfillProgress,
|
||||
);
|
||||
const doneBefore = done.size;
|
||||
|
||||
const budget = googleHealthSliceBudget(
|
||||
@@ -211,6 +215,7 @@ export async function runGoogleHealthBackfillForUser(
|
||||
budget,
|
||||
),
|
||||
cursors,
|
||||
edgeFilterUnsupported,
|
||||
);
|
||||
|
||||
const finished = done.size - doneBefore;
|
||||
@@ -218,6 +223,9 @@ export async function runGoogleHealthBackfillForUser(
|
||||
slice,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
pages: budget.pages,
|
||||
// Of those, the pages read only to pass over samples beyond an edge,
|
||||
// where Google refused the edge filter.
|
||||
skippedPages: budget.skippedPages,
|
||||
stopReason: budget.stopReason,
|
||||
imported,
|
||||
collectionsDone: done.size,
|
||||
|
||||
@@ -23,7 +23,11 @@
|
||||
* 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`).
|
||||
* ceiling: `NODE_OPTIONS=--max-old-space-size=524`);
|
||||
* - and, in a second run over two years, a Google that refuses the resume
|
||||
* edge's upper bound (`<=`): the backfill falls back to reading without
|
||||
* it, notes the refusal so it is met only once, and still imports every
|
||||
* reading exactly once.
|
||||
*
|
||||
* 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
|
||||
@@ -68,6 +72,11 @@ const fake = vi.hoisted(() => ({
|
||||
/** Faults to inject, keyed by the backfill request number they hit. */
|
||||
faults: new Map<number, "503" | "timeout" | "429-long">(),
|
||||
onRequest: null as (() => void) | null,
|
||||
/** Refuse any filter with an upper bound, and count the refusals. */
|
||||
refuseUpperBound: false,
|
||||
refused: 0,
|
||||
/** Backfill requests that carried an upper bound. */
|
||||
upperBoundRequests: 0,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
@@ -93,6 +102,22 @@ vi.mock("@/lib/safe-fetch", async (importOriginal) => {
|
||||
const filter = u.searchParams.get("filter") ?? "";
|
||||
const fromBackfill = !filter.includes(">=");
|
||||
if (fromBackfill) fake.backfillRequests += 1;
|
||||
if (filter.includes("<=")) {
|
||||
fake.upperBoundRequests += 1;
|
||||
if (fake.refuseUpperBound) {
|
||||
fake.refused += 1;
|
||||
return respond(
|
||||
{
|
||||
error: {
|
||||
code: 400,
|
||||
status: "INVALID_ARGUMENT",
|
||||
message: "Invalid filter expression.",
|
||||
},
|
||||
},
|
||||
400,
|
||||
);
|
||||
}
|
||||
}
|
||||
const fault = fromBackfill
|
||||
? fake.faults.get(fake.backfillRequests)
|
||||
: undefined;
|
||||
@@ -197,7 +222,8 @@ function report(line: string): void {
|
||||
process.stderr.write(`[gh-multi-year] ${line}\n`);
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
/** A fresh account with `points` minutes of heart rate behind it. */
|
||||
async function seed(points: number): Promise<void> {
|
||||
const prisma = getPrismaClient();
|
||||
await truncateAllTables(prisma);
|
||||
const { encrypt } = await import("@/lib/crypto");
|
||||
@@ -214,7 +240,19 @@ beforeAll(async () => {
|
||||
},
|
||||
});
|
||||
fake.now = NOW;
|
||||
fake.total = POINTS;
|
||||
fake.total = points;
|
||||
fake.heartRateRequests = 0;
|
||||
fake.backfillRequests = 0;
|
||||
fake.servedPoints = 0;
|
||||
fake.faults.clear();
|
||||
fake.onRequest = null;
|
||||
fake.refuseUpperBound = false;
|
||||
fake.refused = 0;
|
||||
fake.upperBoundRequests = 0;
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
await seed(POINTS);
|
||||
}, 120_000);
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -393,4 +431,106 @@ describe("Google Health backfill over three years of heart rate (#1220)", () =>
|
||||
expect(samples).toBeGreaterThan(5);
|
||||
expect(growth).toBeLessThan(GROWTH_BUDGET_BYTES);
|
||||
}, 3_600_000);
|
||||
|
||||
it("completes when Google refuses the resume edge's upper bound, meeting the refusal once", async () => {
|
||||
// Two years: the first slice reads without an edge, the second resumes
|
||||
// with the bound and is refused, the third reads without it from the start.
|
||||
const points = 2 * 365 * 24 * 60;
|
||||
await seed(points);
|
||||
fake.refuseUpperBound = true;
|
||||
|
||||
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");
|
||||
GOOGLE_HEALTH_TRANSIENT_RETRY.baseDelayMs = 0;
|
||||
|
||||
const prisma = getPrismaClient();
|
||||
let payload: AdmissionPayload | null = {
|
||||
kind: "google-health-backfill",
|
||||
data: { userId: USER, enqueuedAt: new Date().toISOString() },
|
||||
};
|
||||
const slices: Array<{
|
||||
imported: number;
|
||||
continued: number;
|
||||
skippedPages: number;
|
||||
refusalNoted: unknown;
|
||||
}> = [];
|
||||
|
||||
for (let n = 0; payload && n < 40; n++) {
|
||||
sent.jobs = [];
|
||||
logged.events = [];
|
||||
const outcome = await drainIntegrationBackfillAdmission([
|
||||
admissionJob(payload, n),
|
||||
]);
|
||||
expect(outcome.ok).toBe(true);
|
||||
const did = (outcome as { did: Record<string, number> }).did;
|
||||
const runEvents = logged.events.filter(
|
||||
(e) => e.background?.task_name === "job.integration_backfill",
|
||||
);
|
||||
expect(runEvents).toHaveLength(1);
|
||||
const meta = runEvents[0]!.meta ?? {};
|
||||
const sliceMeta = meta["googleHealth.backfill.slice"] as {
|
||||
pages: number;
|
||||
skippedPages: number;
|
||||
stopReason: string | null;
|
||||
};
|
||||
slices.push({
|
||||
imported: did.imported,
|
||||
continued: did.continued,
|
||||
skippedPages: sliceMeta.skippedPages,
|
||||
refusalNoted: meta["googleHealth.backfill.edgeFilterUnsupported"],
|
||||
});
|
||||
report(
|
||||
`refusing slice ${payload.data.slice ?? 0}: pages=${sliceMeta.pages} ` +
|
||||
`skipped=${sliceMeta.skippedPages} stop=${sliceMeta.stopReason} ` +
|
||||
`imported=${did.imported}`,
|
||||
);
|
||||
payload = nextSlice();
|
||||
if (payload) {
|
||||
const hourly = await handleGoogleHealthSync([] as never);
|
||||
expect(hourly).toMatchObject({
|
||||
ok: true,
|
||||
did: { users_skipped: 0, users_failed: 0, users_parked: 0 },
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
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();
|
||||
expect(slices.slice(0, -1).every((s) => s.continued === 1)).toBe(true);
|
||||
|
||||
// The bound was asked for once, refused once, and never again: the
|
||||
// refusal was noted on the slice that met it and honoured after.
|
||||
expect(fake.upperBoundRequests).toBe(1);
|
||||
expect(fake.refused).toBe(1);
|
||||
expect(slices.filter((s) => s.refusalNoted !== undefined)).toEqual([
|
||||
expect.objectContaining({ refusalNoted: ["fetchHeartRate"] }),
|
||||
]);
|
||||
// More than one slice had to pass over what the ones before it wrote,
|
||||
// and each still wrote its share.
|
||||
const skipping = slices.filter((s) => s.skippedPages > 0);
|
||||
expect(skipping.length).toBeGreaterThan(1);
|
||||
expect(skipping.every((s) => s.imported > 0)).toBe(true);
|
||||
|
||||
// Every reading exactly once, none of them rewritten.
|
||||
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);
|
||||
}, 3_600_000);
|
||||
});
|
||||
Reference in new issue
Block a user