fix(api): retain commands after ambiguous Queue sends

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, inArray, 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,16 @@ 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";
+
+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;
+
 export interface EnqueueApiCommandInput {
   dedupeKey: string;
   kind: ApiCommandKind;
@@ -43,27 +54,121 @@ function toQueueMessage(commandId: ApiCommandId): ApiCommandMessage {
 async function readApiCommandByDedupeKey(
   database: D1Database,
   dedupeKey: string,
-): Promise<Pick<ApiCommandRow, "id"> | null> {
+): Promise<Pick<ApiCommandRow, "id" | "lastErrorCode" | "status"> | 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)
       .get()) ?? null
   );
 }
 
-async function deleteApiCommand(input: {
+async function markApiCommandQueueSendFailed(input: {
   commandId: ApiCommandId;
   database: D1Database;
 }): Promise<void> {
   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<void> {
+  await getAppDatabase(input.database)
+    .update(apiCommandsTable)
+    .set({
+      lastErrorCode: null,
+      lastErrorMessage: null,
+      updatedAt: currentTimestampMs(),
+    })
+    .where(
+      and(
+        eq(apiCommandsTable.id, input.commandId),
+        eq(apiCommandsTable.status, "queued"),
+        inArray(apiCommandsTable.lastErrorCode, [
+          API_COMMAND_QUEUE_DELIVERY_PENDING_CODE,
+          API_COMMAND_QUEUE_SEND_FAILED_CODE,
+        ]),
+      ),
+    )
+    .run();
+}
+
+async function sendApiCommandMessage(
+  bindings: Pick<ApiBindings, "API_COMMAND_QUEUE" | "DB">,
+  commandId: ApiCommandId,
+): Promise<void> {
+  try {
+    await bindings.API_COMMAND_QUEUE.send(toQueueMessage(commandId));
+  } catch (error) {
+    // A rejected producer response does not prove that Queue discarded the message.
+    // 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;
+  }
+
+  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(
+  bindings: Pick<ApiBindings, "API_COMMAND_QUEUE" | "DB">,
+): Promise<void> {
+  const commands = await getAppDatabase(bindings.DB)
+    .select({ id: apiCommandsTable.id })
+    .from(apiCommandsTable)
+    .where(
+      and(
+        eq(apiCommandsTable.status, "queued"),
+        inArray(apiCommandsTable.lastErrorCode, [
+          API_COMMAND_QUEUE_DELIVERY_PENDING_CODE,
+          API_COMMAND_QUEUE_SEND_FAILED_CODE,
+        ]),
+      ),
+    )
+    .orderBy(asc(apiCommandsTable.id))
+    .limit(API_COMMAND_QUEUE_REDRIVE_LIMIT)
+    .all();
+
+  for (const command of commands) {
+    await sendApiCommandMessage(bindings, command.id);
+  }
+}
+
 export async function enqueueApiCommand(
   bindings: Pick<ApiBindings, "API_COMMAND_QUEUE" | "DB">,
   input: EnqueueApiCommandInput,
@@ -83,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,
@@ -93,12 +198,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 +209,14 @@ 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_DELIVERY_PENDING_CODE ||
+      current.lastErrorCode === API_COMMAND_QUEUE_SEND_FAILED_CODE)
+  ) {
+    await sendApiCommandMessage(bindings, current.id);
+  }
+
   return current.id;
 }
 

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<ApiBindings> {
       return response;
     },
     async scheduled(controller: ScheduledController, env: ApiBindings): Promise<void> {
+      await redriveFailedApiCommandEnqueues(env);
       await enqueueScheduledMaintenanceCommand(env, {
         scheduledTime: controller.scheduledTime,
       });

apps/api/tests/api-command-queue.test.ts

@@ -5,8 +5,11 @@ import { eq } from "drizzle-orm";
 
 import {
   API_COMMAND_LEASE_MS,
+  API_COMMAND_QUEUE_DELIVERY_PENDING_CODE,
+  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 +99,144 @@ 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<void> {
+        retainedMessages.push(body);
+        throw new Error("Queue response timed out.");
+      },
+    };
+    const bindings = createPublicHttpTestBindings(database, {
+      apiCommandQueue: queue,
+    }) as ApiBindings;
+
+    const commandId = await enqueueApiCommand(bindings, {
+      dedupeKey: "scheduled:ambiguous-send",
+      kind: "scheduled_maintenance",
+      payload: { scheduledTime: nowMsForTest() },
+    });
+
+    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",
+    });
+    expect(commandId).toBe(retained.commandId);
+    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<void> {
+        attempts += 1;
+        if (attempts === 1) {
+          throw new Error("Queue is unavailable.");
+        }
+        sent.push(body);
+      },
+    };
+    const bindings = createPublicHttpTestBindings(database, {
+      apiCommandQueue: queue,
+    }) as ApiBindings;
+
+    await enqueueApiCommand(bindings, {
+      dedupeKey: "scheduled:redrive",
+      kind: "scheduled_maintenance",
+      payload: { scheduledTime: nowMsForTest() },
+    });
+
+    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("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();