diff --git a/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts new file mode 100644 index 000000000..35300709e --- /dev/null +++ b/src/app/api/measurements/batch/__tests__/hr-bucket-observability.test.ts @@ -0,0 +1,269 @@ +/** + * v1.32.1 (issue #585) — wide-event observability for the go-forward + * aggregated 10-min heart-rate bucket wire contract. + * + * The intraday pulse chart reads exactly the `stats:HKQuantityTypeIdentifier + * HeartRate:<10-min-Z>` rows this batch route accepts (or rejects). A report + * of "the chart looks sparser after the ZIP-import cutoff" was previously + * undiagnosable from server logs: the baseline `measurement.batch.ingest` + * annotation carried only a total `skipped` count with no reason breakdown + * and no visibility into how many of a batch's entries specifically targeted + * an HR bucket. These tests pin the new `measurement.batch.hr-bucket` + * annotation (fires only when the batch actually carried an HR-bucket entry) + * and the `skipped_by_reason` breakdown on the baseline annotation. + */ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { NextRequest } from "next/server"; + +const mocks = vi.hoisted(() => ({ annotate: vi.fn() })); + +vi.mock("@/lib/db", () => ({ + prisma: { + user: { + update: vi.fn(), + }, + measurement: { + findMany: vi.fn(), + createManyAndReturn: vi.fn(), + updateMany: vi.fn(), + }, + $transaction: vi.fn(async (fn: unknown) => { + if (typeof fn === "function") { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return (fn as any)(prisma as unknown as { measurement: unknown }); + } + }), + }, +})); + +vi.mock("@/lib/logging/context", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, annotate: mocks.annotate }; +}); + +vi.mock("@/lib/measurements/reconcile-external-measurement", () => ({ + reconcileExternalMeasurement: async ( + tx: { + measurement: { + findMany: (args: unknown) => Promise< + Array<{ + id?: string; + type?: string; + source?: string; + externalId?: string | null; + }> + >; + createManyAndReturn: (args: { + data: Record; + }) => Promise>>; + updateMany: (args: { + where: Record; + data: Record; + }) => Promise; + }; + }, + input: Record & { + userId: string; + type: string; + source: string; + externalId: string; + }, + options: { exactExternalMatch?: "update" | "duplicate" } = {}, + ) => { + const existing = await tx.measurement.findMany({}); + const exact = existing.find( + (row: { type?: string; source?: string; externalId?: string | null }) => + row.type === input.type && + row.source === input.source && + row.externalId === input.externalId, + ); + if (exact && options.exactExternalMatch !== "update") { + return { status: "duplicate", row: exact }; + } + if (exact) { + await tx.measurement.updateMany({ + where: { id: exact.id }, + data: { ...input, deletedAt: null }, + }); + return { status: "updated", row: { ...exact, ...input } }; + } + const [inserted] = await tx.measurement.createManyAndReturn({ + data: input, + }); + if (inserted) return { status: "inserted", row: inserted }; + if (options.exactExternalMatch === "update") { + await tx.measurement.updateMany({ where: {}, data: input }); + return { status: "updated", row: input }; + } + return { status: "duplicate" }; + }, +})); + +vi.mock("@/lib/auth/session", () => ({ getSession: vi.fn() })); +vi.mock("@/lib/auth/audit", () => ({ + auditLog: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/logging/transports", () => ({ emitIfSampled: vi.fn() })); +vi.mock("@/lib/db-compat", () => ({ + ensureDbCompatibility: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rate-limit", () => ({ checkRateLimit: vi.fn() })); +vi.mock("@/lib/jobs/pr-detection", () => ({ + enqueuePrDetection: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/jobs/reminder-satisfy", () => ({ + enqueueReminderSatisfy: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/cache/invalidate", () => ({ + invalidateUserMeasurements: vi.fn(), +})); +vi.mock("@/lib/insights/comprehensive-generate", () => ({ + invalidateStatusInsightsForTypes: vi.fn().mockResolvedValue(undefined), +})); +vi.mock("@/lib/rollups/measurement-rollups", () => ({ + recomputeBucketsForMeasurement: vi.fn().mockResolvedValue(undefined), + collapseToTypeDayKeys: vi.fn(() => []), +})); + +vi.mock("next/headers", () => ({ + headers: vi.fn(async () => ({ get: () => null })), + cookies: vi.fn(async () => ({ + get: () => undefined, + set: () => {}, + delete: () => {}, + })), +})); + +import { POST } from "../route"; +import { prisma } from "@/lib/db"; +import { getSession } from "@/lib/auth/session"; +import { checkRateLimit } from "@/lib/rate-limit"; + +const SESSION_OK = { + session: { id: "sess-1", expiresAt: new Date(Date.now() + 3_600_000) }, + user: { id: "user-1", username: "testuser", role: "USER" as const }, +}; + +const FRESH_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:00:00.000Z"; +const REPOST_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T15:00:00.000Z"; +const MALFORMED_BUCKET_ID = + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-21T14:35:00.000Z"; +const SAMPLE_ID = "C0FFEE00-0000-4000-8000-000000000000"; + +function makeRequest(body: unknown): NextRequest { + return new NextRequest("http://localhost/api/measurements/batch", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }); +} + +function hrEntry(externalId: string, value: number) { + return { + hkIdentifier: "HKQuantityTypeIdentifierHeartRate", + value, + unit: "count/min", + startDate: "2026-06-21T14:00:00.000Z", + endDate: "2026-06-21T14:59:59.000Z", + externalId, + }; +} + +function annotateCallsFor(actionName: string) { + return mocks.annotate.mock.calls.filter( + ([arg]) => + (arg as { action?: { name?: string } })?.action?.name === actionName, + ); +} + +beforeEach(() => { + vi.clearAllMocks(); + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + vi.mocked(checkRateLimit).mockResolvedValue({ + allowed: true, + remaining: 60, + resetAt: Date.now() + 60_000, + }); + vi.mocked(prisma.measurement.createManyAndReturn).mockImplementation((async ( + args: unknown, + ) => { + const { data } = args as { + data: + | { type: unknown; source: string; externalId: string | null } + | Array<{ type: unknown; source: string; externalId: string | null }>; + }; + const rows = Array.isArray(data) ? data : [data]; + return rows.map(({ type, source, externalId }) => ({ + type, + source, + externalId, + })); + }) as never); + vi.mocked(prisma.measurement.updateMany).mockResolvedValue({ count: 1 }); +}); + +describe("POST /api/measurements/batch — HR-bucket observability (issue #585)", () => { + it("reports a per-outcome breakdown covering only HR-bucket entries", async () => { + // REPOST_BUCKET_ID already exists — the reconcile mock overwrites it. + vi.mocked(prisma.measurement.findMany).mockResolvedValue([ + { + type: "PULSE", + source: "APPLE_HEALTH", + externalId: REPOST_BUCKET_ID, + }, + ] as never); + + const res = await POST( + makeRequest({ + entries: [ + hrEntry(FRESH_BUCKET_ID, 72), // inserted + hrEntry(REPOST_BUCKET_ID, 78), // updated (overwrite) + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped (malformed) + hrEntry(SAMPLE_ID, 64), // NOT an HR bucket — must be excluded + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.hr-bucket"); + expect(calls).toHaveLength(1); + const meta = (calls[0][0] as { meta: Record }).meta; + expect(meta).toEqual({ + attempted: 3, + inserted: 1, + updated: 1, + duplicate: 0, + skipped: 1, + failed: 0, + processed: 4, + }); + }); + + it("never fires the HR-bucket annotation for a batch with no bucket entries", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST(makeRequest({ entries: [hrEntry(SAMPLE_ID, 64)] })); + expect(res.status).toBe(200); + expect(annotateCallsFor("measurement.batch.hr-bucket")).toHaveLength(0); + }); + + it("breaks the baseline ingest annotation's skip count down by reason", async () => { + vi.mocked(prisma.measurement.findMany).mockResolvedValue([]); + const res = await POST( + makeRequest({ + entries: [ + hrEntry(MALFORMED_BUCKET_ID, 70), // skipped: malformed_hr_bucket_id + ], + }), + ); + expect(res.status).toBe(200); + + const calls = annotateCallsFor("measurement.batch.ingest"); + expect(calls).toHaveLength(1); + const meta = calls[0][0] as { meta: { skipped_by_reason: unknown } }; + expect(meta.meta.skipped_by_reason).toEqual({ + malformed_hr_bucket_id: 1, + }); + }); +}); diff --git a/src/app/api/measurements/batch/route.ts b/src/app/api/measurements/batch/route.ts index b739ed6b4..a48cf1c66 100644 --- a/src/app/api/measurements/batch/route.ts +++ b/src/app/api/measurements/batch/route.ts @@ -603,6 +603,18 @@ async function postBatch(request: NextRequest): Promise { .filter((r) => r.status === "skipped") .map((r) => ({ index: r.index, reason: r.reason ?? "unknown" })); + // v1.32.1 (issue #585) — per-reason skip breakdown. The baseline + // annotation below only ever carried the total `skipped` count, so an + // operator investigating "why does this account's uploaded data look + // sparser than expected" had no way to see WHICH validation gate an + // entry tripped (malformed bucket id, out-of-range value, unmappable + // identifier, …) without pulling raw rows. Grouping by reason turns + // that into a one-glance wide-event field. + const skippedByReason: Record = {}; + for (const s of skipped) { + skippedByReason[s.reason] = (skippedByReason[s.reason] ?? 0) + 1; + } + await auditLog("measurement.batch.ingest", { userId: user.id, ipAddress: getClientIp(request), @@ -648,6 +660,64 @@ async function postBatch(request: NextRequest): Promise { }); } + // v1.32.1 (issue #585) — dedicated outcome breakdown for entries + // targeting the go-forward aggregated 10-min heart-rate bucket prefix + // (`stats:HKQuantityTypeIdentifierHeartRate:<10-min-Z>`). The intraday + // pulse chart reads exactly these rows for a live-synced day; a report + // of "the chart looks sparser after the ZIP import cutoff" is + // unanswerable from the baseline ingest trace alone, since it never + // distinguished an HR-bucket write from any other entry in the batch. + // Grepping `measurement.batch.hr-bucket` now shows, per sync, how many + // bucket entries the client sent and what happened to each — confirming + // (or ruling out) server-side rejection/dedup as the density-gap cause. + // Only fires when the batch actually carried at least one HR-bucket + // entry so the baseline trace is unchanged for every other sync. + let hrBucketInserted = 0; + let hrBucketUpdated = 0; + let hrBucketDuplicate = 0; + let hrBucketSkipped = 0; + let hrBucketFailed = 0; + for (let index = 0; index < entries.length; index++) { + if (!targetsAggregatedBucket(entries[index].externalId)) continue; + switch (results[index]?.status) { + case "inserted": + hrBucketInserted++; + break; + case "updated": + hrBucketUpdated++; + break; + case "duplicate": + hrBucketDuplicate++; + break; + case "skipped": + hrBucketSkipped++; + break; + case "failed": + hrBucketFailed++; + break; + } + } + const hrBucketAttempted = + hrBucketInserted + + hrBucketUpdated + + hrBucketDuplicate + + hrBucketSkipped + + hrBucketFailed; + if (hrBucketAttempted > 0) { + annotate({ + action: { name: "measurement.batch.hr-bucket" }, + meta: { + attempted: hrBucketAttempted, + inserted: hrBucketInserted, + updated: hrBucketUpdated, + duplicate: hrBucketDuplicate, + skipped: hrBucketSkipped, + failed: hrBucketFailed, + processed: entries.length, + }, + }); + } + // v1.4.25 W16c — kick off PR detection for this user. We always // enqueue when at least one row was written (or the batch had any // measurements to consider) so a single off-day reading still gets @@ -693,6 +763,9 @@ async function postBatch(request: NextRequest): Promise { updated: updatedCount, duplicates: duplicateCount, skipped: skipped.length, + // v1.32.1 (issue #585) — per-reason skip breakdown; see the + // `skippedByReason` build above. + skipped_by_reason: skippedByReason, failed: failedCount, }, }); diff --git a/src/app/insights/pulse/__tests__/page.test.tsx b/src/app/insights/pulse/__tests__/page.test.tsx new file mode 100644 index 000000000..13e2432ff --- /dev/null +++ b/src/app/insights/pulse/__tests__/page.test.tsx @@ -0,0 +1,144 @@ +import { describe, expect, it, vi, beforeEach } from "vitest"; +import { renderToStaticMarkup } from "react-dom/server"; + +/** + * v1.32.1 (issue #584) — `/insights/pulse` must chart the metric its own + * title, capture action, and stat strip claim. `hasRestingHr` used to swap + * the ENTIRE primary chart + Coach read strip to RESTING_HEART_RATE the + * moment any resting-HR row existed, so the page could show a resting-only + * series while every other surface (title, stat strip, capture action, + * "show all values") stayed labelled "Pulse" — a silent full swap. + * RESTING_HEART_RATE may now appear only as a second, clearly-labelled + * series alongside PULSE (the same multi-series pattern the blood-pressure + * page uses for systolic/diastolic) — never a swap. + * + * `SubPageShell` is stubbed to a probe recording its props (including the + * `children` array holding the chart element) so these assert the resolved + * chart/coach-strip props rather than rendered text — responsive classes + * have broken text-based queries in this repo. + */ + +const shellProps = vi.fn(); +vi.mock("@/components/insights/sub-page-shell", () => ({ + SubPageShell: (props: Record) => { + shellProps(props); + return null; + }, +})); + +vi.mock("@/lib/i18n/context", () => ({ + useTranslations: () => ({ + t: (key: string) => key, + }), +})); + +vi.mock("@/hooks/use-auth", () => ({ + useAuth: () => ({ + isAuthenticated: true, + user: { dateOfBirth: null, gender: null, timezone: "UTC" }, + }), +})); + +const analytics = { + current: { summaries: {} as Record }, +}; +vi.mock("@/hooks/use-insights-analytics", () => ({ + useInsightsAnalytics: () => ({ + data: analytics.current, + isLoading: false, + isEmpty: false, + error: null, + refetch: () => {}, + }), +})); + +// `useInsightsLayoutPrefs` is the one real hook left that reaches +// `@tanstack/react-query`; stub `useQuery` so the page renders without a +// `QueryClientProvider` ancestor (mirrors the coach-page-launch-params test). +vi.mock("@tanstack/react-query", () => ({ + useQuery: () => ({ data: undefined, isLoading: false, error: null }), +})); + +import InsightsPulsPage from "../page"; + +interface CapturedShellProps { + captureType?: string; + showAllValuesType?: string; + coachReadStrip?: { props?: Record }; + children?: unknown; +} + +function renderPage( + summaries: Record, +): CapturedShellProps { + analytics.current = { summaries }; + renderToStaticMarkup(); + const calls = shellProps.mock.calls; + return calls[calls.length - 1][0] as CapturedShellProps; +} + +/** Locate the primary `` element among the page's + * children by its distinctive `types` array prop. Never rendered — the + * `SubPageShell` stub does not paint `children` — so this reads the plain + * React element object the page constructed. */ +function findChartElement( + children: unknown, +): { props: { types?: string[] } } | undefined { + const arr = Array.isArray(children) ? children : [children]; + return arr.find( + (el): el is { props: { types?: string[] } } => + Boolean(el) && + typeof el === "object" && + el !== null && + "props" in el && + Array.isArray((el as { props?: { types?: unknown } }).props?.types), + ); +} + +describe("/insights/pulse page — chart identity (issue #584)", () => { + beforeEach(() => { + shellProps.mockClear(); + }); + + it("charts PULSE as the primary (first) series even when resting-HR data exists", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + const chart = findChartElement(props.children); + expect(chart).toBeDefined(); + expect(chart!.props.types).toEqual(["PULSE", "RESTING_HEART_RATE"]); + }); + + it("never swaps the chart to a RESTING_HEART_RATE-only series", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + const chart = findChartElement(props.children); + expect(chart!.props.types).not.toEqual(["RESTING_HEART_RATE"]); + }); + + it("charts PULSE alone when there is no resting-HR data", () => { + const props = renderPage({ PULSE: { count: 50 } }); + const chart = findChartElement(props.children); + expect(chart!.props.types).toEqual(["PULSE"]); + }); + + it("keeps the Coach read strip pinned to PULSE regardless of resting-HR data", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + expect(props.coachReadStrip?.props?.metricType).toBe("PULSE"); + }); + + it("keeps captureType and showAllValuesType pinned to PULSE", () => { + const props = renderPage({ + PULSE: { count: 50 }, + RESTING_HEART_RATE: { count: 30 }, + }); + expect(props.captureType).toBe("PULSE"); + expect(props.showAllValuesType).toBe("PULSE"); + }); +}); diff --git a/src/app/insights/pulse/page.tsx b/src/app/insights/pulse/page.tsx index 0b1b1d313..1f63ab298 100644 --- a/src/app/insights/pulse/page.tsx +++ b/src/app/insights/pulse/page.tsx @@ -35,20 +35,23 @@ const IntradayPulseChart = dynamic( })), { ssr: false, loading: () => }, ); -import { - getAgeFromDateOfBirth, - getPersonalizedPulseTarget, -} from "@/lib/analytics/pulse-targets"; /** * v1.4.25 W4 — `/insights/pulse`. * - * Routed Pulse sub-page. Renders the pulse chart with the personalized - * Karvonen-derived target band plus the per-section AI assessment. - * Note: `chartKey="pulse"` so the chart-cog can override the - * comparison-overlay independently from the dashboard pulse card; the - * MeasurementType filter is `PULSE` (the same field used elsewhere - * in the codebase). + * Routed Pulse sub-page. Renders the raw/workout-inclusive pulse chart + * plus the per-section AI assessment. Note: `chartKey="pulse"` so the + * chart-cog can override the comparison-overlay independently from the + * dashboard pulse card; the MeasurementType filter is `PULSE` (the same + * field used elsewhere in the codebase). + * + * v1.32.1 (issue #584) — this page no longer paints the personalized + * resting-pulse target band (`getPersonalizedPulseTarget()`, CDC/NCHS + * resting-pulse percentiles) behind the chart: that band is calibrated + * to steady-state resting readings and would misreport an expected + * workout spike in the raw PULSE series as "out of target". The + * personal target + its band belong on the dedicated + * `/insights/resting-pulse` page, against the clean resting series. * * Analytics fetch + empty-state branch consume `useInsightsAnalytics()` * + ``. VO₂ max is a cardio-fitness metric, so the page @@ -66,11 +69,16 @@ export default function InsightsPulsPage() { // dedicated cardio-fitness page. const hasVo2 = (analytics?.summaries?.VO2_MAX?.count ?? 0) > 0; const pulseSummary = analytics?.summaries?.PULSE ?? null; - // v1.15.12 A2 — the resting-pulse band is judged against - // RESTING_HEART_RATE (Apple's clean daily resting figure). When the - // user has resting rows the chart shows that series against the band; - // otherwise it charts raw heart rate WITHOUT the resting-band overlay - // (which would flag expected-high workout HR as "outside target"). + // v1.32.1 (issue #584) — this page is titled, captured, and stat-stripped + // as PULSE throughout; the chart must chart PULSE too. `hasRestingHr` used + // to swap the ENTIRE primary chart + Coach read to RESTING_HEART_RATE + // whenever any resting row existed, which silently showed a different + // metric than the page's own title/stat-strip/capture action claimed. + // RESTING_HEART_RATE now only ever appears as a second, clearly-labelled + // series alongside PULSE (same multi-series pattern the blood-pressure + // page uses for systolic/diastolic) — never a full swap. The dedicated + // `/insights/resting-pulse` page owns the resting-only view + its target + // band. const hasRestingHr = (analytics?.summaries?.RESTING_HEART_RATE?.count ?? 0) > 0; @@ -108,44 +116,6 @@ export default function InsightsPulsPage() { ); } - const pulseAge = getAgeFromDateOfBirth(user?.dateOfBirth ?? null); - const pulseTarget = getPersonalizedPulseTarget( - pulseAge, - (user?.gender as "MALE" | "FEMALE" | null | undefined) ?? null, - ); - const pulseBands = [ - { - min: 30, - max: pulseTarget.orangeMin, - color: "var(--destructive)", - opacity: 0.16, - }, - { - min: pulseTarget.orangeMin, - max: pulseTarget.greenMin, - color: "var(--warning)", - opacity: 0.18, - }, - { - min: pulseTarget.greenMin, - max: pulseTarget.greenMax, - color: "var(--success)", - opacity: 0.2, - }, - { - min: pulseTarget.greenMax, - max: pulseTarget.orangeMax, - color: "var(--warning)", - opacity: 0.18, - }, - { - min: pulseTarget.orangeMax, - max: 220, - color: "var(--destructive)", - opacity: 0.16, - }, - ].filter((band) => band.max > band.min); - return ( } coachReadStrip={ - + } diversityNudge={ ({ annotateMock: vi.fn() })); vi.mock("@/lib/db", () => ({ prisma: { @@ -21,6 +22,8 @@ vi.mock("@/lib/db", () => ({ }, })); +vi.mock("@/lib/logging/context", () => ({ annotate: annotateMock })); + import { loadIntradayPulse } from "@/lib/analytics/intraday-pulse-io"; import { resolveRestingPulseSeries, @@ -42,6 +45,7 @@ describe("loadIntradayPulse", () => { findManyMock.mockReset(); workoutFindManyMock.mockReset(); workoutFindManyMock.mockResolvedValue([]); + annotateMock.mockReset(); }); it("M2 — queries PULSE/step rows against the exact local-day window, not the padded lead/trail superset", async () => { @@ -209,4 +213,59 @@ describe("loadIntradayPulse", () => { { startMinute: 490, mean: 75, count: 2, min: 70, max: 81 }, ]); }); + + it("v1.32.1 (issue #585) — annotates the per-shape row/bucket breakdown for observability", async () => { + const tz = "UTC"; + const dateKey = "2026-06-15"; + + const restingRows = Array.from({ length: 20 }, (_, i) => ({ + value: 55, + measuredAt: new Date( + `2026-05-${String((i % 28) + 1).padStart(2, "0")}T06:00:00.000Z`, + ), + })); + // One uploaded 10-min bucket row and one raw per-sample row sharing the + // day — the DTO merges them (bucket overlays raw on a slot collision), + // but the annotation must still report each SHAPE's own row count, not + // just the merged bucket total. + const uploadedBucket = { + value: 72, + valueMin: 64, + valueMax: 88, + measuredAt: new Date("2026-06-15T08:09:00.000Z"), + externalId: + "stats:HKQuantityTypeIdentifierHeartRate:2026-06-15T08:00:00.000Z", + }; + const rawSample = { + value: 65, + valueMin: null, + valueMax: null, + measuredAt: new Date("2026-06-15T09:05:00.000Z"), + externalId: "sample:abc123", + }; + + findManyMock.mockImplementation( + async (args: { + where: { type: string; measuredAt?: unknown }; + take?: number; + }) => { + if (args.where.type === "RESTING_HEART_RATE") return restingRows; + if (args.where.type === "PULSE" && args.where.measuredAt) { + return [uploadedBucket, rawSample]; + } + return []; + }, + ); + + await loadIntradayPulse("user-1", tz, dateKey); + + expect(annotateMock).toHaveBeenCalledWith({ + meta: { + intraday_raw_sample_rows: 1, + intraday_uploaded_bucket_rows: 1, + intraday_folded_hourly_rows: 0, + intraday_ten_min_buckets: 2, + }, + }); + }); }); diff --git a/src/lib/analytics/intraday-pulse-io.ts b/src/lib/analytics/intraday-pulse-io.ts index 9a4df4243..ec7987b34 100644 --- a/src/lib/analytics/intraday-pulse-io.ts +++ b/src/lib/analytics/intraday-pulse-io.ts @@ -16,6 +16,7 @@ * digest's `tension_window` builder, so the signal is computed one way. */ import { prisma } from "@/lib/db"; +import { annotate } from "@/lib/logging/context"; import { percentile } from "@/lib/insights/strain-score"; import { resolveRestingPulseSeries } from "@/lib/analytics/resting-pulse"; import { localDayWindow } from "@/lib/measurements/consolidation-tz"; @@ -349,6 +350,26 @@ export async function loadIntradayPulse( }) : null; + // v1.32.1 (issue #585) — per-shape row/bucket breakdown. A report of + // "the chart looks sparser after the ZIP-import cutoff" is otherwise + // undiagnosable from server logs: the route's own annotation only ever + // surfaced the FINAL bucket count, with no visibility into which + // ingestion shape (raw ZIP-imported samples vs uploaded 10-min + // aggregates) the day's data actually came from, nor how many DB rows + // fed each shape before bucketing. Grepping `insights.pulse.intraday` + // now shows both — e.g. `uploaded_bucket_rows: 6` next to + // `ten_min_buckets: 6` confirms the uploaded shape IS being read and + // placed 1:1, narrowing a future report to "the client uploaded too few + // buckets" rather than leaving the read path itself as a live suspect. + annotate({ + meta: { + intraday_raw_sample_rows: rawPulseRows.length, + intraday_uploaded_bucket_rows: uploadedBucketRows.length, + intraday_folded_hourly_rows: foldedPulseRows.length, + intraday_ten_min_buckets: tenMinSeries.length, + }, + }); + return { dateKey, timezone, diff --git a/src/lib/import/__tests__/unzip-export-xml.test.ts b/src/lib/import/__tests__/unzip-export-xml.test.ts index 3151558c5..3863f79b8 100644 --- a/src/lib/import/__tests__/unzip-export-xml.test.ts +++ b/src/lib/import/__tests__/unzip-export-xml.test.ts @@ -3,16 +3,27 @@ import { writeFileSync, mkdtempSync, readFileSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { deflateRawSync } from "node:zlib"; +import { pipeline } from "node:stream/promises"; +import { Readable } from "node:stream"; -import { extractExportXml, readCentralDirectory } from "../unzip-export-xml"; +import { + createByteCap, + extractExportXml, + readCentralDirectory, +} from "../unzip-export-xml"; /** - * Hand-build a single-entry deflate-compressed ZIP archive with the - * supplied filename + payload. Enough to exercise the central-directory - * walker + extractor without pulling in a full zip library. + * Hand-build a single-entry ZIP archive with the supplied filename + + * payload. Enough to exercise the central-directory walker + extractor + * without pulling in a full zip library. `method` selects compression + * method 8 (deflate, default) or 0 (stored — bytes copied verbatim). */ -function buildMinimalZip(filename: string, payload: Buffer): Buffer { - const compressed = deflateRawSync(payload); +function buildMinimalZip( + filename: string, + payload: Buffer, + method: 0 | 8 = 8, +): Buffer { + const compressed = method === 8 ? deflateRawSync(payload) : payload; const crc32 = (() => { // Pre-computed lookup table for CRC-32. node:zlib has crc32 in 22+ // but we keep the function inline to stay version-agnostic. @@ -38,7 +49,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer { localHeader.writeUInt32LE(0x04034b50, 0); localHeader.writeUInt16LE(20, 4); // version localHeader.writeUInt16LE(0, 6); // flags - localHeader.writeUInt16LE(8, 8); // method: deflate + localHeader.writeUInt16LE(method, 8); // method localHeader.writeUInt16LE(0, 10); // time localHeader.writeUInt16LE(0, 12); // date localHeader.writeUInt32LE(crc32, 14); @@ -53,7 +64,7 @@ function buildMinimalZip(filename: string, payload: Buffer): Buffer { cdh.writeUInt16LE(20, 4); // version made by cdh.writeUInt16LE(20, 6); // version needed cdh.writeUInt16LE(0, 8); // flags - cdh.writeUInt16LE(8, 10); // method + cdh.writeUInt16LE(method, 10); // method cdh.writeUInt16LE(0, 12); // time cdh.writeUInt16LE(0, 14); // date cdh.writeUInt32LE(crc32, 16); @@ -95,7 +106,7 @@ describe("readCentralDirectory", () => { }); describe("extractExportXml", () => { - it("extracts the export.xml member to a temp file", () => { + it("extracts a deflated export.xml member to a temp file", async () => { const payload = Buffer.from( ``, ); @@ -104,7 +115,7 @@ describe("extractExportXml", () => { const zipPath = join(tmp, "export.zip"); writeFileSync(zipPath, zip); - const out = extractExportXml(zipPath); + const out = await extractExportXml(zipPath); expect(out.xmlBytes).toBe(payload.length); expect(out.otherMembers).toHaveLength(0); @@ -112,13 +123,65 @@ describe("extractExportXml", () => { expect(extracted.toString("utf8")).toBe(payload.toString("utf8")); }); - it("throws when the export.xml member is missing", () => { + // v1.32.1 (issue #588) — the stored (uncompressed) path now runs + // through the same streaming pipeline as deflate; a stored member + // must still round-trip byte-exact through it. + it("extracts a stored (uncompressed) export.xml member to a temp file", async () => { + const payload = Buffer.from( + ``, + ); + const zip = buildMinimalZip("apple_health_export/export.xml", payload, 0); + const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-")); + const zipPath = join(tmp, "export.zip"); + writeFileSync(zipPath, zip); + + const out = await extractExportXml(zipPath); + expect(out.xmlBytes).toBe(payload.length); + + const extracted = readFileSync(out.xmlPath); + expect(extracted.toString("utf8")).toBe(payload.toString("utf8")); + }); + + it("throws when the export.xml member is missing", async () => { const payload = Buffer.from("noop"); const zip = buildMinimalZip("other-file.txt", payload); const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-")); const zipPath = join(tmp, "export.zip"); writeFileSync(zipPath, zip); - expect(() => extractExportXml(zipPath)).toThrow(/missing/); + await expect(extractExportXml(zipPath)).rejects.toThrow(/missing/); + }); +}); + +// v1.32.1 (issue #588) — the streaming inflate Transform has no +// `maxOutputLength` option, so the zip-bomb defence moved to a +// hand-rolled byte-counting stage. Exercise it directly at a tiny +// limit; the real 8 GiB ceiling is not practical to trip from a +// fixture. +describe("createByteCap", () => { + it("passes bytes through unchanged while under the limit", async () => { + const chunks: Buffer[] = []; + await pipeline( + Readable.from([Buffer.from("hello"), Buffer.from("world")]), + createByteCap(10), + async function collect(source) { + for await (const chunk of source) chunks.push(chunk as Buffer); + }, + ); + expect(Buffer.concat(chunks).toString("utf8")).toBe("helloworld"); + }); + + it("rejects the stream once the running total exceeds the limit", async () => { + await expect( + pipeline( + Readable.from([Buffer.from("hello"), Buffer.from("world")]), + createByteCap(6), + async function drain(source) { + for await (const chunk of source) { + void chunk; + } + }, + ), + ).rejects.toThrow(/exceeds the 6-byte cap/); }); }); diff --git a/src/lib/import/unzip-export-xml.ts b/src/lib/import/unzip-export-xml.ts index 7333f2d09..6379ce0ec 100644 --- a/src/lib/import/unzip-export-xml.ts +++ b/src/lib/import/unzip-export-xml.ts @@ -7,21 +7,49 @@ * deflate-compressed ZIP archive. Pulling in `yauzl` / `adm-zip` / etc. * for one file we only ever read once would balloon the build graph * (and `yauzl` ships a non-trivial number of transitive deps). Node 22 - * ships `node:zlib` `inflateRawSync` already; the only missing piece - * is a tiny ZIP central-directory walker. + * ships `node:zlib` already; the only missing piece is a tiny ZIP + * central-directory walker. * * Coverage scope: * - Stored entries (compression method 0) — copy bytes verbatim. - * - Deflated entries (compression method 8) — `inflateRawSync`. + * - Deflated entries (compression method 8) — streamed through + * `zlib.createInflateRaw()`. * - Encrypted entries → unsupported (Apple does not encrypt the * export.zip; reject with a clear error). * - Zip64 — supported for the central directory record (Apple * exports easily push past the 4 GB limit on multi-year accounts). * + * v1.32.1 (issue #588) — extracting the member used to run through + * `inflateRawSync()` on a single in-memory buffer holding the WHOLE + * inflated `export.xml` (up to `MAX_DECOMPRESSED_BYTES`) at the same + * time the compressed-archive buffer was still resident, then + * `writeFileSync()`'d the result in one blocking call. `extractExportXml` + * runs on the same event loop as the web server in the default + * single-container deployment (`src/instrumentation.ts` starts the + * pg-boss worker in-process), so that synchronous path could pin the + * Node main thread for minutes on a real multi-year export — freezing + * every other request the app was serving — while briefly needing + * roughly 2x the decompressed size in JS heap. A user hitting this saw + * an import that never leaves "Unpacking the archive…": the phase + * label is written to `ImportJob.status` BEFORE extraction starts and + * only advances once `extractExportXml()` returns, so a process that + * dies mid-extraction (OOM, restart) before that return leaves the row + * silently stuck with no failure ever recorded. The member now streams + * through `zlib.createInflateRaw()` into a file write stream — + * decompression work moves off the JS event loop onto the libuv + * threadpool in bounded chunks, and peak memory is bounded by the + * chunk size rather than the whole decompressed payload. The + * `MAX_DECOMPRESSED_BYTES` zip-bomb defence moves with it: the + * streaming inflate API has no `maxOutputLength` option, so a + * byte-counting transform enforces the cap on the real output as it + * flows. + * * Locks per `.planning/research/v1434-r-1-xml-import.md` §6.1. */ -import { readFileSync, writeFileSync } from "node:fs"; -import { inflateRawSync } from "node:zlib"; +import { readFileSync, createWriteStream, statSync, unlinkSync } from "node:fs"; +import { createInflateRaw } from "node:zlib"; +import { Readable, Transform } from "node:stream"; +import { pipeline } from "node:stream/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -44,11 +72,15 @@ const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024 * 1024; * Pre-flight refusal threshold for the central-directory's advertised * ratio. Legitimate Apple Health XML compresses at maybe 10–20× under * DEFLATE; anything claiming a 200× expansion is a synthesized bomb. - * This is a coarse signal — the `maxOutputLength` ceiling on the - * inflate call below is the load-bearing defence. + * This is a coarse signal — the byte-counting cap on the streamed + * inflate output (`streamEntryToFile()` below) is the load-bearing + * defence. */ const MAX_COMPRESSION_RATIO = 200; +/** Chunk size fed into the inflate/write pipeline per `write()` call. */ +const STREAM_CHUNK_BYTES = 1 * 1024 * 1024; + /** A single entry within the archive's central directory. */ interface CentralDirectoryEntry { fileName: string; @@ -78,7 +110,9 @@ export interface UnzipResult { * Throws when the member is missing, encrypted, or compressed with * an unsupported method. */ -export function extractExportXml(archivePath: string): UnzipResult { +export async function extractExportXml( + archivePath: string, +): Promise { const buf = readFileSync(archivePath); const entries = readCentralDirectory(buf); @@ -105,8 +139,8 @@ export function extractExportXml(archivePath: string): UnzipResult { // Pre-flight zip-bomb defence. The central directory's advertised // `uncompressedSize` is attacker-controlled (a malicious archive can // lie), so this catches honest-but-oversized payloads early; the - // load-bearing defence is the `maxOutputLength` cap inside - // `extractEntry()` which trips on the actual inflate output. + // load-bearing defence is the byte-counting cap inside + // `streamEntryToFile()` which trips on the actual inflate output. if (exportXmlEntry.uncompressedSize > MAX_DECOMPRESSED_BYTES) { throw new Error( `export.xml declares an uncompressed size of ${exportXmlEntry.uncompressedSize} bytes` + @@ -126,13 +160,11 @@ export function extractExportXml(archivePath: string): UnzipResult { ); } - const xmlBytes = extractEntry(buf, exportXmlEntry); - const xmlPath = join( tmpdir(), `healthlog-import-${Date.now()}-${Math.random().toString(36).slice(2, 10)}.xml`, ); - writeFileSync(xmlPath, xmlBytes); + const xmlBytes = await streamEntryToFile(buf, exportXmlEntry, xmlPath); const otherMembers = entries .filter((e) => e !== exportXmlEntry) @@ -140,7 +172,7 @@ export function extractExportXml(archivePath: string): UnzipResult { return { xmlPath, - xmlBytes: xmlBytes.length, + xmlBytes, otherMembers, }; } @@ -260,8 +292,59 @@ export function readCentralDirectory(buf: Buffer): CentralDirectoryEntry[] { return entries; } -/** Extract the bytes of a single ZIP entry into a Buffer. */ -function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer { +/** + * Split a Buffer into fixed-size slices (no copying — `subarray` views + * share the backing memory) so the inflate/write pipeline processes the + * entry in bounded chunks instead of one giant `write()` call. + */ +function* chunkBuffer(buf: Buffer, size: number): Generator { + for (let offset = 0; offset < buf.length; offset += size) { + yield buf.subarray(offset, Math.min(offset + size, buf.length)); + } +} + +/** + * Byte-counting passthrough that refuses to forward more than `limit` + * bytes — the streaming equivalent of `inflateRawSync`'s + * `maxOutputLength`, since the Transform-based zlib API exposes no such + * option. Trips on the actual bytes flowing through even when the + * central-directory metadata lied about the uncompressed size. Exported + * so a unit test can exercise the cap directly against a small limit — + * the real `MAX_DECOMPRESSED_BYTES` ceiling (8 GiB) is not practical to + * trip from a test fixture. + */ +export function createByteCap(limit: number): Transform { + let total = 0; + return new Transform({ + transform(chunk: Buffer, _enc, callback) { + total += chunk.length; + if (total > limit) { + callback( + new Error( + `Decompressed export.xml exceeds the ${limit}-byte cap` + + " — refusing as a suspected zip bomb.", + ), + ); + return; + } + callback(null, chunk); + }, + }); +} + +/** + * Stream a single ZIP entry's bytes out to `destPath`, decompressing + * through `zlib.createInflateRaw()` when needed. Runs entirely off the + * JS main thread (stream I/O + the libuv-threadpool-backed zlib + * binding), so a multi-GB member no longer blocks the event loop the + * way the old `inflateRawSync()` + `writeFileSync()` pair did. Returns + * the number of bytes written. + */ +async function streamEntryToFile( + buf: Buffer, + entry: CentralDirectoryEntry, + destPath: string, +): Promise { const local = entry.localHeaderOffset; if (buf.readUInt32LE(local) !== LOCAL_FILE_HEADER) { throw new Error( @@ -271,18 +354,32 @@ function extractEntry(buf: Buffer, entry: CentralDirectoryEntry): Buffer { const localFileNameLen = buf.readUInt16LE(local + 26); const localExtraLen = buf.readUInt16LE(local + 28); const dataStart = local + 30 + localFileNameLen + localExtraLen; - const compressed = buf.slice(dataStart, dataStart + entry.compressedSize); + const compressed = buf.subarray(dataStart, dataStart + entry.compressedSize); - if (entry.compressionMethod === 0) { - return compressed; + const source = Readable.from(chunkBuffer(compressed, STREAM_CHUNK_BYTES)); + const dest = createWriteStream(destPath); + const cap = createByteCap(MAX_DECOMPRESSED_BYTES); + + try { + if (entry.compressionMethod === 0) { + await pipeline(source, cap, dest); + } else { + // Method 8 — DEFLATE, streamed through the async zlib Transform. + await pipeline(source, createInflateRaw(), cap, dest); + } + } catch (err) { + // A mid-stream failure (byte cap trip, corrupt deflate stream) can + // leave a partial file on disk — the old sync path only ever wrote + // once inflate had already fully succeeded. Clean up so a rejected + // archive doesn't leak a partial `/tmp` file. + try { + unlinkSync(destPath); + } catch { + // ignore — best-effort + } + throw err; } - // Method 8 — DEFLATE. `maxOutputLength` instructs `node:zlib` to - // refuse expansion past the cap with a thrown error, defeating - // zip-bomb amplification even when the central-directory metadata - // lies about the uncompressed size. - return inflateRawSync(compressed, { - maxOutputLength: MAX_DECOMPRESSED_BYTES, - }); + return statSync(destPath).size; } /** diff --git a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts new file mode 100644 index 000000000..13e240628 --- /dev/null +++ b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts @@ -0,0 +1,186 @@ +/** + * v1.32.1 (issue #588) — periodic orphan-`ImportJob` reconcile. + * + * Before this change `reconcileOrphanImportJobs()` only ran once, at + * worker boot. A worker that crashed mid-extraction (OOM, restart) and + * came back up BEFORE the stuck row's heartbeat went stale (30 min) or + * pg-boss stopped reporting the backing job as live left that row + * stuck in `unpacking` / `parsing` / `upserting` forever — an import + * that never leaves "Unpacking the archive… 0 rows imported" with no + * failure ever surfaced, because nothing re-evaluated the row after + * that one boot-time pass. These tests cover both the periodic-tick + * wiring (queue provisioned, cron scheduled, handler bound) and the + * reconcile logic itself (a genuinely-abandoned row flips to `failed`; + * a row another worker is still actively running does not). + */ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + findMany: vi.fn(), + updateMany: vi.fn(), + getGlobalBoss: vi.fn(), + getJobById: vi.fn(), +})); + +vi.mock("@/lib/db", () => ({ + prisma: { + importJob: { + findMany: mocks.findMany, + updateMany: mocks.updateMany, + }, + }, + toJson: (value: unknown) => value, +})); + +vi.mock("@/lib/jobs/boss-instance", () => ({ + getGlobalBoss: mocks.getGlobalBoss, +})); + +import { + APPLE_HEALTH_IMPORT_PARSER_REVISION, + APPLE_HEALTH_IMPORT_V2_QUEUE, + IMPORT_JOB_RECONCILE_CRON, + IMPORT_JOB_RECONCILE_QUEUE, + handleImportJobReconcileTick, + reconcileOrphanImportJobs, +} from "../apple-health-import-worker"; + +const maintenanceSource = readFileSync( + join(process.cwd(), "src/lib/jobs/reminder/register-maintenance.ts"), + "utf8", +); + +beforeEach(() => { + vi.resetAllMocks(); + mocks.getGlobalBoss.mockReturnValue({ getJobById: mocks.getJobById }); + mocks.updateMany.mockResolvedValue({ count: 0 }); +}); + +describe("periodic reconcile — wiring", () => { + it("provisions the queue, schedules a 15-minute cron, and binds the handler", () => { + expect(IMPORT_JOB_RECONCILE_QUEUE).toBe("apple-health-import-reconcile"); + expect(IMPORT_JOB_RECONCILE_CRON).toBe("*/15 * * * *"); + + const allQueues = maintenanceSource.match( + /const allQueues\s*=\s*\[([\s\S]*?)\];/, + ); + expect(allQueues).not.toBeNull(); + expect(allQueues![1]).toMatch(/\bIMPORT_JOB_RECONCILE_QUEUE\b/); + + const schedules = maintenanceSource.match( + /const schedules[\s\S]*?=\s*\[([\s\S]*?)\];/, + ); + expect(schedules).not.toBeNull(); + expect(schedules![1]).toMatch( + /\[IMPORT_JOB_RECONCILE_QUEUE,\s*IMPORT_JOB_RECONCILE_CRON\]/, + ); + + expect(maintenanceSource).toMatch( + /boss\.work\(\s*IMPORT_JOB_RECONCILE_QUEUE[\s\S]{0,120}handleImportJobReconcileTick/, + ); + }); +}); + +describe("reconcileOrphanImportJobs", () => { + it("flips a row to failed once its heartbeat has gone stale", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-stale", + pgBossJobId: "boss-1", + updatedAt: new Date(Date.now() - 31 * 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue({ state: "active" }); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).toHaveBeenCalledWith({ + where: { id: { in: ["import-stale"] } }, + data: { + status: "failed", + failureReason: "interrupted_by_restart", + completedAt: expect.any(Date), + }, + }); + }); + + it("leaves a row alone while its heartbeat is fresh and pg-boss reports the job active", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-live", + pgBossJobId: "boss-2", + updatedAt: new Date(Date.now() - 2 * 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue({ state: "active" }); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).not.toHaveBeenCalled(); + expect(mocks.getJobById).toHaveBeenCalledWith( + APPLE_HEALTH_IMPORT_V2_QUEUE, + "boss-2", + ); + }); + + it("flips a row whose backing pg-boss job is gone even with a fresh heartbeat", async () => { + mocks.findMany.mockResolvedValue([ + { + id: "import-gone", + pgBossJobId: "boss-3", + updatedAt: new Date(Date.now() - 60 * 1000), + }, + ]); + mocks.getJobById.mockResolvedValue(null); + + await reconcileOrphanImportJobs(); + + expect(mocks.updateMany).toHaveBeenCalledWith({ + where: { id: { in: ["import-gone"] } }, + data: { + status: "failed", + failureReason: "interrupted_by_restart", + completedAt: expect.any(Date), + }, + }); + }); + + it("only ever scopes the candidate query to the current parser revision", async () => { + mocks.findMany.mockResolvedValue([]); + + await reconcileOrphanImportJobs(); + + expect(mocks.findMany).toHaveBeenCalledWith({ + where: { + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + status: { in: ["unpacking", "parsing", "upserting"] }, + }, + select: { id: true, pgBossJobId: true, updatedAt: true }, + }); + }); +}); + +describe("handleImportJobReconcileTick", () => { + it("delegates to reconcileOrphanImportJobs and resolves on success", async () => { + mocks.findMany.mockResolvedValue([]); + + await expect( + handleImportJobReconcileTick([] as never), + ).resolves.toBeUndefined(); + expect(mocks.findMany).toHaveBeenCalledTimes(1); + }); + + it("swallows a reconcile failure instead of throwing", async () => { + mocks.findMany.mockRejectedValue(new Error("db unavailable")); + + // A background sweep this cheap must never spam pg-boss's + // retry/backoff machinery — it just re-runs on the next 15-minute + // tick. The handler must resolve even when the underlying reconcile + // pass throws. + await expect( + handleImportJobReconcileTick([] as never), + ).resolves.toBeUndefined(); + }); +}); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 662b144c2..9bf197017 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -37,6 +37,55 @@ import { type ImportJobResult, } from "@/lib/measurements/import-apple-health-export"; import { recomputeUserRollups } from "@/lib/rollups/measurement-rollups"; +import { withBackgroundEvent } from "@/lib/logging/background"; + +/** + * Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1 + * (issue #588) — `reconcileOrphanImportJobs()` used to run only once, + * at worker boot. `restart: unless-stopped` in `docker-compose.yml` + * DOES bring a crashed/OOM-killed worker back up, so the boot-time + * pass usually catches a dead run — but only when that restart lands + * AFTER the row's heartbeat has already gone stale (30 minutes) or + * pg-boss no longer reports the backing job as live. A fast restart + * lands neither condition true yet, and because the pass runs + * exactly once per boot, a worker that then stays up and healthy + * NEVER re-evaluates that row — it is stuck in `unpacking` / + * `parsing` / `upserting` with "0 rows imported" forever and no + * failure ever surfaced to the status poll. A 15-minute cron closes + * that gap: within two ticks of the heartbeat going stale, the same + * idempotent reconcile flips the row to `failed` with an honest + * reason, so the UI's "Unpacking the archive…" spinner eventually + * turns into a visible, actionable error instead of an indefinite + * silent stall. + */ +export const IMPORT_JOB_RECONCILE_QUEUE = "apple-health-import-reconcile"; +export const IMPORT_JOB_RECONCILE_CRON = "*/15 * * * *"; + +/** + * Cron handler for the periodic orphan-ImportJob sweep. Delegates to + * the same `reconcileOrphanImportJobs()` the boot path calls — the + * heartbeat + live pg-boss-job check already make it safe to re-run + * on a healthy worker (a live import's fresh heartbeat and `active` + * pg-boss state both keep its row untouched). Errors are logged, not + * rethrown — a reconcile miss must never spam pg-boss's retry/backoff + * machinery for a background sweep this cheap to just re-run in 15 + * minutes. + */ +export async function handleImportJobReconcileTick( + jobs: Job[], +): Promise { + void jobs; + await withBackgroundEvent( + "job.apple_health_import_reconcile", + async (evt) => { + try { + await reconcileOrphanImportJobs(); + } catch (err) { + evt.addWarning(`apple-health-import-reconcile failed: ${err}`); + } + }, + ); +} /** Queue name isolated from revision-1 workers during rolling deployments. */ export const APPLE_HEALTH_IMPORT_V2_QUEUE = "apple-health-import-v2"; @@ -260,7 +309,7 @@ export async function handleAppleHealthImport( elapsedMs: 0, }); - const unzip = extractExportXml(uploadPath); + const unzip = await extractExportXml(uploadPath); extractedXmlPath = unzip.xmlPath; // Phase 2 + 3: parsing + upserting (the parser tracks both diff --git a/src/lib/jobs/reminder/register-maintenance.ts b/src/lib/jobs/reminder/register-maintenance.ts index 198784ad9..7008e2e90 100644 --- a/src/lib/jobs/reminder/register-maintenance.ts +++ b/src/lib/jobs/reminder/register-maintenance.ts @@ -41,7 +41,10 @@ import { APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, APPLE_HEALTH_IMPORT_CONCURRENCY, + IMPORT_JOB_RECONCILE_QUEUE, + IMPORT_JOB_RECONCILE_CRON, handleAppleHealthImport, + handleImportJobReconcileTick, migrateLegacyAppleHealthImport, type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; @@ -314,6 +317,11 @@ const allQueues = [ INTAKE_AUTO_SKIP_QUEUE, APPLE_HEALTH_IMPORT_V2_QUEUE, APPLE_HEALTH_IMPORT_LEGACY_QUEUE, + // v1.32.1 (issue #588) — periodic sweep for ImportJob rows orphaned by a + // worker crash/restart mid-run. Without this entry the 15-minute cron + // below silently no-ops and a stuck "unpacking"/"parsing"/"upserting" row + // that survives past the next worker boot is never revisited. + IMPORT_JOB_RECONCILE_QUEUE, MEDICATION_INTAKE_IMPORT_QUEUE, // v1.8.2 — one-time duplicate dose-slot cleanup. Boot discovery enqueues // one job per user holding two live intake rows that snap to the same @@ -462,6 +470,13 @@ const schedules: ScheduleEntry[] = [ // Document vault — daily 04:10 Europe/Berlin purge for tombstoned // documents past the 30-day undo grace. [DOCUMENT_PURGE_QUEUE, DOCUMENT_PURGE_CRON], + // v1.32.1 (issue #588) — every-15-minute orphan-ImportJob sweep. Re-runs + // the same reconcile the boot path uses, so a stuck "unpacking" row + // whose worker crashed/restarted without the boot-time pass catching it + // (heartbeat not yet stale, pg-boss job not yet expired) still converges + // to a visible `failed` state within two ticks instead of staying stuck + // forever on an otherwise-healthy worker. + [IMPORT_JOB_RECONCILE_QUEUE, IMPORT_JOB_RECONCILE_CRON], ]; /** @@ -706,6 +721,14 @@ export async function registerMaintenanceQueues( } }, ); + // v1.32.1 (issue #588) — periodic orphan-ImportJob sweep. Single-flight: + // the underlying `updateMany` is idempotent and two ticks racing the same + // read-then-write pass would just be wasted work. + await boss.work( + IMPORT_JOB_RECONCILE_QUEUE, + { localConcurrency: 1 }, + handleImportJobReconcileTick, + ); await boss.work( MEDICATION_INTAKE_IMPORT_QUEUE, { diff --git a/src/lib/multipart/__tests__/stream-to-disk.test.ts b/src/lib/multipart/__tests__/stream-to-disk.test.ts index 890b46319..42f8329c3 100644 --- a/src/lib/multipart/__tests__/stream-to-disk.test.ts +++ b/src/lib/multipart/__tests__/stream-to-disk.test.ts @@ -431,7 +431,7 @@ describe("streamMultipartToDisk — large multi-chunk binary round-trip", () => expect(written.equals(zip), `zip byte-exact for ${sizes}`).toBe(true); // The load-bearing assertion: the saved file is still a parseable // archive (EOCD intact) and the member round-trips. - const extracted = extractExportXml(result.filePath); + const extracted = await extractExportXml(result.filePath); xmlPath = extracted.xmlPath; expect(readFileSync(extracted.xmlPath).equals(xmlBuf)).toBe(true); } finally {