Optimize terrain rendering for large exports

worker/main.ts

@@ -15,8 +15,13 @@ const client = createClient<Database>(
 );
 const worker = "worker-" + randomUUID();
 let stopping = false;
-process.on("SIGINT", () => (stopping = true));
-process.on("SIGTERM", () => (stopping = true));
+let activeAbort: AbortController | null = null;
+function stopWorker() {
+  stopping = true;
+  activeAbort?.abort();
+}
+process.on("SIGINT", stopWorker);
+process.on("SIGTERM", stopWorker);
 function check<T>(r: {
   data: T;
   error: { message: string } | null;
@@ -31,6 +36,7 @@ async function processJob(job: {
 }) {
   const abort = new AbortController(),
     directory = `.cache/jobs/${job.id}`;
+  activeAbort = abort;
   await mkdir(directory, { recursive: true });
   let lastProgressWrite = 0;
   let lastLoggedAt = 0;
@@ -146,18 +152,30 @@ async function processJob(job: {
         )
       )
         throw new Error("Snapshot bounds disagree with manifest");
-      for (const b of terrain.blocks) {
+      for (let i = 0; i < terrain.blocks.length; i++) {
+        const b = terrain.blocks[i];
         if (
           !Array.isArray(b) ||
           b.length < 4 ||
-          !b.slice(0, 3).every((v) => Number.isInteger(v)) ||
+          !Number.isSafeInteger(b[0]) ||
+          !Number.isSafeInteger(b[1]) ||
+          !Number.isSafeInteger(b[2]) ||
+          b[0] < terrain.bounds.minX ||
+          b[0] > terrain.bounds.maxX ||
+          b[2] < terrain.bounds.minZ ||
+          b[2] > terrain.bounds.maxZ ||
+          b[1] < -4096 ||
+          b[1] > 4096 ||
           typeof b[3] !== "string" ||
           !/^minecraft:[a-z0-9_]+(\[[a-z0-9_=,]+\])?$/.test(b[3])
         )
           throw new Error("Invalid block record");
+        // Avoid starving lease heartbeats while validating millions of records.
+        if (i % 65536 === 0)
+          await new Promise<void>((resolve) => setImmediate(resolve));
       }
       console.log(
-        `Job ${job.id}: ${terrain.blocks.length.toLocaleString()} blocks, ${manifest.viewpoints.length} panoramas; software rendering may take time.`,
+        `Job ${job.id}: ${terrain.blocks.length.toLocaleString()} blocks, ${manifest.viewpoints.length} panoramas; starting cached geometry builder.`,
       );
       const report = await renderSnapshot(
         terrain,
@@ -217,28 +235,43 @@ async function processJob(job: {
     const images = (await readdir(output, { recursive: true })).filter(
       (entry) => entry.endsWith(".webp"),
     );
-    let uploaded = 0;
-    for (const entry of images) {
-      if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
-      const normalized = entry.replaceAll("\\", "/");
-      check(
-        await client.storage
-          .from("minegeo-media")
-          .upload(
-            `${job.dataset_id}/${normalized}`,
-            await readFile(`${output}/${entry}`),
-            {
-              contentType: "image/webp",
-              upsert: true,
-              cacheControl: "31536000",
-            },
-          ),
-      );
-      await progress(
-        90 + (9 * ++uploaded) / images.length,
-        `Saved ${uploaded}/${images.length} processed images`,
-      );
-    }
+    let uploaded = 0,
+      nextImage = 0;
+    let uploadError: unknown;
+    // Small bounded pool avoids paying Storage round-trip latency serially.
+    // Finish every in-flight operation before cleanup/retry can touch the files.
+    await Promise.all(
+      Array.from({ length: Math.min(4, images.length) }, async () => {
+        try {
+          while (nextImage < images.length && !uploadError) {
+            if (abort.signal.aborted)
+              throw new Error("Lease lost or cancelled");
+            const entry = images[nextImage++],
+              normalized = entry.replaceAll("\\", "/");
+            check(
+              await client.storage
+                .from("minegeo-media")
+                .upload(
+                  `${job.dataset_id}/${normalized}`,
+                  await readFile(`${output}/${entry}`),
+                  {
+                    contentType: "image/webp",
+                    upsert: true,
+                    cacheControl: "31536000",
+                  },
+                ),
+            );
+            await progress(
+              90 + (9 * ++uploaded) / images.length,
+              `Saved ${uploaded}/${images.length} processed images`,
+            );
+          }
+        } catch (error) {
+          uploadError ??= error;
+        }
+      }),
+    );
+    if (uploadError) throw uploadError;
     if (abort.signal.aborted) throw new Error("Lease lost or cancelled");
     check(
       await client.rpc("complete_job", {
@@ -253,11 +286,18 @@ async function processJob(job: {
     const message =
       error instanceof Error ? error.message : "Processing failed";
     console.error(`Job ${job.id}: ${message}`);
-    const state = job.attempts >= 3 ? "failed" : "queued";
+    const state = stopping ? "queued" : job.attempts >= 3 ? "failed" : "queued";
     const rows = check(
       await client
         .from("upload_jobs")
-        .update({ state, error: message, lease_until: null, progress: 0 })
+        .update({
+          state,
+          error: stopping ? null : message,
+          lease_owner: null,
+          lease_until: null,
+          progress: 0,
+          attempts: stopping ? Math.max(0, job.attempts - 1) : job.attempts,
+        })
         .eq("id", job.id)
         .eq("lease_owner", worker)
         .eq("state", "running")
@@ -270,6 +310,7 @@ async function processJob(job: {
         .eq("id", job.dataset_id)
         .eq("state", "processing");
   } finally {
+    if (activeAbort === abort) activeAbort = null;
     clearInterval(heartbeat);
     await rm(directory, { recursive: true, force: true });
   }