mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-14 22:50:51 +00:00
chore(release): v1.28.36 — re-importing after a failed Apple Health import actually retries
The v1.28.33 rollup + staging fixes were correct but never reached the reporter's case: the upload dedup keys on the content SHA-256 and the lookup matched a previously FAILED job too, so re-uploading the same export.zip returned the dead job's id and replayed its stale failureReason without ever staging or sending again. The dedup now excludes failed rows (short-circuits only on queued/unpacking/parsing/ upserting/done); the same bytes retry cleanly. A redundant staged upload is unlinked on a dedup hit instead of leaking in the staging dir. reconcileOrphanImportJobs no longer unconditionally fails every non-terminal ImportJob at boot: it now checks pg-boss job state plus a progress heartbeat (new import_jobs.updated_at, migration 0242), so a split or multi-worker deployment cannot kill an import running in another worker. The missing-staging message names the separate-container cause explicitly, with a matching note in docs/self-hosting/scaling.md.
This commit is contained in:
1 parent
a614542cc3
commit
82aa66b400
13 files changed
+590
-19
No files matched your search
@@ -32,6 +32,8 @@ const mocks = vi.hoisted(() => ({
|
||||
importJobFindFirst: vi.fn().mockResolvedValue(null),
|
||||
importJobUpdate: vi.fn(),
|
||||
userFindUnique: vi.fn(),
|
||||
streamToDisk: vi.fn(),
|
||||
unlink: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/rate-limit", () => ({
|
||||
@@ -42,6 +44,15 @@ vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: mocks.getGlobalBoss,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/multipart/stream-to-disk", () => ({
|
||||
streamMultipartToDisk: mocks.streamToDisk,
|
||||
}));
|
||||
|
||||
vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("node:fs/promises")>();
|
||||
return { ...actual, unlink: mocks.unlink };
|
||||
});
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
importJob: {
|
||||
@@ -67,6 +78,8 @@ const USER_SESSION = {
|
||||
user: { id: "user-2", username: "alice", role: "USER" as const },
|
||||
};
|
||||
|
||||
const STAGED_PATH = "/tmp/healthlog-admin-apple-health-import-abc.bin";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetAllMocks();
|
||||
mocks.checkRateLimit.mockResolvedValue({ allowed: true });
|
||||
@@ -76,8 +89,27 @@ beforeEach(() => {
|
||||
mocks.importJobFindFirst.mockResolvedValue(null);
|
||||
mocks.importJobUpdate.mockResolvedValue({ id: "ij-1" });
|
||||
mocks.userFindUnique.mockResolvedValue({ id: "user-2", username: "alice" });
|
||||
mocks.streamToDisk.mockResolvedValue({
|
||||
filePath: STAGED_PATH,
|
||||
bytes: 100,
|
||||
sha256: "deadbeef",
|
||||
originalFilename: "export.zip",
|
||||
textFields: { userId: "user-2" },
|
||||
});
|
||||
mocks.unlink.mockResolvedValue(undefined);
|
||||
});
|
||||
|
||||
function multipartReq(): NextRequest {
|
||||
return new NextRequest(
|
||||
"http://localhost/api/admin/import-apple-health-export",
|
||||
{
|
||||
method: "POST",
|
||||
headers: { "content-type": "multipart/form-data; boundary=---x" },
|
||||
body: "stub",
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
describe("POST /api/admin/import-apple-health-export", () => {
|
||||
it("returns 401 when unauthenticated", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(null);
|
||||
@@ -139,4 +171,45 @@ describe("POST /api/admin/import-apple-health-export", () => {
|
||||
const res = await POST(req);
|
||||
expect(res.status).toBe(429);
|
||||
});
|
||||
|
||||
it("re-stages and enqueues when the only prior job for the same bytes is failed (issue #486)", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(ADMIN_SESSION as never);
|
||||
mocks.importJobFindFirst.mockImplementation(async (args: unknown) => {
|
||||
const where = (args as { where?: { status?: { not?: string } } }).where;
|
||||
if (where?.status?.not === "failed") return null;
|
||||
return { id: "old-failed-job", status: "failed" };
|
||||
});
|
||||
|
||||
const res = await POST(multipartReq());
|
||||
expect(res.status).toBe(202);
|
||||
const body = await res.json();
|
||||
expect(body.data.status).toBe("queued");
|
||||
expect(body.data.idempotent).toBeUndefined();
|
||||
expect(body.data.jobId).not.toBe("old-failed-job");
|
||||
expect(mocks.bossSend).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
where: expect.objectContaining({ status: { not: "failed" } }),
|
||||
}),
|
||||
);
|
||||
expect(mocks.unlink).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("short-circuits to a viable prior job and unlinks the redundant upload", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(ADMIN_SESSION as never);
|
||||
mocks.importJobFindFirst.mockResolvedValue({
|
||||
id: "live-job",
|
||||
status: "parsing",
|
||||
});
|
||||
|
||||
const res = await POST(multipartReq());
|
||||
expect(res.status).toBe(202);
|
||||
const body = await res.json();
|
||||
expect(body.data.idempotent).toBe(true);
|
||||
expect(body.data.jobId).toBe("live-job");
|
||||
expect(mocks.bossSend).not.toHaveBeenCalled();
|
||||
expect(mocks.importJobCreate).not.toHaveBeenCalled();
|
||||
expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
|
||||
});
|
||||
});
|
||||
@@ -26,6 +26,7 @@ import {
|
||||
type AppleHealthImportPayload,
|
||||
} from "@/lib/jobs/apple-health-import-worker";
|
||||
import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
|
||||
import { unlink } from "node:fs/promises";
|
||||
|
||||
export const dynamic = "force-dynamic";
|
||||
|
||||
@@ -112,15 +113,22 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
|
||||
// Same content-hash idempotency as the user-facing route, but scoped
|
||||
// by the target user — an admin re-uploading the same file for the
|
||||
// same user resolves to the previous job.
|
||||
// same user resolves to the previous job. As on the user route, a
|
||||
// prior `failed` job is excluded (issue #486) so the same export can
|
||||
// be retried; only still-viable jobs (queued / in-flight / done)
|
||||
// short-circuit.
|
||||
const existing = await prisma.importJob.findFirst({
|
||||
where: {
|
||||
userId: targetUser.id,
|
||||
uploadSha256: uploaded.sha256,
|
||||
status: { not: "failed" },
|
||||
},
|
||||
orderBy: { startedAt: "desc" },
|
||||
});
|
||||
if (existing) {
|
||||
// Redundant staging file — the existing viable job owns the bytes.
|
||||
// Unlink so a deduped admin re-upload does not leak `/tmp`.
|
||||
await unlink(uploaded.filePath).catch(() => {});
|
||||
annotate({ meta: { idempotent_hit: true, job_id: existing.id } });
|
||||
return apiSuccess(
|
||||
{
|
||||
|
||||
@@ -33,6 +33,8 @@ const mocks = vi.hoisted(() => ({
|
||||
importJobCreate: vi.fn(),
|
||||
importJobFindFirst: vi.fn().mockResolvedValue(null),
|
||||
importJobUpdate: vi.fn(),
|
||||
streamToDisk: vi.fn(),
|
||||
unlink: vi.fn().mockResolvedValue(undefined),
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/rate-limit", () => ({
|
||||
@@ -43,6 +45,15 @@ vi.mock("@/lib/jobs/boss-instance", () => ({
|
||||
getGlobalBoss: mocks.getGlobalBoss,
|
||||
}));
|
||||
|
||||
vi.mock("@/lib/multipart/stream-to-disk", () => ({
|
||||
streamMultipartToDisk: mocks.streamToDisk,
|
||||
}));
|
||||
|
||||
vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("node:fs/promises")>();
|
||||
return { ...actual, unlink: mocks.unlink };
|
||||
});
|
||||
|
||||
vi.mock("@/lib/db", () => ({
|
||||
prisma: {
|
||||
importJob: {
|
||||
@@ -61,6 +72,8 @@ const SESSION_OK = {
|
||||
user: { id: "user-1", username: "testuser", role: "USER" as const },
|
||||
};
|
||||
|
||||
const STAGED_PATH = "/tmp/healthlog-apple-health-import-abc.bin";
|
||||
|
||||
beforeEach(() => {
|
||||
vi.resetAllMocks();
|
||||
mocks.checkRateLimit.mockResolvedValue({ allowed: true });
|
||||
@@ -69,8 +82,24 @@ beforeEach(() => {
|
||||
mocks.importJobCreate.mockResolvedValue({ id: "ij-1" });
|
||||
mocks.importJobFindFirst.mockResolvedValue(null);
|
||||
mocks.importJobUpdate.mockResolvedValue({ id: "ij-1" });
|
||||
mocks.streamToDisk.mockResolvedValue({
|
||||
filePath: STAGED_PATH,
|
||||
bytes: 100,
|
||||
sha256: "deadbeef",
|
||||
originalFilename: "export.zip",
|
||||
textFields: {},
|
||||
});
|
||||
mocks.unlink.mockResolvedValue(undefined);
|
||||
});
|
||||
|
||||
function multipartReq(): NextRequest {
|
||||
return new NextRequest("http://localhost/api/import/apple-health-export", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "multipart/form-data; boundary=---x" },
|
||||
body: "stub",
|
||||
});
|
||||
}
|
||||
|
||||
describe("POST /api/import/apple-health-export", () => {
|
||||
it("returns 401 when unauthenticated", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(null);
|
||||
@@ -119,3 +148,53 @@ describe("POST /api/import/apple-health-export", () => {
|
||||
expect(res.status).toBe(429);
|
||||
});
|
||||
});
|
||||
|
||||
describe("POST /api/import/apple-health-export — dedup by content hash (issue #486)", () => {
|
||||
it("re-stages and enqueues when the only prior job for the same bytes is failed", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
|
||||
// The DB holds a FAILED prior job for this sha. The dedup query
|
||||
// excludes failed rows, so it returns null and the upload must fall
|
||||
// through to a fresh stage + boss.send instead of replaying the dead
|
||||
// job's id/failureReason forever.
|
||||
mocks.importJobFindFirst.mockImplementation(async (args: unknown) => {
|
||||
const where = (args as { where?: { status?: { not?: string } } }).where;
|
||||
if (where?.status?.not === "failed") return null;
|
||||
return { id: "old-failed-job", status: "failed" };
|
||||
});
|
||||
|
||||
const res = await POST(multipartReq());
|
||||
expect(res.status).toBe(202);
|
||||
const body = await res.json();
|
||||
expect(body.data.status).toBe("queued");
|
||||
expect(body.data.idempotent).toBeUndefined();
|
||||
expect(body.data.jobId).not.toBe("old-failed-job");
|
||||
expect(mocks.bossSend).toHaveBeenCalledTimes(1);
|
||||
expect(mocks.importJobCreate).toHaveBeenCalledTimes(1);
|
||||
// The dedup lookup must exclude failed rows.
|
||||
expect(mocks.importJobFindFirst).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
where: expect.objectContaining({ status: { not: "failed" } }),
|
||||
}),
|
||||
);
|
||||
// A fresh job was staged — nothing to unlink on this path.
|
||||
expect(mocks.unlink).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("short-circuits to a viable prior job and unlinks the redundant upload", async () => {
|
||||
vi.mocked(getSession).mockResolvedValue(SESSION_OK as never);
|
||||
mocks.importJobFindFirst.mockResolvedValue({
|
||||
id: "live-job",
|
||||
status: "parsing",
|
||||
});
|
||||
|
||||
const res = await POST(multipartReq());
|
||||
expect(res.status).toBe(202);
|
||||
const body = await res.json();
|
||||
expect(body.data.idempotent).toBe(true);
|
||||
expect(body.data.jobId).toBe("live-job");
|
||||
expect(mocks.bossSend).not.toHaveBeenCalled();
|
||||
expect(mocks.importJobCreate).not.toHaveBeenCalled();
|
||||
// The freshly staged upload is redundant and must be cleaned up.
|
||||
expect(mocks.unlink).toHaveBeenCalledWith(STAGED_PATH);
|
||||
});
|
||||
});
|
||||
@@ -12,7 +12,10 @@
|
||||
* the bytes) short-circuits to the previous `ImportJob` id without
|
||||
* re-queueing — the iOS client cannot reliably emit a stable
|
||||
* `Idempotency-Key` for a 1 GB file, so we hash the bytes and dedup
|
||||
* on content.
|
||||
* on content. The dedup only covers still-viable jobs (queued,
|
||||
* in-flight, or already `done`); a prior `failed` job is NOT matched,
|
||||
* so the same export can always be retried with the same bytes
|
||||
* (issue #486 — a failed row must never tombstone its own hash).
|
||||
*
|
||||
* Locks per `.planning/research/v1434-r-1-xml-import.md` §5.3.
|
||||
*/
|
||||
@@ -31,6 +34,7 @@ import {
|
||||
type AppleHealthImportPayload,
|
||||
} from "@/lib/jobs/apple-health-import-worker";
|
||||
import { streamMultipartToDisk } from "@/lib/multipart/stream-to-disk";
|
||||
import { unlink } from "node:fs/promises";
|
||||
|
||||
export const dynamic = "force-dynamic";
|
||||
|
||||
@@ -104,15 +108,25 @@ export const POST = apiHandler(async (request: NextRequest) => {
|
||||
}
|
||||
|
||||
// Content-hash idempotency: a re-upload of the same bytes resolves
|
||||
// to the previous ImportJob row instead of re-queueing.
|
||||
// to the previous ImportJob row instead of re-queueing — but ONLY
|
||||
// when that prior job is still viable (queued / in-flight / done). A
|
||||
// `failed` prior job must NOT short-circuit (issue #486): matching it
|
||||
// returned its stale failureReason forever and made retrying the same
|
||||
// export impossible. Excluding `failed` lets the same bytes fall
|
||||
// through to a fresh stage + enqueue.
|
||||
const existing = await prisma.importJob.findFirst({
|
||||
where: {
|
||||
userId: user.id,
|
||||
uploadSha256: uploaded.sha256,
|
||||
status: { not: "failed" },
|
||||
},
|
||||
orderBy: { startedAt: "desc" },
|
||||
});
|
||||
if (existing) {
|
||||
// The existing viable job owns the canonical bytes; the upload we
|
||||
// just streamed to `/tmp` is redundant. Unlink it so a deduped
|
||||
// re-upload does not leak a gigabyte-scale staging file.
|
||||
await unlink(uploaded.filePath).catch(() => {});
|
||||
annotate({ meta: { idempotent_hit: true, job_id: existing.id } });
|
||||
return apiSuccess(
|
||||
{
|
||||
|
||||
Reference in new issue
Block a user