From 1b4594e69f0148b4455ea7a9a7aed5b4fdd94dfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Sat, 26 Sep 2026 20:11:45 +0200 Subject: [PATCH] fix(import): bound what an export may inflate to, allow one import per account, and never leave the upload behind The extractor held the inflated export.xml to a flat 8 GiB whatever the member occupied, and the header ratio check reads sizes the archive writes itself: a member declaring an honest ratio could inflate at 1000:1, so an 8 MB upload could fill the temp directory. The streamed byte count is now held to 200 times the member's compressed size on disk (never less than 64 MiB, never more than 8 GiB), and a stored member may not grow at all. An account may have one Apple Health import queued or running. A second upload with different bytes is refused with 409 import.apple_health.busy and the running job's id; the check and the new row are one transaction under an advisory lock. The staged upload is the person's export in plain text. It is now removed on every exit that leaves no worker to read it: a refused or broken multipart body, no worker running, a queue that takes no job (the row is failed as well), the admin route's missing or unknown target account, and an account deleted before the worker reached it. The worker removes the files before recording a failure, so a row that went with a deleted account no longer strands them. The 15-minute import reconcile pass also sweeps staged files older than the longest an import may run. --- .../__tests__/route.test.ts | 11 ++ .../admin/import-apple-health-export/route.ts | 46 +++++-- .../__tests__/route.test.ts | 63 +++++++++- .../api/import/apple-health-export/route.ts | 46 +++++-- .../__tests__/apple-health-staging.test.ts | 52 ++++++++ .../import/__tests__/unzip-export-xml.test.ts | 56 ++++++++- src/lib/import/apple-health-staging.ts | 114 ++++++++++++++++++ src/lib/import/unzip-export-xml.ts | 35 +++++- .../apple-health-import-reconcile.test.ts | 6 +- src/lib/jobs/apple-health-import-worker.ts | 69 ++++++++--- src/lib/jobs/job-outcome.ts | 1 + .../__tests__/stream-to-disk.test.ts | 27 ++++- src/lib/multipart/stream-to-disk.ts | 46 +++++-- src/lib/openapi/error-codes.ts | 2 +- src/lib/openapi/routes/import.ts | 7 +- ...pple-health-import-worker-failures.test.ts | 25 ++++ 16 files changed, 549 insertions(+), 57 deletions(-) create mode 100644 src/lib/import/__tests__/apple-health-staging.test.ts create mode 100644 src/lib/import/apple-health-staging.ts diff --git a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts b/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts index 952f2f006..fcb453ed6 100644 --- a/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts +++ b/src/app/api/admin/import-apple-health-export/__tests__/route.test.ts @@ -25,6 +25,7 @@ vi.mock("next/headers", () => ({ })); const mocks = vi.hoisted(() => ({ + runningImportFindFirst: vi.fn(async () => null), checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }), bossSend: vi.fn().mockResolvedValue("boss-job-1"), getGlobalBoss: vi.fn(), @@ -60,6 +61,16 @@ vi.mock("@/lib/db", () => ({ findFirst: mocks.importJobFindFirst, update: mocks.importJobUpdate, }, + // The one-running-import check and the new row, in one transaction. + // The check finds nothing running unless a test says otherwise. + $transaction: async (fn: (tx: unknown) => unknown) => + fn({ + $executeRaw: async () => 0, + importJob: { + findFirst: mocks.runningImportFindFirst, + create: mocks.importJobCreate, + }, + }), user: { findUnique: mocks.userFindUnique, }, diff --git a/src/app/api/admin/import-apple-health-export/route.ts b/src/app/api/admin/import-apple-health-export/route.ts index 67e8ecd32..b3ea40528 100644 --- a/src/app/api/admin/import-apple-health-export/route.ts +++ b/src/app/api/admin/import-apple-health-export/route.ts @@ -27,6 +27,11 @@ import { type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk"; +import { + createImportJobUnlessBusy, + discardStagedUpload, + IMPORT_BUSY_CODE, +} from "@/lib/import/apple-health-staging"; import { unlink } from "node:fs/promises"; export const dynamic = "force-dynamic"; @@ -96,6 +101,7 @@ export const POST = apiHandler(async (request: NextRequest) => { ipAddress: getClientIp(request), details: { reason: "missing_user_id" }, }); + await discardStagedUpload(uploaded.filePath); return apiError("Multipart field 'userId' is required", 422); } @@ -109,6 +115,7 @@ export const POST = apiHandler(async (request: NextRequest) => { ipAddress: getClientIp(request), details: { reason: "target_user_missing", targetUserId }, }); + await discardStagedUpload(uploaded.filePath); return apiError(`Target user '${targetUserId}' does not exist`, 422); } @@ -149,19 +156,27 @@ export const POST = apiHandler(async (request: NextRequest) => { ipAddress: getClientIp(request), details: { reason: "worker_not_running" }, }); + await discardStagedUpload(uploaded.filePath); throw new HttpError(503, "Background worker is not running"); } - const importJob = await prisma.importJob.create({ - data: { - userId: targetUser.id, - triggeredByAdminId: admin.id, - status: "queued", - uploadBytes: uploaded.bytes, - uploadSha256: uploaded.sha256, - parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, - }, + const staged = await createImportJobUnlessBusy(prisma, { + userId: targetUser.id, + triggeredByAdminId: admin.id, + status: "queued", + uploadBytes: uploaded.bytes, + uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }); + if ("busy" in staged) { + await discardStagedUpload(uploaded.filePath); + return apiError( + "An Apple Health import is already running for this account. Wait for it to finish, then upload again.", + 409, + { errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id }, + ); + } + const importJob = staged.created; const payload: AppleHealthImportPayload = { userId: targetUser.id, @@ -178,6 +193,19 @@ export const POST = apiHandler(async (request: NextRequest) => { payload, APPLE_HEALTH_IMPORT_SEND_OPTIONS, ); + if (!bossJobId) { + await prisma.importJob.update({ + where: { id: importJob.id }, + data: { + status: "failed", + failureReason: + "The import could not be queued. Upload the export again.", + completedAt: new Date(), + }, + }); + await discardStagedUpload(uploaded.filePath); + throw new HttpError(503, "The import could not be queued"); + } await prisma.importJob.update({ where: { id: importJob.id }, diff --git a/src/app/api/import/apple-health-export/__tests__/route.test.ts b/src/app/api/import/apple-health-export/__tests__/route.test.ts index 5a0aad7ef..8acfc3b04 100644 --- a/src/app/api/import/apple-health-export/__tests__/route.test.ts +++ b/src/app/api/import/apple-health-export/__tests__/route.test.ts @@ -27,6 +27,7 @@ vi.mock("next/headers", () => ({ })); const mocks = vi.hoisted(() => ({ + runningImportFindFirst: vi.fn(async () => null), checkRateLimit: vi.fn().mockResolvedValue({ allowed: true }), bossSend: vi.fn().mockResolvedValue("boss-job-1"), getGlobalBoss: vi.fn(), @@ -35,6 +36,7 @@ const mocks = vi.hoisted(() => ({ importJobUpdate: vi.fn(), streamToDisk: vi.fn(), unlink: vi.fn().mockResolvedValue(undefined), + rm: vi.fn().mockResolvedValue(undefined), })); vi.mock("@/lib/rate-limit", () => ({ @@ -51,7 +53,7 @@ vi.mock("@/lib/multipart/stream-to-disk", () => ({ vi.mock("node:fs/promises", async (importOriginal) => { const actual = await importOriginal(); - return { ...actual, unlink: mocks.unlink }; + return { ...actual, unlink: mocks.unlink, rm: mocks.rm }; }); vi.mock("@/lib/db", () => ({ @@ -61,6 +63,16 @@ vi.mock("@/lib/db", () => ({ findFirst: mocks.importJobFindFirst, update: mocks.importJobUpdate, }, + // The one-running-import check and the new row, in one transaction. + // The check finds nothing running unless a test says otherwise. + $transaction: async (fn: (tx: unknown) => unknown) => + fn({ + $executeRaw: async () => 0, + importJob: { + findFirst: mocks.runningImportFindFirst, + create: mocks.importJobCreate, + }, + }), }, })); @@ -90,6 +102,7 @@ beforeEach(() => { textFields: {}, }); mocks.unlink.mockResolvedValue(undefined); + mocks.rm.mockResolvedValue(undefined); }); function multipartReq(): NextRequest { @@ -220,3 +233,51 @@ describe("POST /api/import/apple-health-export — dedup by content hash (issue ); }); }); + +// The staged upload is the person's export in plain text. Every exit that +// leaves no worker to consume it removes it, and a second import for the same +// account is refused while one is running. +describe("POST /api/import/apple-health-export — the staged upload on every exit", () => { + const removed = () => + mocks.rm.mock.calls.some(([path]) => path === STAGED_PATH); + + it("refuses a second import while one is running, and removes the upload", async () => { + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + mocks.runningImportFindFirst.mockResolvedValue({ + id: "ij-running", + status: "parsing", + } as never); + const res = await POST(multipartReq()); + expect(res.status).toBe(409); + const body = await res.json(); + expect(body.meta).toMatchObject({ + errorCode: "import.apple_health.busy", + jobId: "ij-running", + }); + expect(mocks.importJobCreate).not.toHaveBeenCalled(); + expect(mocks.bossSend).not.toHaveBeenCalled(); + expect(removed()).toBe(true); + }); + + it("removes the upload when no worker is running", async () => { + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + mocks.getGlobalBoss.mockReturnValue(null); + const res = await POST(multipartReq()); + expect(res.status).toBe(503); + expect(removed()).toBe(true); + }); + + it("fails the row and removes the upload when the queue takes no job", async () => { + vi.mocked(getSession).mockResolvedValue(SESSION_OK as never); + mocks.bossSend.mockResolvedValue(null); + const res = await POST(multipartReq()); + expect(res.status).toBe(503); + expect(mocks.importJobUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: "ij-1" }, + data: expect.objectContaining({ status: "failed" }), + }), + ); + expect(removed()).toBe(true); + }); +}); diff --git a/src/app/api/import/apple-health-export/route.ts b/src/app/api/import/apple-health-export/route.ts index 131fd2b68..6919703ff 100644 --- a/src/app/api/import/apple-health-export/route.ts +++ b/src/app/api/import/apple-health-export/route.ts @@ -35,6 +35,11 @@ import { type AppleHealthImportPayload, } from "@/lib/jobs/apple-health-import-worker"; import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk"; +import { + createImportJobUnlessBusy, + discardStagedUpload, + IMPORT_BUSY_CODE, +} from "@/lib/import/apple-health-staging"; import { unlink } from "node:fs/promises"; export const dynamic = "force-dynamic"; @@ -147,18 +152,28 @@ export const POST = apiHandler(async (request: NextRequest) => { ipAddress: getClientIp(request), details: { reason: "worker_not_running" }, }); + // Nothing will ever read the staged export; it must not stay in /tmp. + await discardStagedUpload(uploaded.filePath); throw new HttpError(503, "Background worker is not running"); } - const importJob = await prisma.importJob.create({ - data: { - userId: user.id, - status: "queued", - uploadBytes: uploaded.bytes, - uploadSha256: uploaded.sha256, - parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, - }, + // One running import per account (see `apple-health-staging.ts`). + const staged = await createImportJobUnlessBusy(prisma, { + userId: user.id, + status: "queued", + uploadBytes: uploaded.bytes, + uploadSha256: uploaded.sha256, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, }); + if ("busy" in staged) { + await discardStagedUpload(uploaded.filePath); + return apiError( + "An Apple Health import is already running for this account. Wait for it to finish, then upload again.", + 409, + { errorCode: IMPORT_BUSY_CODE, jobId: staged.busy.id }, + ); + } + const importJob = staged.created; const payload: AppleHealthImportPayload = { userId: user.id, @@ -174,6 +189,21 @@ export const POST = apiHandler(async (request: NextRequest) => { payload, APPLE_HEALTH_IMPORT_SEND_OPTIONS, ); + if (!bossJobId) { + // The queue did not take the job, so no worker will consume the upload + // or move the row on. Fail the row now and remove the export. + await prisma.importJob.update({ + where: { id: importJob.id }, + data: { + status: "failed", + failureReason: + "The import could not be queued. Upload the export again.", + completedAt: new Date(), + }, + }); + await discardStagedUpload(uploaded.filePath); + throw new HttpError(503, "The import could not be queued"); + } await prisma.importJob.update({ where: { id: importJob.id }, diff --git a/src/lib/import/__tests__/apple-health-staging.test.ts b/src/lib/import/__tests__/apple-health-staging.test.ts new file mode 100644 index 000000000..a8132d17e --- /dev/null +++ b/src/lib/import/__tests__/apple-health-staging.test.ts @@ -0,0 +1,52 @@ +/** + * The sweep removes staged exports no running import can own, and nothing + * else. Mutations that must turn this red: drop the age check (the fresh + * upload of a running import goes), or widen the name pattern (a file the app + * did not write goes). + */ +import { mkdtempSync, readdirSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { + STAGING_MAX_AGE_MS, + sweepStaleImportStaging, +} from "../apple-health-staging"; + +describe("sweepStaleImportStaging", () => { + it("removes old staged uploads and extracted XML, and leaves the rest", async () => { + const dir = mkdtempSync(join(tmpdir(), "healthlog-sweep-")); + const uuid = "0b7f3c2a-1d4e-4f5a-9b8c-7d6e5f4a3b2c"; + const names = { + oldUpload: `healthlog-apple-health-import-${uuid}.bin`, + oldAdminUpload: `healthlog-admin-apple-health-import-${uuid}.bin`, + oldXml: `healthlog-import-${"ab".repeat(12)}.xml`, + freshUpload: `healthlog-upload-${uuid}.bin`, + foreign: "somebody-elses-file.bin", + }; + for (const name of Object.values(names)) { + writeFileSync(join(dir, name), "x"); + } + const now = Date.now(); + const old = new Date(now - STAGING_MAX_AGE_MS - 60_000); + for (const name of [ + names.oldUpload, + names.oldAdminUpload, + names.oldXml, + names.foreign, + ]) { + utimesSync(join(dir, name), old, old); + } + + expect(await sweepStaleImportStaging(dir, now)).toBe(3); + expect(readdirSync(dir).sort()).toEqual( + [names.foreign, names.freshUpload].sort(), + ); + }); + + it("answers zero for a directory that is not there", async () => { + expect(await sweepStaleImportStaging("/nonexistent/healthlog")).toBe(0); + }); +}); diff --git a/src/lib/import/__tests__/unzip-export-xml.test.ts b/src/lib/import/__tests__/unzip-export-xml.test.ts index 10046cbaf..9325647da 100644 --- a/src/lib/import/__tests__/unzip-export-xml.test.ts +++ b/src/lib/import/__tests__/unzip-export-xml.test.ts @@ -6,9 +6,12 @@ import { deflateRawSync } from "node:zlib"; import { pipeline } from "node:stream/promises"; import { Readable } from "node:stream"; +import { readdirSync } from "node:fs"; + import { createByteCap, extractExportXml, + memberOutputCap, readCentralDirectory, } from "../unzip-export-xml"; @@ -22,6 +25,8 @@ function buildMinimalZip( filename: string, payload: Buffer, method: 0 | 8 = 8, + /** What the headers claim the member inflates to; the truth by default. */ + declaredSize?: number, ): Buffer { const compressed = method === 8 ? deflateRawSync(payload) : payload; const crc32 = (() => { @@ -54,7 +59,7 @@ function buildMinimalZip( localHeader.writeUInt16LE(0, 12); // date localHeader.writeUInt32LE(crc32, 14); localHeader.writeUInt32LE(compressed.length, 18); - localHeader.writeUInt32LE(payload.length, 22); + localHeader.writeUInt32LE(declaredSize ?? payload.length, 22); localHeader.writeUInt16LE(nameBuf.length, 26); localHeader.writeUInt16LE(0, 28); // extra len @@ -69,7 +74,7 @@ function buildMinimalZip( cdh.writeUInt16LE(0, 14); // date cdh.writeUInt32LE(crc32, 16); cdh.writeUInt32LE(compressed.length, 20); - cdh.writeUInt32LE(payload.length, 24); + cdh.writeUInt32LE(declaredSize ?? payload.length, 24); cdh.writeUInt16LE(nameBuf.length, 28); cdh.writeUInt16LE(0, 30); // extra cdh.writeUInt16LE(0, 32); // comment @@ -166,6 +171,30 @@ describe("extractExportXml", () => { expect(seenDeclaredBytes).toBe(payload.length); }); + // A member whose headers claim an honest ratio and whose deflate stream + // inflates a thousandfold. The header check passes it; only the count of + // the bytes actually produced, held to the compressed size on disk, stops + // it, and it must leave nothing behind in the staging directory. + it("refuses a member that inflates past its compressed size's cap, whatever it declares", async () => { + const payload = Buffer.alloc(80 * 1024 * 1024); + const compressedLength = deflateRawSync(payload).length; + const zip = buildMinimalZip( + "apple_health_export/export.xml", + payload, + 8, + compressedLength * 2, + ); + const tmp = mkdtempSync(join(tmpdir(), "healthlog-unzip-")); + const zipPath = join(tmp, "export.zip"); + writeFileSync(zipPath, zip); + const staged = () => + readdirSync(tmpdir()).filter((n) => n.startsWith("healthlog-import-")); + const before = new Set(staged()); + + await expect(extractExportXml(zipPath)).rejects.toThrow(/zip bomb/); + expect(staged().filter((n) => !before.has(n))).toEqual([]); + }); + it("throws when the export.xml member is missing", async () => { const payload = Buffer.from("noop"); const zip = buildMinimalZip("other-file.txt", payload); @@ -209,3 +238,26 @@ describe("createByteCap", () => { ).rejects.toThrow(/exceeds the 6-byte cap/); }); }); + +describe("memberOutputCap", () => { + it("holds a deflated member to 200 times what it occupies, with a floor", () => { + expect( + memberOutputCap({ compressionMethod: 8, compressedSize: 1024 }), + ).toBe(64 * 1024 * 1024); + expect( + memberOutputCap({ + compressionMethod: 8, + compressedSize: 10 * 1024 * 1024, + }), + ).toBe(2000 * 1024 * 1024); + expect( + memberOutputCap({ compressionMethod: 8, compressedSize: 1024 ** 3 }), + ).toBe(8 * 1024 ** 3); + }); + + it("lets a stored member grow by nothing", () => { + expect(memberOutputCap({ compressionMethod: 0, compressedSize: 500 })).toBe( + 500, + ); + }); +}); diff --git a/src/lib/import/apple-health-staging.ts b/src/lib/import/apple-health-staging.ts new file mode 100644 index 000000000..08819a490 --- /dev/null +++ b/src/lib/import/apple-health-staging.ts @@ -0,0 +1,114 @@ +/** + * The staged Apple Health upload, from the request that writes it to the job + * that consumes it, and what happens to it on every way out. + * + * The upload is the person's whole health export in plain text, written to + * the shared temp directory by the kick-off request and read by the import + * worker. Until v1.39.3 several exits left it there: a worker that was not + * running (503), a queue that did not accept the job, an account deleted + * before the worker reached it, and nothing ever swept what was left. Every + * exit now removes it, and a periodic sweep removes anything older than the + * longest an import may run, whatever left it. + * + * It also holds the rule of one running import per account. An import holds + * the archive, the extracted XML (up to 8 GiB) and a worker slot; a second + * one for the same account only doubles that and races the first on the same + * rows. The check and the new row are one transaction under an advisory lock, + * so two uploads that arrive together cannot both pass. + */ +import { readdir, rm, stat } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { Prisma, PrismaClient } from "@/generated/prisma/client"; + +/** An import in one of these holds its staged files and a worker slot. */ +export const ACTIVE_IMPORT_STATUSES = [ + "queued", + "unpacking", + "parsing", + "upserting", +] as const; + +/** + * An active row older than this is not holding anything any more: the queue + * expires an import after six hours (`APPLE_HEALTH_IMPORT_SEND_OPTIONS`) and + * the reconcile pass fails it soon after. It must not block the account. + */ +const ACTIVE_IMPORT_MAX_AGE_MS = 7 * 60 * 60 * 1000; + +/** Staged files older than this belong to no running import. */ +export const STAGING_MAX_AGE_MS = ACTIVE_IMPORT_MAX_AGE_MS; + +/** The names the kick-off and the extractor give their temp files. */ +const STAGING_NAME = + /^healthlog-(?:apple-health-import|admin-apple-health-import|upload)-[0-9a-f-]{36}\.bin$|^healthlog-import-[0-9a-f]{24}\.xml$/; + +/** The error code a second concurrent import is refused with. */ +export const IMPORT_BUSY_CODE = "import.apple_health.busy"; + +export type CreateImportJobResult = + { created: { id: string } } | { busy: { id: string; status: string } }; + +/** + * Create the ImportJob row unless the account already has an import running. + * `data` is the row as the caller would create it. + */ +export async function createImportJobUnlessBusy( + prisma: PrismaClient, + data: Prisma.ImportJobUncheckedCreateInput, + now: Date = new Date(), +): Promise { + return prisma.$transaction(async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtext(${`apple-health-import:${data.userId}`}))`; + const running = await tx.importJob.findFirst({ + where: { + userId: data.userId, + status: { in: [...ACTIVE_IMPORT_STATUSES] }, + startedAt: { gte: new Date(now.getTime() - ACTIVE_IMPORT_MAX_AGE_MS) }, + }, + orderBy: { startedAt: "desc" }, + select: { id: true, status: true }, + }); + if (running) return { busy: running }; + const created = await tx.importJob.create({ data, select: { id: true } }); + return { created }; + }); +} + +/** Remove a staged file. Never throws; a file already gone is fine. */ +export async function discardStagedUpload(path: string): Promise { + await rm(path, { force: true }).catch(() => {}); +} + +/** + * Remove staged uploads and extracted XML files older than + * `STAGING_MAX_AGE_MS` from `dir`. Only names this app writes are touched. + * Answers how many were removed. + */ +export async function sweepStaleImportStaging( + dir: string = tmpdir(), + now: number = Date.now(), +): Promise { + let names: string[]; + try { + names = await readdir(dir); + } catch { + return 0; + } + let removed = 0; + for (const name of names) { + if (!STAGING_NAME.test(name)) continue; + const path = join(dir, name); + try { + const info = await stat(path); + if (!info.isFile()) continue; + if (now - info.mtimeMs < STAGING_MAX_AGE_MS) continue; + await rm(path, { force: true }); + removed++; + } catch { + // Gone between the listing and here, or not ours to remove. + } + } + return removed; +} diff --git a/src/lib/import/unzip-export-xml.ts b/src/lib/import/unzip-export-xml.ts index e9922a7ad..4f046ef57 100644 --- a/src/lib/import/unzip-export-xml.ts +++ b/src/lib/import/unzip-export-xml.ts @@ -73,6 +73,39 @@ const MAX_DECOMPRESSED_BYTES = 8 * 1024 * 1024 * 1024; */ const MAX_COMPRESSION_RATIO = 200; +/** + * Floor for the output cap of a small member. The ratio cap alone would + * refuse a tiny, legitimately very compressible export; below this many + * bytes of output the ratio is not what protects the disk. + */ +const MIN_OUTPUT_CAP_BYTES = 64 * 1024 * 1024; + +/** + * How many bytes one member may inflate to, from what it ACTUALLY occupies + * in the archive rather than what its header claims. The header is the + * attacker's to write: a member declaring an honest-looking 10:1 ratio can + * still inflate at 1000:1, and with only the flat 8 GiB ceiling an 8 MB + * upload could fill `/tmp` with 8 GB before anything noticed. The streamed + * byte count is held to the ratio the preflight states, applied to the + * compressed bytes on disk, and never above the absolute ceiling. A stored + * member cannot grow at all. + */ +export function memberOutputCap(entry: { + compressionMethod: number; + compressedSize: number; +}): number { + if (entry.compressionMethod === 0) { + return Math.min(entry.compressedSize, MAX_DECOMPRESSED_BYTES); + } + return Math.min( + MAX_DECOMPRESSED_BYTES, + Math.max( + MIN_OUTPUT_CAP_BYTES, + entry.compressedSize * MAX_COMPRESSION_RATIO, + ), + ); +} + /** * Cap on the central-directory slice the extractor is willing to hold * in memory. A real Apple export's central directory is a few KB (one @@ -636,7 +669,7 @@ async function streamEntryToFile( } const dest = createWriteStream(destPath); - const cap = createByteCap(MAX_DECOMPRESSED_BYTES); + const cap = createByteCap(memberOutputCap(entry)); try { if (entry.compressedSize === 0) { diff --git a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts index 219978aae..cff18e33f 100644 --- a/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts +++ b/src/lib/jobs/__tests__/apple-health-import-reconcile.test.ts @@ -24,6 +24,10 @@ const mocks = vi.hoisted(() => ({ getJobById: vi.fn(), })); +vi.mock("@/lib/import/apple-health-staging", () => ({ + sweepStaleImportStaging: vi.fn(async () => 2), +})); + vi.mock("@/lib/db", () => ({ prisma: { importJob: { @@ -172,7 +176,7 @@ describe("handleImportJobReconcileTick", () => { await expect(handleImportJobReconcileTick([] as never)).resolves.toEqual({ ok: true, - did: {}, + did: { import_staging_swept: 2 }, }); expect(mocks.findMany).toHaveBeenCalledTimes(1); }); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 257649e72..adc523730 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -46,6 +46,7 @@ import { ROLLUP_FOLD_WINDOW_MS, } from "@/lib/rollups/measurement-rollups"; import { withBackgroundEvent } from "@/lib/logging/background"; +import { sweepStaleImportStaging } from "@/lib/import/apple-health-staging"; /** * Queue + cron for the periodic orphan-ImportJob sweep. v1.32.1 @@ -87,13 +88,18 @@ export async function handleImportJobReconcileTick( return withBackgroundEvent( "job.apple_health_import_reconcile", async (evt) => { + // Staged uploads and extracted XML that no running import can still + // own: whatever exit left them (a crash, a restart mid-import, a + // deploy), the person's export does not stay in /tmp past this. + const swept = await sweepStaleImportStaging().catch(() => 0); + evt.addMeta("import_staging_swept", swept); try { await reconcileOrphanImportJobs(); } catch (err) { evt.addWarning(`apple-health-import-reconcile failed: ${err}`); return jobFailed("apple health import reconcile failed", err); } - return jobDone(); + return jobDone({ import_staging_swept: swept }); }, ); } @@ -324,16 +330,31 @@ export async function handleAppleHealthImport( `[apple-health-import] No ImportJob row for pgBossJobId=${job.id};` + " creating a stand-in", ); - importJob = await prisma.importJob.create({ - data: { - userId, - triggeredByAdminId: triggeredByAdminId ?? null, - pgBossJobId: job.id, - status: "queued", - uploadBytes, - parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, - }, - }); + // The row is gone when the account is: an account deleted between the + // upload and this run takes its ImportJob with it by cascade. Then there + // is nobody to import for, and the staged export is removed rather than + // left in /tmp by a stand-in create that fails on the missing account. + const accountExists = + (await prisma.user.count({ where: { id: userId } })) > 0; + if (!accountExists) { + safeUnlink(uploadPath); + return jobDone({ skipped: "account_gone" }); + } + try { + importJob = await prisma.importJob.create({ + data: { + userId, + triggeredByAdminId: triggeredByAdminId ?? null, + pgBossJobId: job.id, + status: "queued", + uploadBytes, + parserRevision: APPLE_HEALTH_IMPORT_PARSER_REVISION, + }, + }); + } catch (err) { + safeUnlink(uploadPath); + throw err; + } } const importJobId = importJob.id; @@ -566,16 +587,26 @@ export async function handleAppleHealthImport( : err instanceof Error ? err.message : String(err); - await prisma.importJob.update({ - where: { id: importJobId }, - data: { - status: "failed", - failureReason: reason.slice(0, 1000), - completedAt: new Date(), - }, - }); + // The files first: the row update below throws when the account was + // deleted mid-import (the row went with it), and the export must not be + // left behind in that case either. if (extractedXmlPath) safeUnlink(extractedXmlPath); safeUnlink(uploadPath); + await prisma.importJob + .update({ + where: { id: importJobId }, + data: { + status: "failed", + failureReason: reason.slice(0, 1000), + completedAt: new Date(), + }, + }) + .catch((updateErr: unknown) => { + console.warn( + `[apple-health-import] Could not record the failure on ImportJob=${importJobId}`, + updateErr, + ); + }); throw err; } } diff --git a/src/lib/jobs/job-outcome.ts b/src/lib/jobs/job-outcome.ts index 78dc759b8..eadaa01e7 100644 --- a/src/lib/jobs/job-outcome.ts +++ b/src/lib/jobs/job-outcome.ts @@ -107,6 +107,7 @@ export const JOB_FACT_ALLOWLIST: ReadonlySet = new Set([ "geolite2_fetch_status", "host_metric_pruned", "hourly_rows_upserted", + "import_staging_swept", "imported", "in_slot", "in_window", diff --git a/src/lib/multipart/__tests__/stream-to-disk.test.ts b/src/lib/multipart/__tests__/stream-to-disk.test.ts index 42f8329c3..2d5386bd0 100644 --- a/src/lib/multipart/__tests__/stream-to-disk.test.ts +++ b/src/lib/multipart/__tests__/stream-to-disk.test.ts @@ -1,5 +1,7 @@ import { describe, expect, it } from "vitest"; -import { readFileSync, unlinkSync } from "node:fs"; +import { mkdtempSync, readdirSync, readFileSync, unlinkSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { createHash } from "node:crypto"; import { deflateRawSync } from "node:zlib"; @@ -112,6 +114,29 @@ describe("streamMultipartToDisk", () => { ).rejects.toThrow(/size cap/); }); + // What was written before the refusal is the person's export in plain + // text; it must not stay behind in the temp directory. + it("removes the partial file when the upload is refused part-way", async () => { + const dir = mkdtempSync(join(tmpdir(), "healthlog-stream-")); + const body = buildMultipartBody("export.zip", Buffer.alloc(4096, 0x41)); + const chunked = new ReadableStream({ + start(controller) { + for (let i = 0; i < body.length; i += 256) { + controller.enqueue(new Uint8Array(body.subarray(i, i + 256))); + } + controller.close(); + }, + }); + await expect( + streamMultipartToDisk( + chunked, + `multipart/form-data; boundary=${BOUNDARY}`, + { maxBytes: 1000, fieldName: "file", tmpDir: dir }, + ), + ).rejects.toThrow(/size cap/); + expect(readdirSync(dir)).toEqual([]); + }); + it("throws when the named field is missing", async () => { const body = buildMultipartBody("export.zip", Buffer.from("x"), { something: "else", diff --git a/src/lib/multipart/stream-to-disk.ts b/src/lib/multipart/stream-to-disk.ts index 1bc734df9..fcc8fe650 100644 --- a/src/lib/multipart/stream-to-disk.ts +++ b/src/lib/multipart/stream-to-disk.ts @@ -17,6 +17,7 @@ */ import { createHash, randomUUID } from "node:crypto"; import { createWriteStream } from "node:fs"; +import { rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Writable } from "node:stream"; @@ -293,22 +294,41 @@ export async function streamMultipartToDisk( } }; - while (!streamDone) { - const { value, done } = await reader.read(); - if (done) { - streamDone = true; - break; + try { + while (!streamDone) { + const { value, done } = await reader.read(); + if (done) { + streamDone = true; + break; + } + buffer = Buffer.concat([buffer, Buffer.from(value)]); + await processBuffer(); } - buffer = Buffer.concat([buffer, Buffer.from(value)]); await processBuffer(); - } - await processBuffer(); - await closeFileSink(); + await closeFileSink(); - if (bytesWritten === 0) { - throw new Error( - `Multipart body did not include a '${opts.fieldName}' file field`, - ); + if (bytesWritten === 0) { + throw new Error( + `Multipart body did not include a '${opts.fieldName}' file field`, + ); + } + } catch (err) { + // Whatever was written so far is the user's plaintext health export, + // and it must not outlive a refused upload in the shared temp + // directory: the size cap, a malformed body and a dropped connection + // all end here. The reader is cancelled so the rest of the body is not + // read for nothing. + await reader.cancel().catch(() => {}); + const sink = fileSink as Writable | null; + if (sink && !sink.closed) { + fileSinkOpen = false; + await new Promise((resolve) => { + sink.once("close", () => resolve()); + sink.destroy(); + }); + } + await rm(filePath, { force: true }).catch(() => {}); + throw err; } return { diff --git a/src/lib/openapi/error-codes.ts b/src/lib/openapi/error-codes.ts index bf1eaa48c..bb2c28a46 100644 --- a/src/lib/openapi/error-codes.ts +++ b/src/lib/openapi/error-codes.ts @@ -168,7 +168,7 @@ export const ERROR_CODE_CATALOGUE: Readonly> = "illness.episode.parent-not-found", "illness.insights.invalid", ], - import: ["import.write_failed"], + import: ["import.apple_health.busy", "import.write_failed"], ingest: ["ingest.rate_limited"], insights: [ "insights.generate.budgetExceeded", diff --git a/src/lib/openapi/routes/import.ts b/src/lib/openapi/routes/import.ts index 04cc258eb..82c8431dd 100644 --- a/src/lib/openapi/routes/import.ts +++ b/src/lib/openapi/routes/import.ts @@ -324,6 +324,11 @@ export const importPaths: NonNullable = { content: { "application/json": { schema: errorEnvelope } }, }, ...stdResponses, + "409": { + description: + "An import for this account is already queued or running, and these bytes are not the ones it carries (the same bytes resolve to that job as a 202). `meta.errorCode` is `import.apple_health.busy` and `meta.jobId` names the running job; poll it, then upload again. The upload was discarded.", + content: { "application/json": { schema: errorEnvelope } }, + }, "422": { description: "The multipart body could not be streamed to disk — no `file` part, a malformed boundary, or the size cap tripped mid-stream. The message names the failure.", @@ -335,7 +340,7 @@ export const importPaths: NonNullable = { }, "503": { description: - "No background worker is bound to this process, so the job cannot be queued. The upload was staged and is discarded; retry once a worker is running.", + "No background worker is bound to this process, or the queue did not accept the job, so nothing will run it. The staged upload is discarded (and a created job is marked failed); retry once a worker is running.", content: { "application/json": { schema: errorEnvelope } }, }, }, diff --git a/tests/integration/apple-health-import-worker-failures.test.ts b/tests/integration/apple-health-import-worker-failures.test.ts index b10a0ed21..2793396ad 100644 --- a/tests/integration/apple-health-import-worker-failures.test.ts +++ b/tests/integration/apple-health-import-worker-failures.test.ts @@ -219,3 +219,28 @@ describe("apple health import worker — failure paths (issue #486)", () => { expect(existsSync(uploadPath)).toBe(false); }); }); + +describe("an account deleted before the worker reaches its upload", () => { + it("removes the staged export and does not fail on the missing account", async () => { + const user = await createUser("deleted-before-import"); + const uploadPath = join( + scratchDir, + "healthlog-apple-health-import-gone.bin", + ); + writeFileSync(uploadPath, "the export, in plain text"); + // The account goes, and its ImportJob row with it by cascade. + await getPrismaClient().user.delete({ where: { id: user.id } }); + + const outcome = await handleAppleHealthImport( + bossJob("boss-gone-1", { + userId: user.id, + uploadPath, + uploadBytes: 25, + enqueuedAt: new Date().toISOString(), + }), + ); + expect(outcome).toMatchObject({ ok: true }); + expect(existsSync(uploadPath)).toBe(false); + expect(await getPrismaClient().importJob.count()).toBe(0); + }); +});