fix: faster, cheaper Loom imports: fewer progress webhooks and real restart checks

apps/media-server/src/__tests__/loom-importer/app.test.ts

@@ -95,6 +95,20 @@ describe("Loom importer API", () => {
 			videoId: "second",
 			phase: "queued",
 		});
+		const listing = (await (
+			await app.request("/loom-importer/video/status", {
+				headers: { "x-media-server-secret": "importer-secret" },
+			})
+		).json()) as {
+			jobs: Array<{ jobId: string; videoId: string; phase: string }>;
+		};
+		expect(listing.jobs).toContainEqual(
+			expect.objectContaining({
+				jobId: second.jobId,
+				videoId: "second",
+				phase: "queued",
+			}),
+		);
 		const full = await post(app, body("third"));
 		expect(full.status).toBe(503);
 		expect(await full.json()).toMatchObject({ code: "SERVER_BUSY" });

apps/media-server/src/__tests__/loom-importer/progress-webhooks.test.ts

@@ -0,0 +1,44 @@
+import { describe, expect, test } from "bun:test";
+import { ProgressWebhookThrottle } from "../../loom-importer/job";
+
+describe("ProgressWebhookThrottle", () => {
+	test("sends the first update for a job", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		expect(throttle.shouldSend("job-1", "downloading", 0)).toBe(true);
+	});
+
+	test("holds same-phase updates inside the interval", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		throttle.shouldSend("job-1", "downloading", 0);
+		expect(throttle.shouldSend("job-1", "downloading", 1_000)).toBe(false);
+		expect(throttle.shouldSend("job-1", "downloading", 14_999)).toBe(false);
+	});
+
+	test("sends again once the interval passes", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		throttle.shouldSend("job-1", "downloading", 0);
+		expect(throttle.shouldSend("job-1", "downloading", 15_000)).toBe(true);
+		expect(throttle.shouldSend("job-1", "downloading", 20_000)).toBe(false);
+		expect(throttle.shouldSend("job-1", "downloading", 30_000)).toBe(true);
+	});
+
+	test("sends a phase change straight away", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		throttle.shouldSend("job-1", "downloading", 0);
+		expect(throttle.shouldSend("job-1", "uploading", 1)).toBe(true);
+		expect(throttle.shouldSend("job-1", "uploading", 2)).toBe(false);
+	});
+
+	test("tracks jobs separately", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		throttle.shouldSend("job-1", "downloading", 0);
+		expect(throttle.shouldSend("job-2", "downloading", 1)).toBe(true);
+	});
+
+	test("forgets finished jobs", () => {
+		const throttle = new ProgressWebhookThrottle(15_000);
+		throttle.shouldSend("job-1", "downloading", 0);
+		throttle.forget("job-1");
+		expect(throttle.shouldSend("job-1", "downloading", 1)).toBe(true);
+	});
+});

apps/media-server/src/loom-importer/app.ts

@@ -151,7 +151,16 @@ export function createLoomImporter(
 					phase: job.phase,
 					progress: job.progress,
 					ageSeconds: Math.round((Date.now() - job.createdAt) / 1000),
-				})),
+				}))
+				.concat(
+					queue.map((entry) => ({
+						jobId: entry.jobId,
+						videoId: entry.request.videoId,
+						phase: "queued" as Job["phase"],
+						progress: 0,
+						ageSeconds: 0,
+					})),
+				),
 		});
 	});
 

apps/media-server/src/loom-importer/job.ts

@@ -140,14 +140,35 @@ async function deliverTerminalWebhook(job: Job, attempts: number) {
 	);
 }
 
