fix(studio): recover uncertain imports with persistent request identities
CI / check (push) Successful in 3m12s
CI / deploy (push) Successful in 19s
CI / publish-container (push) Successful in 1m12s

This commit is contained in:
Simo committed 2026-09-13 18:33:06 +02:00
1 parent afea80708c
commit c977fe95ba
12 files changed
+365 -31

No files matched your search

@@ -7,6 +7,7 @@ const mocks = vi.hoisted(() => ({
attachment: vi.fn(),
guard: vi.fn(),
create: vi.fn(),
read: vi.fn(),
list: vi.fn(),
page: vi.fn(),
retry: vi.fn(),
@@ -33,6 +34,7 @@ vi.mock("@/lib/services/furni-job-store", async (original) => ({
...(await original<typeof import("@/lib/services/furni-job-store")>()),
ImportJobStore: class {
create = mocks.create;
read = mocks.read;
list = mocks.list;
page = mocks.page;
retry = mocks.retry;
@@ -64,6 +66,9 @@ const request = (body: unknown) =>
});
beforeEach(() => {
mocks.create.mockClear();
mocks.read.mockRejectedValue(
Object.assign(Error("missing"), { code: "ENOENT" }),
);
mocks.list.mockClear();
mocks.after.mockClear();
mocks.ping.mockResolvedValue("PONG");
@@ -234,3 +239,36 @@ it("rejects invalid recovery attachment paths", async () => {
).toBe(400);
expect(mocks.retry).not.toHaveBeenCalled();
});
it("recovers accepted requests even when Redis and the source are unavailable", async () => {
const body = { id: randomUUID(), sourceId: "source", items: [item] };
mocks.source.mockResolvedValue({ id: "source" });
const accepted = await POST(request(body), ctx);
const saved = (await accepted.json()).job;
mocks.read.mockResolvedValue({ ...saved, state: "completed" });
mocks.ping.mockRejectedValue(Error("offline"));
mocks.source.mockResolvedValue(null);
const replay = await POST(request(body), ctx);
expect(replay.status).toBe(200);
expect((await replay.json()).job).toMatchObject({
id: body.id,
state: "completed",
});
expect(mocks.create).toHaveBeenCalledTimes(1);
});
it("rejects changed content for an accepted request instead of reporting false success", async () => {
const body = { id: randomUUID(), items: [item] };
const accepted = await POST(request(body), ctx);
mocks.read.mockResolvedValue((await accepted.json()).job);
const replay = await POST(request({ ...body, translate: true }), ctx);
expect(replay.status).toBe(409);
expect(mocks.create).toHaveBeenCalledTimes(1);
});
it("does not recover a different owner's request", async () => {
const body = { id: randomUUID(), items: [item] };
const accepted = await POST(request(body), ctx);
mocks.read.mockResolvedValue({ ...(await accepted.json()).job, userId: 99 });
const replay = await POST(request(body), ctx);
expect(replay.status).toBe(409);
expect((await replay.json()).job).toBeUndefined();
});
+48 -16
View File
@@ -1,3 +1,4 @@
import { createHash } from "node:crypto";
import { after } from "next/server";
import { z } from "zod";
import { apiError, apiJson, apiOk } from "@/lib/api";
@@ -8,7 +9,11 @@ import { PERMS } from "@/lib/permission-slugs";
import { redis } from "@/lib/redis";
import { getSource } from "@/lib/services/clone-sources";
import { readFurnitureAttachment } from "@/lib/services/furni-attachment";
import { ImportJobStore, validJobId } from "@/lib/services/furni-job-store";
import {
ImportJobStore,
ImportRequestConflict,
validJobId,
} from "@/lib/services/furni-job-store";
import { drainFurnitureImports } from "@/lib/services/furni-job-worker";
const schema = z.object({
@@ -78,28 +83,55 @@ export const POST = withAdmin(
"Invalid furniture import request (maximum 500 items)",
400,
);
const store = new ImportJobStore(),
body = parsed.data;
const unique = [
...new Map(body.items.map((item) => [item.classname, item])).values(),
];
// Hash validated request fields before worker checkpoints can change the job.
const { id: _id, ...input } = body;
const requestFingerprint = createHash("sha256")
.update(JSON.stringify({ ...input, items: unique }))
.digest("hex");
const existing = await store.read(body.id).catch((error) => {
if (error.code === "ENOENT") return null;
throw error;
});
if (existing) {
// Legacy jobs have no immutable fingerprint: never infer equivalence from progress.
if (
existing.userId !== ctx.session.user.id ||
existing.requestFingerprint !== requestFingerprint
)
return apiError("Request ID already used", 409);
after(drainFurnitureImports);
return apiOk({ job: existing });
}
if (!redis || (await redis.ping().catch(() => null)) !== "PONG")
return apiError(
"Background import queue is temporarily unavailable. Please retry shortly.",
503,
);
if (parsed.data.sourceId && !(await getSource(parsed.data.sourceId)))
if (body.sourceId && !(await getSource(body.sourceId)))
return apiError("Source not found", 404);
const store = new ImportJobStore(),
body = parsed.data;
const createdAt = new Date().toISOString();
const unique = [
...new Map(body.items.map((item) => [item.classname, item])).values(),
];
const job = await store.create({
...body,
userId: ctx.session.user.id,
operationId: getRequestId(),
createdAt,
updatedAt: createdAt,
state: "queued",
items: unique.map((item) => ({ ...item, state: "pending" })),
});
let job: Awaited<ReturnType<ImportJobStore["create"]>>;
try {
job = await store.create({
...body,
requestFingerprint,
userId: ctx.session.user.id,
operationId: getRequestId(),
createdAt,
updatedAt: createdAt,
state: "queued",
items: unique.map((item) => ({ ...item, state: "pending" })),
});
} catch (error) {
if (error instanceof ImportRequestConflict)
return apiError("Request ID already used", 409);
throw error;
}
after(drainFurnitureImports);
return apiOk({ job });
},