remco/EpicNext-Cms
1
1 Fork 0
Code Issues Pull Requests 1 Actions 1 Packages Projects Releases Wiki Activity

feat(import): durable batch checkpoint and transient auto-retry
CI / check (push) Successful in 2m24s
Details
CI / deploy (push) Successful in 1m32s
Details
CI / publish-container (push) Successful in 47s
Details

Browse Source 
Mirror interactive batch runs into the import-job store so interrupted imports (restart, time-out, disconnect) can be resumed from Import History. Items are checkpointed as they settle (coalesced, serialized saves) and the mirror starts 'running' so the boot-time worker marks it 'interrupted' instead of double-importing; done items are never re-imported. Add bounded backoff retry for transient download/connection failures before marking an item failed, and point the client's time-out/network toasts at Import History.
This commit is contained in:
openhands committed 2026-09-13 12:50:17 +02:00
1 parent af1a06b0a3
commit ffcd4232d6
4 files changed
+263 -26
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/app/api/admin/import/furni/batch/route.ts
+143 -24
View File
@@ -1,5 +1,8 @@
import { randomUUID } from "node:crypto";
import { apiError } from "@/lib/api";
import { withAdmin } from "@/lib/api-handler";
import { getRequestId } from "@/lib/foundation/request-context";
import type { ImportJob } from "@/lib/furni/import-job";
import { PERMS } from "@/lib/permissions";
import { logAudit } from "@/lib/services/audit";
import { getSource } from "@/lib/services/clone-sources";
@@ -20,6 +23,8 @@ import {
verifyAndFixInteractionModesCount,
} from "@/lib/services/furni-import";
import { clearFurniImportCache } from "@/lib/services/furni-import-cache";
import { ImportJobStore } from "@/lib/services/furni-job-store";
import { withTransientImportRetry } from "@/lib/services/import/transient-retry";
import { rcon } from "@/lib/services/rcon";
import type { ImportSingleResult } from "@/types/furni";
@@ -68,6 +73,49 @@ export const POST = withAdmin(
await ensureDirectories();
const encoder = new TextEncoder();
// Durable checkpoint: mirror this run into the import-job store so an
// interrupted batch can be resumed from Import History after a restart,
// time-out or disconnect. The job starts "running" so the boot-time
// worker drain marks it "interrupted" instead of double-importing.
const store = new ImportJobStore();
const createdAt = new Date().toISOString();
const mirrorJob: ImportJob | null = await (async () => {
try {
const job: ImportJob = {
id: randomUUID(),
userId: ctx.session.user.id,
operationId: getRequestId(),
createdAt,
updatedAt: createdAt,
state: "running",
sourceId,
translate: body.translate === true,
langs:
Array.isArray(body.langs) && body.langs.length > 0
? body.langs
: undefined,
items: items.map((item) => ({
id: item.id ?? 0,
classname: item.classname,
name: item.name,
description: item.description ?? "",
type: item.type === "wallitem" ? "wallitem" : "flooritem",
revision: item.revision ?? 0,
category: item.category ?? "unknown",
state: "pending",
})),
};
await store.save(job);
return job;
} catch (error) {
console.warn(
"[import-furni] Could not create durable batch job",
error,
);
return null;
}
})();
// Wire client disconnect to abort controller so we stop processing
// when the user navigates away or closes the browser.
const abortController = new AbortController();
@@ -90,6 +138,45 @@ export const POST = withAdmin(
}
};
// Checkpoint writes are coalesced and chained so concurrent item
// settles can't interleave saves on the same job file; the final
// state is always flushed by finalizeMirror.
const mirrorEntry = (classname: string) =>
mirrorJob?.items.find((m) => m.classname === classname);
let saveChain: Promise<void> = Promise.resolve();
let saveQueued = false;
const checkpoint = () => {
if (!mirrorJob || saveQueued) return;
saveQueued = true;
saveChain = saveChain.then(async () => {
try {
await store.save(mirrorJob);
} catch (error) {
console.warn("[import-furni] Checkpoint write failed", error);
} finally {
saveQueued = false;
}
});
};
const finalizeMirror = async (state: "completed" | "interrupted") => {
if (!mirrorJob) return;
if (state === "interrupted")
for (const entry of mirrorJob.items)
if (entry.state === "running") {
entry.state = "interrupted";
entry.error =
"Import stream was interrupted before finishing. Check imported data before resuming it from history.";
}
mirrorJob.state = state;
mirrorJob.updatedAt = new Date().toISOString();
await saveChain;
try {
await store.save(mirrorJob);
} catch (error) {
console.warn("[import-furni] Final checkpoint write failed", error);
}
};
const startTime = Date.now();
send({ type: "batch_start", total: items.length, concurrency });
@@ -116,37 +203,47 @@ export const POST = withAdmin(
index,
});
const entry = mirrorEntry(item.classname);
if (entry) {
entry.state = "running";
checkpoint();
}
try {
// Coalesce micro-step progress events to at most one per
// ~120ms per item so large imports don't flood the client
// (terminal states are always emitted by importSingleFurni
// and sent below).
let lastProgressSent = 0;
const result: ImportSingleResult = await importSingleFurni({
id: item.id ?? 0,
classname: item.classname,
name: item.name,
description: item.description ?? "",
type: item.type ?? "flooritem",
revision: item.revision ?? 0,
category: item.category ?? "unknown",
skipFurniDataWrite: false,
repairExisting: body.repairExisting === true,
sourceSwfBaseUrl: source?.sourceSwfBaseUrl,
nitroBaseUrl: source?.nitroBaseUrl,
iconBaseUrl: source?.iconBaseUrl,
onProgress: (status: string) => {
const now = Date.now();
if (now - lastProgressSent < 120) return;
lastProgressSent = now;
send({
type: "item_progress",
const result: ImportSingleResult = await withTransientImportRetry(
() =>
importSingleFurni({
id: item.id ?? 0,
classname: item.classname,
status,
index,
});
},
});
name: item.name,
description: item.description ?? "",
type: item.type ?? "flooritem",
revision: item.revision ?? 0,
category: item.category ?? "unknown",
skipFurniDataWrite: false,
repairExisting: body.repairExisting === true,
sourceSwfBaseUrl: source?.sourceSwfBaseUrl,
nitroBaseUrl: source?.nitroBaseUrl,
iconBaseUrl: source?.iconBaseUrl,
onProgress: (status: string) => {
const now = Date.now();
if (now - lastProgressSent < 120) return;
lastProgressSent = now;
send({
type: "item_progress",
classname: item.classname,
status,
index,
});
},
}),
2,
);
if (result.ok) {
succeeded++;
@@ -154,6 +251,14 @@ export const POST = withAdmin(
if (result.furniDataEntry)
furniDataEntries.push(result.furniDataEntry);
if (entry) {
entry.state = "done";
entry.itemId = result.itemId;
entry.warnings =
result.warnings.length > 0 ? result.warnings : undefined;
checkpoint();
}
send({
type: "item_progress",
classname: item.classname,
@@ -180,6 +285,11 @@ export const POST = withAdmin(
});
} else {
failed++;
if (entry) {
entry.state = "failed";
entry.error = result.error;
checkpoint();
}
send({
type: "item_progress",
classname: item.classname,
@@ -190,6 +300,11 @@ export const POST = withAdmin(
}
} catch (err) {
failed++;
if (entry) {
entry.state = "failed";
entry.error = (err as Error).message;
checkpoint();
}
send({
type: "item_progress",
classname: item.classname,
@@ -210,6 +325,7 @@ export const POST = withAdmin(
}
if (aborted) {
await finalizeMirror("interrupted");
send({
type: "batch_complete",
succeeded,
@@ -336,6 +452,8 @@ export const POST = withAdmin(
const { catalogNameFixed, haveOfferFixed, costCreditsFixed } =
await fixDatabaseConsistencyAfterImport();
await finalizeMirror("completed");
send({
type: "batch_complete",
succeeded,
@@ -359,6 +477,7 @@ export const POST = withAdmin(
duration: Date.now() - startTime,
});
} else {
await finalizeMirror("interrupted");
send({
type: "batch_complete",
succeeded,