fix(channels): retain final-delivery jobs after Queue send failure

apps/api/src/modules/channels/application/channel-final-delivery-jobs.ts

@@ -2,8 +2,9 @@ import { channelFinalDeliveryJobsTable } from "@mosoo/db";
 import type { AgentChannelBindingProvider, ChannelFinalDeliveryJobId } from "@mosoo/db";
 import { createPlatformId } from "@mosoo/id";
 import type { ChannelBindingId, SessionId, SessionRunId } from "@mosoo/id";
-import { and, eq } from "drizzle-orm";
+import { and, asc, eq, inArray } 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";
@@ -57,6 +58,14 @@ export interface JobLedgerRow {
 
 const DELIVERY_CLAIM_PREFIX = "delivery_claim:";
 
+export const CHANNEL_FINAL_DELIVERY_QUEUE_DELIVERY_PENDING_CODE =
+  "channel_final_delivery_queue_delivery_pending";
+
+export const CHANNEL_FINAL_DELIVERY_QUEUE_SEND_FAILED_CODE =
+  "channel_final_delivery_queue_send_failed";
+
+const CHANNEL_FINAL_DELIVERY_QUEUE_REDRIVE_LIMIT = 100;
+
 export interface ChannelFinalDeliveryClaim {
   attemptCount: number;
   claimCode: string;
@@ -222,16 +231,115 @@ export async function recordJobWait(input: {
     .run();
 }
 
-async function deleteJobDedupeRow(input: {
+async function markJobQueueSendFailed(input: {
   database: D1Database;
   jobId: ChannelFinalDeliveryJobId;
 }): Promise<void> {
   await getAppDatabase(input.database)
-    .delete(channelFinalDeliveryJobsTable)
-    .where(eq(channelFinalDeliveryJobsTable.id, input.jobId))
+    .update(channelFinalDeliveryJobsTable)
+    .set({
+      lastErrorCode: CHANNEL_FINAL_DELIVERY_QUEUE_SEND_FAILED_CODE,
+      updatedAt: currentTimestampMs(),
+    })
+    .where(
+      and(
+        eq(channelFinalDeliveryJobsTable.id, input.jobId),
+        eq(channelFinalDeliveryJobsTable.status, "dispatched"),
+      ),
+    )
+    .run();
+}
+
+async function clearJobQueueSendFailure(input: {
+  database: D1Database;
+  jobId: ChannelFinalDeliveryJobId;
+}): Promise<void> {
+  await getAppDatabase(input.database)
+    .update(channelFinalDeliveryJobsTable)
+    .set({
+      lastErrorCode: null,
+      updatedAt: currentTimestampMs(),
+    })
+    .where(
+      and(
+        eq(channelFinalDeliveryJobsTable.id, input.jobId),
+        eq(channelFinalDeliveryJobsTable.status, "dispatched"),
+        inArray(channelFinalDeliveryJobsTable.lastErrorCode, [
+          CHANNEL_FINAL_DELIVERY_QUEUE_DELIVERY_PENDING_CODE,
+          CHANNEL_FINAL_DELIVERY_QUEUE_SEND_FAILED_CODE,
+        ]),
+      ),
+    )
     .run();
 }
 
+function toChannelFinalDeliveryMessage(
+  jobId: ChannelFinalDeliveryJobId,
+): ChannelFinalDeliveryMessage {
+  return { jobId };
+}
+
+async function sendChannelFinalDeliveryMessage(
+  bindings: Pick<ApiBindings, "CHANNEL_FINAL_DELIVERY_QUEUE" | "DB">,
+  jobId: ChannelFinalDeliveryJobId,
+): Promise<void> {
+  try {
+    await bindings.CHANNEL_FINAL_DELIVERY_QUEUE.send(toChannelFinalDeliveryMessage(jobId));
+  } catch (error) {
+    // A rejected producer response does not prove that Queue discarded the message.
+    // The durable ledger remains eligible for scheduled redrive either way.
+    try {
+      await markJobQueueSendFailed({ database: bindings.DB, jobId });
+    } catch (markError) {
+      logError("channel-final-delivery.enqueue_failure_mark_failed", {
+        ...createErrorLogContext(markError),
+        jobId,
+      });
+    }
+
+    logError("channel-final-delivery.enqueue_deferred", {
+      ...createErrorLogContext(error),
+      jobId,
+    });
+    return;
+  }
+
+  try {
+    await clearJobQueueSendFailure({ database: bindings.DB, jobId });
+  } catch (error) {
+    // Queue accepted the job. A later redrive can safely send a duplicate because
+    // the consumer claim is idempotent.
+    logError("channel-final-delivery.enqueue_success_clear_failed", {
+      ...createErrorLogContext(error),
+      jobId,
+    });
+  }
+}
+
+export async function redriveFailedChannelFinalDeliveryEnqueues(
+  bindings: Pick<ApiBindings, "CHANNEL_FINAL_DELIVERY_QUEUE" | "DB">,
+): Promise<void> {
+  const jobs = await getAppDatabase(bindings.DB)
+    .select({ id: channelFinalDeliveryJobsTable.id })
+    .from(channelFinalDeliveryJobsTable)
+    .where(
+      and(
+        eq(channelFinalDeliveryJobsTable.status, "dispatched"),
+        inArray(channelFinalDeliveryJobsTable.lastErrorCode, [
+          CHANNEL_FINAL_DELIVERY_QUEUE_DELIVERY_PENDING_CODE,
+          CHANNEL_FINAL_DELIVERY_QUEUE_SEND_FAILED_CODE,
+        ]),
+      ),
+    )
+    .orderBy(asc(channelFinalDeliveryJobsTable.id))
+    .limit(CHANNEL_FINAL_DELIVERY_QUEUE_REDRIVE_LIMIT)
+    .all();
+
+  for (const job of jobs) {
+    await sendChannelFinalDeliveryMessage(bindings, job.id);
+  }
+}
+
 export async function enqueueChannelFinalDeliveryJob(
   bindings: ApiBindings,
   input: EnqueueChannelFinalDeliveryJobInput,
@@ -247,7 +355,7 @@ export async function enqueueChannelFinalDeliveryJob(
       createdAt: nowMs,
       externalEventId: input.externalEventId,
       id: jobId,
-      lastErrorCode: null,
+      lastErrorCode: CHANNEL_FINAL_DELIVERY_QUEUE_DELIVERY_PENDING_CODE,
       payloadJson: JSON.stringify(payload),
       provider: input.provider,
       runId: input.runId,
@@ -268,13 +376,7 @@ export async function enqueueChannelFinalDeliveryJob(
     return null;
   }
 
-  try {
-    const message: ChannelFinalDeliveryMessage = { jobId };
-    await bindings.CHANNEL_FINAL_DELIVERY_QUEUE.send(message);
-  } catch (error) {
-    await deleteJobDedupeRow({ database: bindings.DB, jobId });
-    throw error;
-  }
+  await sendChannelFinalDeliveryMessage(bindings, jobId);
 
   return jobId;
 }

apps/api/src/modules/channels/application/channel-final-delivery.service.ts

@@ -39,6 +39,7 @@ export type { ChannelFinalDeliveryMessage } from "./channel-final-delivery-messa
 export {
   createChannelFinalDeliveryScheduler,
   enqueueChannelFinalDeliveryJob,
+  redriveFailedChannelFinalDeliveryEnqueues,
 } from "./channel-final-delivery-jobs";
 export type {
   ChannelFinalDeliveryScheduler,

apps/api/src/platform/cloudflare/create-api-worker.ts

@@ -7,6 +7,7 @@ import {
   processApiCommandDeadLetterMessage,
   processApiCommandMessage,
 } from "../../modules/api-command/application/api-command-processor";
+import { redriveFailedChannelFinalDeliveryEnqueues } from "../../modules/channels/application/channel-final-delivery-jobs";
 import type { ChannelFinalDeliveryMessage } from "../../modules/channels/application/channel-final-delivery-message";
 import type { ApiBindings } from "./worker-types";
 
@@ -44,6 +45,7 @@ export function createApiWorker(): ExportedHandler<ApiBindings> {
     },
     async scheduled(controller: ScheduledController, env: ApiBindings): Promise<void> {
       await redriveFailedApiCommandEnqueues(env);
+      await redriveFailedChannelFinalDeliveryEnqueues(env);
       await enqueueScheduledMaintenanceCommand(env, {
         scheduledTime: controller.scheduledTime,
       });

apps/api/tests/app-deployment-service.test.ts

@@ -111,6 +111,13 @@ function createDatabase(): SqliteD1Database {
       ON app_deployment_run (app_id)
       WHERE status IN ('queued', 'preparing', 'building', 'submitting', 'submitted', 'activating');
 
+    -- The Worker scheduled handler also redrives final-delivery queue enqueues.
+    CREATE TABLE channel_final_delivery_job (
+      id text PRIMARY KEY NOT NULL,
+      last_error_code text,
+      status text NOT NULL
+    );
+
     CREATE TABLE api_command (
       attempt_count integer DEFAULT 0 NOT NULL,
       claim_expires_at integer,

apps/api/tests/channel-final-delivery-scheduling.test.ts

@@ -12,7 +12,9 @@ import type { ChannelFinalDeliveryMessage } from "../src/modules/channels/applic
 import {
   enqueueChannelFinalDeliveryJob,
   processChannelFinalDeliveryMessage,
+  redriveFailedChannelFinalDeliveryEnqueues,
 } from "../src/modules/channels/application/channel-final-delivery.service";
+import { createApiWorker } from "../src/platform/cloudflare/create-api-worker";
 import type { ApiBindings } from "../src/platform/cloudflare/worker-types";
 import {
   createTestEnvironment,
@@ -666,6 +668,99 @@ describe("channel final delivery scheduling", () => {
     }
   });
 
+  test("retains an ambiguously accepted enqueue and delivers the retained job", async () => {
+    const telegramBodies: unknown[] = [];
+    const restoreFetch = installTelegramFetch(telegramBodies);
+
+    try {
+      const { bindings: baseBindings, database, queue } = await createTestEnvironment();
+      const bindings: ApiBindings = {
+        ...baseBindings,
+        CHANNEL_FINAL_DELIVERY_QUEUE: {
+          sent: queue.sent,
+          async send(body, options): Promise<void> {
+            await queue.send(body, options);
+            throw new Error("Queue response timed out after accepting the message.");
+          },
+        },
+      } as ApiBindings;
+      const seed = await createCompletedTelegramFinalDeliveryJob({
+        bindings,
+        database,
+        externalEventId: "telegram:update:ambiguous-enqueue",
+      });
+
+      const retained = await database
+        .app()
+        .select({
+          lastErrorCode: channelFinalDeliveryJobsTable.lastErrorCode,
+          status: channelFinalDeliveryJobsTable.status,
+        })
+        .from(channelFinalDeliveryJobsTable)
+        .where(eq(channelFinalDeliveryJobsTable.id, seed.jobId))
+        .get();
+      const recorded = createRecordedQueueMessage<ChannelFinalDeliveryMessage>({
+        body: takeQueuedMessageBody(queue, seed.jobId),
+      });
+      await processChannelFinalDeliveryMessage(bindings, recorded.message, {}, nowMsForTest);
+
+      expect(retained).toEqual({
+        lastErrorCode: "channel_final_delivery_queue_send_failed",
+        status: "dispatched",
+      });
+      expect(telegramBodies).toHaveLength(1);
+      expect(recorded.recorded).toEqual([{ type: "ack" }]);
+    } finally {
+      restoreFetch();
+    }
+  });
+
+  test("scheduled redrive delivers jobs when the initial queue send is rejected", async () => {
+    const telegramBodies: unknown[] = [];
+    const restoreFetch = installTelegramFetch(telegramBodies);
+
+    try {
+      const { bindings: baseBindings, database, queue } = await createTestEnvironment();
+      let queueAvailable = false;
+      const bindings: ApiBindings = {
+        ...baseBindings,
+        CHANNEL_FINAL_DELIVERY_QUEUE: {
+          sent: queue.sent,
+          async send(body, options): Promise<void> {
+            if (!queueAvailable) {
+              throw new Error("Queue unavailable.");
+            }
+
+            await queue.send(body, options);
+          },
+        },
+      } as ApiBindings;
+      const seed = await createCompletedTelegramFinalDeliveryJob({
+        bindings,
+        database,
+        externalEventId: "telegram:update:enqueue-redrive",
+      });
+
+      expect(queue.sent).toEqual([]);
+      queueAvailable = true;
+      await createApiWorker().scheduled(
+        { scheduledTime: nowMsForTest() } as ScheduledController,
+        bindings,
+      );
+
+      const queued = takeQueuedMessageBody(queue, seed.jobId);
+      const recorded = createRecordedQueueMessage<ChannelFinalDeliveryMessage>({ body: queued });
+      await processChannelFinalDeliveryMessage(bindings, recorded.message, {}, nowMsForTest);
+
+      expect(telegramBodies).toHaveLength(1);
+      expect(recorded.recorded).toEqual([{ type: "ack" }]);
+      await redriveFailedChannelFinalDeliveryEnqueues(bindings);
+      expect(queue.sent).toHaveLength(1);
+    } finally {
+      restoreFetch();
+    }
+  });
+
   test("consumer-side idempotency: processing a delivered message twice does not resend", async () => {
     const telegramBodies: unknown[] = [];
     const restoreFetch = installTelegramFetch(telegramBodies);