From 2e84a8ac3e849ba4e941f90d7741a7ec391c4284 Mon Sep 17 00:00:00 2001 From: AsperforMias Date: Mon, 13 Jul 2026 02:41:00 +0800 Subject: [PATCH 1/3] fix(api): retain commands after ambiguous Queue sends --- .../application/api-command-ledger.ts | 104 ++++++++++++++++-- .../platform/cloudflare/create-api-worker.ts | 2 + apps/api/tests/api-command-queue.test.ts | 98 +++++++++++++++++ 3 files changed, 192 insertions(+), 12 deletions(-) diff --git a/apps/api/src/modules/api-command/application/api-command-ledger.ts b/apps/api/src/modules/api-command/application/api-command-ledger.ts index 07fba7d3..cc932a8a 100644 --- a/apps/api/src/modules/api-command/application/api-command-ledger.ts +++ b/apps/api/src/modules/api-command/application/api-command-ledger.ts @@ -1,8 +1,9 @@ import { apiCommandsTable } from "@mosoo/db"; import type { ApiCommandId, ApiCommandKind, ApiCommandRow } from "@mosoo/db"; import { createPlatformId } from "@mosoo/id"; -import { and, eq, lt, or, sql } from "drizzle-orm"; +import { and, asc, eq, lt, or, sql } from "drizzle-orm"; +import { createErrorLogContext, logError } from "../../../platform/cloudflare/logger"; import type { ApiBindings } from "../../../platform/cloudflare/worker-types"; import { getAppDatabase, getD1ChangeCount } from "../../../platform/db/drizzle"; import { currentTimestampMs } from "../../../time"; @@ -12,6 +13,12 @@ export const API_COMMAND_LEASE_MS = 5 * 60 * 1000; export const API_COMMAND_LEASE_RENEWAL_INTERVAL_MS = API_COMMAND_LEASE_MS / 5; +export const API_COMMAND_QUEUE_SEND_FAILED_CODE = "queue_send_failed"; + +const API_COMMAND_QUEUE_SEND_FAILED_MESSAGE = "API command queue send failed."; + +const API_COMMAND_QUEUE_REDRIVE_LIMIT = 100; + export interface EnqueueApiCommandInput { dedupeKey: string; kind: ApiCommandKind; @@ -43,10 +50,14 @@ function toQueueMessage(commandId: ApiCommandId): ApiCommandMessage { async function readApiCommandByDedupeKey( database: D1Database, dedupeKey: string, -): Promise | null> { +): Promise | null> { return ( (await getAppDatabase(database) - .select({ id: apiCommandsTable.id }) + .select({ + id: apiCommandsTable.id, + lastErrorCode: apiCommandsTable.lastErrorCode, + status: apiCommandsTable.status, + }) .from(apiCommandsTable) .where(eq(apiCommandsTable.dedupeKey, dedupeKey)) .limit(1) @@ -54,16 +65,86 @@ async function readApiCommandByDedupeKey( ); } -async function deleteApiCommand(input: { +async function markApiCommandQueueSendFailed(input: { commandId: ApiCommandId; database: D1Database; }): Promise { await getAppDatabase(input.database) - .delete(apiCommandsTable) - .where(eq(apiCommandsTable.id, input.commandId)) + .update(apiCommandsTable) + .set({ + lastErrorCode: API_COMMAND_QUEUE_SEND_FAILED_CODE, + lastErrorMessage: API_COMMAND_QUEUE_SEND_FAILED_MESSAGE, + updatedAt: currentTimestampMs(), + }) + .where(and(eq(apiCommandsTable.id, input.commandId), eq(apiCommandsTable.status, "queued"))) .run(); } +async function clearApiCommandQueueSendFailure(input: { + commandId: ApiCommandId; + database: D1Database; +}): Promise { + await getAppDatabase(input.database) + .update(apiCommandsTable) + .set({ + lastErrorCode: null, + lastErrorMessage: null, + updatedAt: currentTimestampMs(), + }) + .where( + and( + eq(apiCommandsTable.id, input.commandId), + eq(apiCommandsTable.status, "queued"), + eq(apiCommandsTable.lastErrorCode, API_COMMAND_QUEUE_SEND_FAILED_CODE), + ), + ) + .run(); +} + +async function sendApiCommandMessage( + bindings: Pick, + commandId: ApiCommandId, +): Promise { + try { + await bindings.API_COMMAND_QUEUE.send(toQueueMessage(commandId)); + } catch (error) { + // A rejected producer response does not prove that Queue discarded the message. + // Keep the outbox row so a delivered duplicate can still be claimed safely. + await markApiCommandQueueSendFailed({ commandId, database: bindings.DB }); + throw error; + } + + await clearApiCommandQueueSendFailure({ commandId, database: bindings.DB }); +} + +export async function redriveFailedApiCommandEnqueues( + bindings: Pick, +): Promise { + const commands = await getAppDatabase(bindings.DB) + .select({ id: apiCommandsTable.id }) + .from(apiCommandsTable) + .where( + and( + eq(apiCommandsTable.status, "queued"), + eq(apiCommandsTable.lastErrorCode, API_COMMAND_QUEUE_SEND_FAILED_CODE), + ), + ) + .orderBy(asc(apiCommandsTable.id)) + .limit(API_COMMAND_QUEUE_REDRIVE_LIMIT) + .all(); + + for (const command of commands) { + try { + await sendApiCommandMessage(bindings, command.id); + } catch (error) { + logError("api-command.enqueue_redrive_failed", { + ...createErrorLogContext(error), + commandId: command.id, + }); + } + } +} + export async function enqueueApiCommand( bindings: Pick, input: EnqueueApiCommandInput, @@ -93,12 +174,7 @@ export async function enqueueApiCommand( .run(); if (getD1ChangeCount(insertResult) > 0) { - try { - await bindings.API_COMMAND_QUEUE.send(toQueueMessage(commandId)); - } catch (error) { - await deleteApiCommand({ commandId, database: bindings.DB }); - throw error; - } + await sendApiCommandMessage(bindings, commandId); return commandId; } @@ -109,6 +185,10 @@ export async function enqueueApiCommand( throw new Error("API command enqueue could not confirm the ledger row."); } + if (current.status === "queued" && current.lastErrorCode === API_COMMAND_QUEUE_SEND_FAILED_CODE) { + await sendApiCommandMessage(bindings, current.id); + } + return current.id; } diff --git a/apps/api/src/platform/cloudflare/create-api-worker.ts b/apps/api/src/platform/cloudflare/create-api-worker.ts index 4cd066f0..46f5cb65 100644 --- a/apps/api/src/platform/cloudflare/create-api-worker.ts +++ b/apps/api/src/platform/cloudflare/create-api-worker.ts @@ -1,4 +1,5 @@ import { enqueueScheduledMaintenanceCommand } from "../../modules/api-command/application/api-command-enqueue"; +import { redriveFailedApiCommandEnqueues } from "../../modules/api-command/application/api-command-ledger"; import type { ApiCommandMessage } from "../../modules/api-command/application/api-command-message"; import { processApiCommandDeadLetterMessage, @@ -30,6 +31,7 @@ export function createApiWorker(): ExportedHandler { return response; }, async scheduled(controller: ScheduledController, env: ApiBindings): Promise { + await redriveFailedApiCommandEnqueues(env); await enqueueScheduledMaintenanceCommand(env, { scheduledTime: controller.scheduledTime, }); diff --git a/apps/api/tests/api-command-queue.test.ts b/apps/api/tests/api-command-queue.test.ts index a24ace26..f456012d 100644 --- a/apps/api/tests/api-command-queue.test.ts +++ b/apps/api/tests/api-command-queue.test.ts @@ -5,8 +5,10 @@ import { eq } from "drizzle-orm"; import { API_COMMAND_LEASE_MS, + API_COMMAND_QUEUE_SEND_FAILED_CODE, claimApiCommand, enqueueApiCommand, + redriveFailedApiCommandEnqueues, renewApiCommandClaim, } from "../src/modules/api-command/application/api-command-ledger"; import type { ApiCommandMessage } from "../src/modules/api-command/application/api-command-message"; @@ -96,6 +98,102 @@ describe("API command queue", () => { expect(recorded.recorded).toEqual([{ type: "ack" }]); }); + test("keeps a command claimable when Queue accepts it but send reports a timeout", async () => { + const database = await createPublicHttpContractDatabase(); + const retainedMessages: ApiCommandMessage[] = []; + const queue = { + sent: [], + async send(body: ApiCommandMessage): Promise { + retainedMessages.push(body); + throw new Error("Queue response timed out."); + }, + }; + const bindings = createPublicHttpTestBindings(database, { + apiCommandQueue: queue, + }) as ApiBindings; + + await expect( + enqueueApiCommand(bindings, { + dedupeKey: "scheduled:ambiguous-send", + kind: "scheduled_maintenance", + payload: { scheduledTime: nowMsForTest() }, + }), + ).rejects.toThrow("Queue response timed out."); + + const retained = retainedMessages[0]; + if (retained === undefined) { + throw new Error("Expected Queue to retain the command message."); + } + + const row = await database + .app() + .select({ + lastErrorCode: apiCommandsTable.lastErrorCode, + status: apiCommandsTable.status, + }) + .from(apiCommandsTable) + .get(); + + expect(row).toEqual({ + lastErrorCode: API_COMMAND_QUEUE_SEND_FAILED_CODE, + status: "queued", + }); + await expect( + claimApiCommand({ + commandId: retained.commandId, + database, + nowMs: nowMsForTest(), + ownerId: "consumer-after-timeout", + }), + ).resolves.toMatchObject({ commandId: retained.commandId }); + }); + + test("redrives a durable command after a definite Queue send failure", async () => { + const database = await createPublicHttpContractDatabase(); + const sent: ApiCommandMessage[] = []; + let attempts = 0; + const queue = { + sent, + async send(body: ApiCommandMessage): Promise { + attempts += 1; + if (attempts === 1) { + throw new Error("Queue is unavailable."); + } + sent.push(body); + }, + }; + const bindings = createPublicHttpTestBindings(database, { + apiCommandQueue: queue, + }) as ApiBindings; + + await expect( + enqueueApiCommand(bindings, { + dedupeKey: "scheduled:redrive", + kind: "scheduled_maintenance", + payload: { scheduledTime: nowMsForTest() }, + }), + ).rejects.toThrow("Queue is unavailable."); + + await redriveFailedApiCommandEnqueues(bindings); + + expect(sent).toHaveLength(1); + const row = await database + .app() + .select({ + lastErrorCode: apiCommandsTable.lastErrorCode, + lastErrorMessage: apiCommandsTable.lastErrorMessage, + status: apiCommandsTable.status, + }) + .from(apiCommandsTable) + .get(); + + expect(row).toEqual({ + lastErrorCode: null, + lastErrorMessage: null, + status: "queued", + }); + }); + test("renews a running command claim for the current owner", async () => { const database = await createPublicHttpContractDatabase(); const queue = createApiCommandQueueStub(); From b7030f6b08f7d71a454417cf04b92f59018e5634 Mon Sep 17 00:00:00 2001 From: AsperforMias Date: Tue, 14 Jul 2026 02:48:40 +0800 Subject: [PATCH 2/3] fix(api): accept durable commands before queue delivery --- .../application/api-command-ledger.ts | 64 ++++++++++++----- apps/api/tests/api-command-queue.test.ts | 71 +++++++++++++++---- apps/api/tests/app-deployment-service.test.ts | 45 ++++++++++++ 3 files changed, 148 insertions(+), 32 deletions(-) diff --git a/apps/api/src/modules/api-command/application/api-command-ledger.ts b/apps/api/src/modules/api-command/application/api-command-ledger.ts index cc932a8a..e241a470 100644 --- a/apps/api/src/modules/api-command/application/api-command-ledger.ts +++ b/apps/api/src/modules/api-command/application/api-command-ledger.ts @@ -1,7 +1,7 @@ import { apiCommandsTable } from "@mosoo/db"; import type { ApiCommandId, ApiCommandKind, ApiCommandRow } from "@mosoo/db"; import { createPlatformId } from "@mosoo/id"; -import { and, asc, eq, lt, or, sql } from "drizzle-orm"; +import { and, asc, eq, inArray, lt, or, sql } from "drizzle-orm"; import { createErrorLogContext, logError } from "../../../platform/cloudflare/logger"; import type { ApiBindings } from "../../../platform/cloudflare/worker-types"; @@ -15,6 +15,10 @@ export const API_COMMAND_LEASE_RENEWAL_INTERVAL_MS = API_COMMAND_LEASE_MS / 5; export const API_COMMAND_QUEUE_SEND_FAILED_CODE = "queue_send_failed"; +export const API_COMMAND_QUEUE_DELIVERY_PENDING_CODE = "queue_delivery_pending"; + +const API_COMMAND_QUEUE_DELIVERY_PENDING_MESSAGE = "API command is awaiting queue delivery."; + const API_COMMAND_QUEUE_SEND_FAILED_MESSAGE = "API command queue send failed."; const API_COMMAND_QUEUE_REDRIVE_LIMIT = 100; @@ -95,7 +99,10 @@ async function clearApiCommandQueueSendFailure(input: { and( eq(apiCommandsTable.id, input.commandId), eq(apiCommandsTable.status, "queued"), - eq(apiCommandsTable.lastErrorCode, API_COMMAND_QUEUE_SEND_FAILED_CODE), + inArray(apiCommandsTable.lastErrorCode, [ + API_COMMAND_QUEUE_DELIVERY_PENDING_CODE, + API_COMMAND_QUEUE_SEND_FAILED_CODE, + ]), ), ) .run(); @@ -109,12 +116,33 @@ async function sendApiCommandMessage( await bindings.API_COMMAND_QUEUE.send(toQueueMessage(commandId)); } catch (error) { // A rejected producer response does not prove that Queue discarded the message. - // Keep the outbox row so a delivered duplicate can still be claimed safely. - await markApiCommandQueueSendFailed({ commandId, database: bindings.DB }); - throw error; + // The durable outbox record remains eligible for scheduled redrive either way. + try { + await markApiCommandQueueSendFailed({ commandId, database: bindings.DB }); + } catch (markError) { + logError("api-command.enqueue_failure_mark_failed", { + ...createErrorLogContext(markError), + commandId, + }); + } + + logError("api-command.enqueue_deferred", { + ...createErrorLogContext(error), + commandId, + }); + return; } - await clearApiCommandQueueSendFailure({ commandId, database: bindings.DB }); + try { + await clearApiCommandQueueSendFailure({ commandId, database: bindings.DB }); + } catch (error) { + // Queue accepted the command. Leaving its delivery marker intact is safe: + // a later redrive may send a duplicate, and consumer claiming is idempotent. + logError("api-command.enqueue_success_clear_failed", { + ...createErrorLogContext(error), + commandId, + }); + } } export async function redriveFailedApiCommandEnqueues( @@ -126,7 +154,10 @@ export async function redriveFailedApiCommandEnqueues( .where( and( eq(apiCommandsTable.status, "queued"), - eq(apiCommandsTable.lastErrorCode, API_COMMAND_QUEUE_SEND_FAILED_CODE), + inArray(apiCommandsTable.lastErrorCode, [ + API_COMMAND_QUEUE_DELIVERY_PENDING_CODE, + API_COMMAND_QUEUE_SEND_FAILED_CODE, + ]), ), ) .orderBy(asc(apiCommandsTable.id)) @@ -134,14 +165,7 @@ export async function redriveFailedApiCommandEnqueues( .all(); for (const command of commands) { - try { - await sendApiCommandMessage(bindings, command.id); - } catch (error) { - logError("api-command.enqueue_redrive_failed", { - ...createErrorLogContext(error), - commandId: command.id, - }); - } + await sendApiCommandMessage(bindings, command.id); } } @@ -164,8 +188,8 @@ export async function enqueueApiCommand( dedupeKey, id: commandId, kind: input.kind, - lastErrorCode: null, - lastErrorMessage: null, + lastErrorCode: API_COMMAND_QUEUE_DELIVERY_PENDING_CODE, + lastErrorMessage: API_COMMAND_QUEUE_DELIVERY_PENDING_MESSAGE, payloadJson: JSON.stringify(input.payload), status: "queued", updatedAt: nowMs, @@ -185,7 +209,11 @@ export async function enqueueApiCommand( throw new Error("API command enqueue could not confirm the ledger row."); } - if (current.status === "queued" && current.lastErrorCode === API_COMMAND_QUEUE_SEND_FAILED_CODE) { + if ( + current.status === "queued" && + (current.lastErrorCode === API_COMMAND_QUEUE_DELIVERY_PENDING_CODE || + current.lastErrorCode === API_COMMAND_QUEUE_SEND_FAILED_CODE) + ) { await sendApiCommandMessage(bindings, current.id); } diff --git a/apps/api/tests/api-command-queue.test.ts b/apps/api/tests/api-command-queue.test.ts index f456012d..ad25884e 100644 --- a/apps/api/tests/api-command-queue.test.ts +++ b/apps/api/tests/api-command-queue.test.ts @@ -5,6 +5,7 @@ import { eq } from "drizzle-orm"; import { API_COMMAND_LEASE_MS, + API_COMMAND_QUEUE_DELIVERY_PENDING_CODE, API_COMMAND_QUEUE_SEND_FAILED_CODE, claimApiCommand, enqueueApiCommand, @@ -112,13 +113,11 @@ describe("API command queue", () => { apiCommandQueue: queue, }) as ApiBindings; - await expect( - enqueueApiCommand(bindings, { - dedupeKey: "scheduled:ambiguous-send", - kind: "scheduled_maintenance", - payload: { scheduledTime: nowMsForTest() }, - }), - ).rejects.toThrow("Queue response timed out."); + const commandId = await enqueueApiCommand(bindings, { + dedupeKey: "scheduled:ambiguous-send", + kind: "scheduled_maintenance", + payload: { scheduledTime: nowMsForTest() }, + }); const retained = retainedMessages[0]; if (retained === undefined) { @@ -138,6 +137,7 @@ describe("API command queue", () => { lastErrorCode: API_COMMAND_QUEUE_SEND_FAILED_CODE, status: "queued", }); + expect(commandId).toBe(retained.commandId); await expect( claimApiCommand({ commandId: retained.commandId, @@ -166,13 +166,11 @@ describe("API command queue", () => { apiCommandQueue: queue, }) as ApiBindings; - await expect( - enqueueApiCommand(bindings, { - dedupeKey: "scheduled:redrive", - kind: "scheduled_maintenance", - payload: { scheduledTime: nowMsForTest() }, - }), - ).rejects.toThrow("Queue is unavailable."); + await enqueueApiCommand(bindings, { + dedupeKey: "scheduled:redrive", + kind: "scheduled_maintenance", + payload: { scheduledTime: nowMsForTest() }, + }); await redriveFailedApiCommandEnqueues(bindings); @@ -194,6 +192,51 @@ describe("API command queue", () => { }); }); + test("redrives a command left pending before its first Queue send", async () => { + const database = await createPublicHttpContractDatabase(); + const queue = createApiCommandQueueStub(); + const bindings = createPublicHttpTestBindings(database, { + apiCommandQueue: queue, + }) as ApiBindings; + const commandId = "01J0000000000000000000000C"; + + await database + .app() + .insert(apiCommandsTable) + .values({ + attemptCount: 0, + claimExpiresAt: null, + claimOwner: null, + completedAt: null, + createdAt: nowMsForTest(), + dedupeKey: "scheduled:pending-before-send", + id: commandId, + kind: "scheduled_maintenance", + lastErrorCode: API_COMMAND_QUEUE_DELIVERY_PENDING_CODE, + lastErrorMessage: "API command is awaiting queue delivery.", + payloadJson: JSON.stringify({ scheduledTime: nowMsForTest() }), + status: "queued", + updatedAt: nowMsForTest(), + }) + .run(); + + await redriveFailedApiCommandEnqueues(bindings); + + const row = await database + .app() + .select({ + lastErrorCode: apiCommandsTable.lastErrorCode, + lastErrorMessage: apiCommandsTable.lastErrorMessage, + }) + .from(apiCommandsTable) + .where(eq(apiCommandsTable.id, commandId)) + .get(); + + expect(queue.sent).toHaveLength(1); + expect(queue.sent[0]?.body).toEqual({ commandId }); + expect(row).toEqual({ lastErrorCode: null, lastErrorMessage: null }); + }); + test("renews a running command claim for the current owner", async () => { const database = await createPublicHttpContractDatabase(); const queue = createApiCommandQueueStub(); diff --git a/apps/api/tests/app-deployment-service.test.ts b/apps/api/tests/app-deployment-service.test.ts index 458a9005..fa78d882 100644 --- a/apps/api/tests/app-deployment-service.test.ts +++ b/apps/api/tests/app-deployment-service.test.ts @@ -4,6 +4,7 @@ import { apiCommandsTable, appDeploymentRunsTable, appDeploymentsTable } from "@ import { eq } from "drizzle-orm"; import { createAppDeploymentRunDispatchDedupeKey } from "../src/modules/api-command/application/api-command-enqueue"; +import { API_COMMAND_QUEUE_SEND_FAILED_CODE } from "../src/modules/api-command/application/api-command-ledger"; import { APP_DEPLOYMENT_RUN_DISPATCH_MAX_ATTEMPTS, APP_DEPLOYMENT_RUN_DISPATCH_RETRY_EXHAUSTED_CODE, @@ -545,6 +546,50 @@ describe("app deployment service", () => { expect(deploymentAfterPointerDrift?.latestRun?.id).toBe(run.id); }); + test("keeps a deployment run queued when Queue delivery is deferred", async () => { + const database = createDatabase(); + const { bindings } = createBindings(database); + const deferredBindings = { + ...bindings, + API_COMMAND_QUEUE: { + async send(): Promise { + throw new Error("Queue response timed out."); + }, + }, + }; + + const run = await deployApp( + deferredBindings, + VIEWER, + { appId: APP_ID, repoUrl: "https://github.com/samzong/awire" }, + { fetch: githubFetch, nowMs: () => NOW_MS }, + ); + + const command = await database + .app() + .select({ + lastErrorCode: apiCommandsTable.lastErrorCode, + status: apiCommandsTable.status, + }) + .from(apiCommandsTable) + .get(); + const runRow = await database + .app() + .select({ + errorCode: appDeploymentRunsTable.errorCode, + status: appDeploymentRunsTable.status, + }) + .from(appDeploymentRunsTable) + .where(eq(appDeploymentRunsTable.id, run.id)) + .get(); + + expect(command).toEqual({ + lastErrorCode: API_COMMAND_QUEUE_SEND_FAILED_CODE, + status: "queued", + }); + expect(runRow).toEqual({ errorCode: null, status: "queued" }); + }); + test("rejects a second deploy while a run is active", async () => { const database = createDatabase(); const { bindings } = createBindings(database); From 26e90bb076e0e6376ebb1f3fbee8bf89d4370025 Mon Sep 17 00:00:00 2001 From: AsperforMias Date: Tue, 14 Jul 2026 02:56:21 +0800 Subject: [PATCH 3/3] test(api): cover deployment dispatch queue recovery --- apps/api/tests/app-deployment-service.test.ts | 59 ++++++++++++++++++- 1 file changed, 56 insertions(+), 3 deletions(-) diff --git a/apps/api/tests/app-deployment-service.test.ts b/apps/api/tests/app-deployment-service.test.ts index fa78d882..2ce8f4ef 100644 --- a/apps/api/tests/app-deployment-service.test.ts +++ b/apps/api/tests/app-deployment-service.test.ts @@ -22,6 +22,7 @@ import { } from "../src/modules/apps/application/app-deployment.service"; import type { AuthenticatedViewer } from "../src/modules/auth/application/viewer-auth.service"; import type { SandboxHandle } from "../src/modules/runtime/infrastructure/sandbox-handles"; +import { createApiWorker } from "../src/platform/cloudflare/create-api-worker"; import type { ApiBindings } from "../src/platform/cloudflare/worker-types"; import { currentTimestampMs } from "../src/time"; import { @@ -546,14 +547,20 @@ describe("app deployment service", () => { expect(deploymentAfterPointerDrift?.latestRun?.id).toBe(run.id); }); - test("keeps a deployment run queued when Queue delivery is deferred", async () => { + test("redrives a dropped deployment dispatch without leaving its run queued forever", async () => { const database = createDatabase(); const { bindings } = createBindings(database); + const deliveredCommandIds: string[] = []; + let queueUnavailable = true; const deferredBindings = { ...bindings, API_COMMAND_QUEUE: { - async send(): Promise { - throw new Error("Queue response timed out."); + async send(input: { commandId: string }): Promise { + if (queueUnavailable) { + throw new Error("Queue response timed out."); + } + + deliveredCommandIds.push(input.commandId); }, }, }; @@ -588,6 +595,52 @@ describe("app deployment service", () => { status: "queued", }); expect(runRow).toEqual({ errorCode: null, status: "queued" }); + + queueUnavailable = false; + await createApiWorker().scheduled( + { scheduledTime: NOW_MS } as ScheduledController, + deferredBindings as ApiBindings, + ); + + const redrivenCommand = await database + .app() + .select({ + id: apiCommandsTable.id, + lastErrorCode: apiCommandsTable.lastErrorCode, + status: apiCommandsTable.status, + }) + .from(apiCommandsTable) + .where(eq(apiCommandsTable.dedupeKey, createAppDeploymentRunDispatchDedupeKey(run.id))) + .get(); + const runner: AppDeploymentBuildRunner = { + async build() {}, + async deploy() { + return { + externalDeploymentId: "pages-deployment-after-redrive", + externalProjectId: "pages-project-after-redrive", + externalVersionId: null, + url: `https://app-${APP_ID.toLowerCase()}.apps.localhost`, + }; + }, + async prepare() { + return { + repoDir: "/repo", + snapshot: { files: { "index.html": "
Recovered
" } }, + }; + }, + }; + + expect(deliveredCommandIds).toContain(redrivenCommand?.id); + expect(redrivenCommand).toMatchObject({ lastErrorCode: null, status: "queued" }); + + await dispatchAppDeploymentRun( + deferredBindings as ApiBindings, + { appDeploymentRunId: run.id }, + { runner }, + ); + + const status = await getAppDeploymentStatus(deferredBindings, VIEWER, APP_ID); + expect(status).toMatchObject({ status: "success" }); }); test("rejects a second deploy while a run is active", async () => {