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); + }); +});