+export class ProgressWebhookThrottle {
+	private readonly last = new Map<string, { at: number; phase: string }>();
+
+	constructor(private readonly intervalMs: number) {}
+
+	shouldSend(jobId: string, phase: string, now: number) {
+		const last = this.last.get(jobId);
+		if (last && last.phase === phase && now - last.at < this.intervalMs)
+			return false;
+		this.last.set(jobId, { at: now, phase });
+		return true;
+	}
+
+	forget(jobId: string) {
+		this.last.delete(jobId);
+	}
+}
+
+const progressWebhooks = new ProgressWebhookThrottle(15_000);
+
 function progressUpdate(
 	jobId: string,
 	phase: Job["phase"],
 	progress: number,
 	message: string,
 ) {
 	const job = updateJob(jobId, { phase, progress, message });
-	if (job) void sendWebhook(job);
+	if (job && progressWebhooks.shouldSend(jobId, phase, Date.now()))
+		void sendWebhook(job);
 }
 
 interface PreparedSource {
@@ -429,6 +450,7 @@ export async function runLoomImport(
 		if (failed)
 			await deliverTerminalWebhook(failed, settings.terminalWebhookAttempts);
 	} finally {
+		progressWebhooks.forget(job.jobId);
 		if (workDir)
 			await rm(workDir, { recursive: true, force: true }).catch(() => {});
 	}

apps/web/__tests__/unit/loom-import-start.test.ts

@@ -1,6 +1,7 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const mocks = vi.hoisted(() => ({
+	importerJobState: vi.fn(),
 	where: vi.fn(),
 	set: vi.fn(),
 	queue: vi.fn(),
@@ -37,6 +38,7 @@ vi.mock("workflow/api", () => ({ start: mocks.start, getRun: mocks.getRun }));
 vi.mock("workflow/runtime", () => ({ getWorld: mocks.getWorld }));
 vi.mock("@/lib/loom-media-dispatch", () => ({
 	isLoomMediaJobFailed: mocks.mediaJobFailed,
+	getLoomImporterJobState: mocks.importerJobState,
 }));
 vi.mock("@/workflows/import-loom-video", () => ({
 	importLoomVideoWorkflow: vi.fn(),
@@ -83,6 +85,73 @@ describe("Loom workflow startup receipts", () => {
 		);
 	});
 
+	it("treats a run that was never created as inactive", async () => {
+		const { WorkflowRunNotFoundError } = await import(
+			"workflow/internal/errors"
+		);
+		mocks.getRun.mockReturnValue({
+			status: Promise.reject(new WorkflowRunNotFoundError("run-1")),
+		});
+		expect(
+			await getLoomImportRunState({ runId: "run-1", dispatch: "uncertain" }),
+		).toBe("inactive");
+		mocks.getRun.mockReturnValue({
+			status: Promise.reject(new Error("network")),
+		});
+		expect(
+			await getLoomImportRunState({ runId: "run-1", dispatch: "uncertain" }),
+		).toBe("unconfirmed");
+	});
+
+	it.each([
+		["running", "active"],
+		["finished", "inactive"],
+		["missing", "inactive"],
+		["unknown", "unconfirmed"],
+	] as const)(
+		"asks the importer about its own jobs: %s",
+		async (importerState, expected) => {
+			mocks.getRun.mockReturnValue({ status: Promise.resolve("failed") });
+			mocks.importerJobState.mockResolvedValue(importerState);
+			mocks.mediaJobFailed.mockResolvedValue(false);
+			expect(
+				await getLoomImportRunState({
+					runId: "run-1",
+					dispatch: "accepted",
+					mediaDispatch: {
+						attemptId: "run-1:0",
+						videoId: "video-1",
+						jobId: "importer-1",
+						state: "accepted",
+						worker: "importer",
+					},
+				}),
+			).toBe(expected);
+			expect(mocks.importerJobState).toHaveBeenCalledWith({
+				jobId: "importer-1",
+				videoId: "video-1",
+			});
+		},
+	);
+
+	it("does not trust a missing importer job for media server dispatches", async () => {
+		mocks.getRun.mockReturnValue({ status: Promise.resolve("failed") });
+		mocks.importerJobState.mockResolvedValue("missing");
+		mocks.mediaJobFailed.mockResolvedValue(false);
+		expect(
+			await getLoomImportRunState({
+				runId: "run-1",
+				dispatch: "accepted",
+				mediaDispatch: {
+					attemptId: "run-1:0",
+					videoId: "video-1",
+					jobId: "worker-1",
+					state: "accepted",
+				},
+			}),
+		).toBe("unconfirmed");
+	});
+
 	it.each([false, true])(
 		"requires a matching failed worker before releasing an accepted dispatch: %s",
 		async (failed) => {

apps/web/__tests__/unit/loom-importer-job-state.test.ts

@@ -0,0 +1,79 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+vi.mock("@cap/env", () => ({
+	serverEnv: () => ({
+		MEDIA_SERVER_URL: "https://worker.example.com",
+		LOOM_IMPORTER_URL: "https://media.example.com/loom-importer",
+		MEDIA_SERVER_WEBHOOK_SECRET: "fixture-secret",
+	}),
+}));
+vi.mock("@cap/database", () => ({ db: vi.fn() }));
+
+import { getLoomImporterJobState } from "@/lib/loom-media-dispatch";
+
+describe("getLoomImporterJobState", () => {
+	afterEach(() => vi.unstubAllGlobals());
+
+	it.each([
+		[{ jobId: "job", videoId: "video", phase: "downloading" }, "running"],
+		[{ jobId: "job", videoId: "video", phase: "queued" }, "running"],
+		[{ jobId: "job", videoId: "video", phase: "complete" }, "finished"],
+		[{ jobId: "job", videoId: "video", phase: "error" }, "finished"],
+		[{ jobId: "job", videoId: "video", phase: "cancelled" }, "finished"],
+		[{ jobId: "job", videoId: "other", phase: "error" }, "unknown"],
+	])("reads the importer's job %j", async (body, expected) => {
+		const fetch = vi.fn().mockResolvedValue(Response.json(body));
+		vi.stubGlobal("fetch", fetch);
+		await expect(
+			getLoomImporterJobState({ jobId: "job", videoId: "video" }),
+		).resolves.toBe(expected);
+		expect(fetch).toHaveBeenCalledWith(
+			"https://media.example.com/loom-importer/video/process/job/status",
+			expect.objectContaining({ redirect: "error" }),
+		);
+	});
+
+	it("knows a job the importer no longer has", async () => {
+		vi.stubGlobal(
+			"fetch",
+			vi.fn().mockResolvedValue(new Response("missing", { status: 404 })),
+		);
+		await expect(
+			getLoomImporterJobState({ jobId: "job", videoId: "video" }),
+		).resolves.toBe("missing");
+	});
+
+	it.each([
+		[() => Promise.resolve(new Response("down", { status: 503 }))],
+		[() => Promise.reject(new Error("network"))],
+	])("cannot tell when the importer does not answer: %#", async (respond) => {
+		vi.stubGlobal("fetch", vi.fn().mockImplementation(respond));
+		await expect(
+			getLoomImporterJobState({ jobId: "job", videoId: "video" }),
+		).resolves.toBe("unknown");
+	});
+
+	it("looks a video up in the importer's job list when there is no job id", async () => {
+		const fetch = vi.fn().mockResolvedValue(
+			Response.json({
+				jobs: [{ jobId: "other", videoId: "other-video", phase: "queued" }],
+			}),
+		);
+		vi.stubGlobal("fetch", fetch);
+		await expect(getLoomImporterJobState({ videoId: "video" })).resolves.toBe(
+			"missing",
+		);
+		fetch.mockResolvedValue(
+			Response.json({
+				jobs: [{ jobId: "j", videoId: "video", phase: "queued" }],
+			}),
+		);
+		await expect(getLoomImporterJobState({ videoId: "video" })).resolves.toBe(
+			"running",
+		);
+		expect(fetch).toHaveBeenCalledWith(
+			"https://media.example.com/loom-importer/video/status",
+			expect.anything(),
+		);
+	});
+});

apps/web/__tests__/unit/loom-importer-redrive.test.ts

@@ -149,7 +149,7 @@ describe("redriveLoomImport", () => {
 		[[], "not an unfinished Loom import"],
 		[[failedImport({ phase: "uploading" })], "phase uploading"],
 		[
-			[failedImport({ phase: "processing", uploadUpdatedAt: hoursAgo(30) })],
+			[failedImport({ phase: "processing", uploadUpdatedAt: hoursAgo(0.1) })],
 			"phase processing",
 		],
 	])("skips imports that are not failed: %#", async (rows, reason) => {
@@ -192,14 +192,34 @@ describe("redriveLoomImport", () => {
 		expect(mocks.updates).toEqual([]);
 	});
 
+	it("restarts an import left processing when nothing is running it", async () => {
+		mocks.rows = [
+			failedImport({ phase: "processing", uploadUpdatedAt: hoursAgo(1) }),
+		];
+		await expect(redriveLoomImport("video", now)).resolves.toMatchObject({
+			status: "restarted",
+		});
+	});
+
+	it("leaves an import processing while its run is active", async () => {
+		mocks.runState.mockResolvedValue("active");
+		mocks.rows = [
+			failedImport({ phase: "processing", uploadUpdatedAt: hoursAgo(1) }),
+		];
+		await expect(redriveLoomImport("video", now)).resolves.toMatchObject({
+			status: "skipped",
+		});
+		expect(mocks.start).not.toHaveBeenCalled();
+	});
+
 	it("only restarts an unconfirmed worker once it can no longer be running", async () => {
 		mocks.runState.mockResolvedValue("unconfirmed");
-		mocks.rows = [failedImport({ uploadUpdatedAt: hoursAgo(2) })];
+		mocks.rows = [failedImport({ uploadUpdatedAt: hoursAgo(0.5) })];
 		await expect(redriveLoomImport("video", now)).resolves.toMatchObject({
 			status: "skipped",
 			reason: "run unconfirmed",
 		});
-		mocks.rows = [failedImport({ uploadUpdatedAt: hoursAgo(7) })];
+		mocks.rows = [failedImport({ uploadUpdatedAt: hoursAgo(2) })];
 		await expect(redriveLoomImport("video", now)).resolves.toMatchObject({
 			status: "restarted",
 		});

apps/web/lib/loom-import-redrive.ts

@@ -9,7 +9,8 @@ import {
 	startLoomImportWorkflow,
 } from "@/lib/loom-import-start";
 
-const FINISHED_WORKER_AGE_MS = 6 * 60 * 60 * 1000;
+const ABANDONED_PROCESSING_AGE_MS = 10 * 60 * 1000;
+const FINISHED_WORKER_AGE_MS = 60 * 60 * 1000;
 const STALE_PROCESSING_AGE_MS = 7 * 24 * 60 * 60 * 1000;
 
 export type LoomRedriveOutcome =
@@ -58,9 +59,13 @@ export async function redriveLoomImport(
 	const stalled =
 		(row.phase === "processing" || row.phase === "uploading") &&
 		age > STALE_PROCESSING_AGE_MS;
-	if (row.phase !== "error" && !stalled)
-		return { videoId, status: "skipped", reason: `phase ${row.phase}` };
 	const runState = await getLoomImportRunState(row.metadata?.loomImportRun);
+	const abandoned =
+		row.phase === "processing" &&
+		runState === "inactive" &&
+		age > ABANDONED_PROCESSING_AGE_MS;
+	if (row.phase !== "error" && !stalled && !abandoned)
+		return { videoId, status: "skipped", reason: `phase ${row.phase}` };
 	if (runState === "pending" || runState === "active")
 		return { videoId, status: "skipped", reason: `run ${runState}` };
 	if (runState === "unconfirmed" && age < FINISHED_WORKER_AGE_MS)