fix(channels): recover exhausted final-delivery jobs

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, like, or } 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,21 @@ 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;
+
+const CHANNEL_FINAL_DELIVERY_RETRY_EXHAUSTED_PREFIX = "delivery_retry_exhausted:";
+
+const CHANNEL_FINAL_DELIVERY_RECOVERY_QUEUE_PENDING_CODE =
+  "channel_final_delivery_recovery_queue_pending";
+
+const CHANNEL_FINAL_DELIVERY_RECOVERY_REDRIVE_LIMIT = 100;
+
 export interface ChannelFinalDeliveryClaim {
   attemptCount: number;
   claimCode: string;
@@ -180,6 +196,30 @@ export async function markJobFailed(input: {
     .run();
 }
 
+export async function markJobDeliveryRetryExhausted(input: {
+  attemptCount: number;
+  database: D1Database;
+  errorCode: string;
+  jobId: ChannelFinalDeliveryJobId;
+  nowMs: number;
+}): Promise<void> {
+  await getAppDatabase(input.database)
+    .update(channelFinalDeliveryJobsTable)
+    .set({
+      attemptCount: input.attemptCount,
+      lastErrorCode: `${CHANNEL_FINAL_DELIVERY_RETRY_EXHAUSTED_PREFIX}${input.errorCode}`,
+      status: "failed",
+      updatedAt: input.nowMs,
+    })
+    .where(
+      and(
+        eq(channelFinalDeliveryJobsTable.id, input.jobId),
+        eq(channelFinalDeliveryJobsTable.status, "dispatched"),
+      ),
+    )
+    .run();
+}
+
 export async function recordJobAttempt(input: {
   attemptCount: number;
   database: D1Database;
@@ -222,16 +262,230 @@ 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);
+  }
+}
+
+async function clearRecoveredJobPendingMarker(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"),
+        eq(
+          channelFinalDeliveryJobsTable.lastErrorCode,
+          CHANNEL_FINAL_DELIVERY_RECOVERY_QUEUE_PENDING_CODE,
+        ),
+      ),
+    )
     .run();
 }
 
+async function sendRecoveredJobMessage(
+  bindings: Pick<ApiBindings, "CHANNEL_FINAL_DELIVERY_QUEUE" | "DB">,
+  jobId: ChannelFinalDeliveryJobId,
+): Promise<void> {
+  try {
+    await bindings.CHANNEL_FINAL_DELIVERY_QUEUE.send({ jobId });
+  } catch (error) {
+    // A rejected producer response does not prove Queue discarded the message.
+    // Keep the pending marker so both actual rejections and ambiguous accepts
+    // are retried safely by a later scheduled run.
+    logError("channel-final-delivery.recovery_enqueue_deferred", {
+      ...createErrorLogContext(error),
+      jobId,
+    });
+    return;
+  }
+
+  try {
+    await clearRecoveredJobPendingMarker({ database: bindings.DB, jobId });
+  } catch (error) {
+    // Queue accepted the job. Retaining the marker causes a safe duplicate on
+    // a later scheduled recovery if the consumer has not claimed it yet.
+    logError("channel-final-delivery.recovery_success_clear_failed", {
+      ...createErrorLogContext(error),
+      jobId,
+    });
+  }
+}
+
+export async function redriveExhaustedChannelFinalDeliveryJobs(
+  bindings: Pick<ApiBindings, "CHANNEL_FINAL_DELIVERY_QUEUE" | "DB">,
+): Promise<void> {
+  const jobs = await getAppDatabase(bindings.DB)
+    .select({
+      id: channelFinalDeliveryJobsTable.id,
+      lastErrorCode: channelFinalDeliveryJobsTable.lastErrorCode,
+      status: channelFinalDeliveryJobsTable.status,
+    })
+    .from(channelFinalDeliveryJobsTable)
+    .where(
+      or(
+        and(
+          eq(channelFinalDeliveryJobsTable.status, "failed"),
+          like(
+            channelFinalDeliveryJobsTable.lastErrorCode,
+            `${CHANNEL_FINAL_DELIVERY_RETRY_EXHAUSTED_PREFIX}%`,
+          ),
+        ),
+        and(
+          eq(channelFinalDeliveryJobsTable.status, "dispatched"),
+          eq(
+            channelFinalDeliveryJobsTable.lastErrorCode,
+            CHANNEL_FINAL_DELIVERY_RECOVERY_QUEUE_PENDING_CODE,
+          ),
+        ),
+      ),
+    )
+    .orderBy(asc(channelFinalDeliveryJobsTable.id))
+    .limit(CHANNEL_FINAL_DELIVERY_RECOVERY_REDRIVE_LIMIT)
+    .all();
+
+  for (const job of jobs) {
+    if (job.status === "failed") {
+      if (job.lastErrorCode === null) {
+        continue;
+      }
+
+      const transitioned = await getAppDatabase(bindings.DB)
+        .update(channelFinalDeliveryJobsTable)
+        .set({
+          lastErrorCode: CHANNEL_FINAL_DELIVERY_RECOVERY_QUEUE_PENDING_CODE,
+          status: "dispatched",
+          updatedAt: currentTimestampMs(),
+        })
+        .where(
+          and(
+            eq(channelFinalDeliveryJobsTable.id, job.id),
+            eq(channelFinalDeliveryJobsTable.status, "failed"),
+            eq(channelFinalDeliveryJobsTable.lastErrorCode, job.lastErrorCode),
+          ),
+        )
+        .run();
+
+      if (getD1ChangeCount(transitioned) === 0) {
+        continue;
+      }
+    }
+
+    await sendRecoveredJobMessage(bindings, job.id);
+  }
+}
+
 export async function enqueueChannelFinalDeliveryJob(
   bindings: ApiBindings,
   input: EnqueueChannelFinalDeliveryJobInput,
@@ -247,7 +501,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 +522,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

@@ -14,6 +14,7 @@ import {
 } from "./channel-final-delivery-errors";
 import {
   claimJobForDelivery,
+  markJobDeliveryRetryExhausted,
   markJobDelivered,
   markJobFailed,
   parseActiveChannelFinalDeliveryClaim,
@@ -39,6 +40,8 @@ export type { ChannelFinalDeliveryMessage } from "./channel-final-delivery-messa
 export {
   createChannelFinalDeliveryScheduler,
   enqueueChannelFinalDeliveryJob,
+  redriveFailedChannelFinalDeliveryEnqueues,
+  redriveExhaustedChannelFinalDeliveryJobs,
 } from "./channel-final-delivery-jobs";
 export type {
   ChannelFinalDeliveryScheduler,
@@ -305,7 +308,7 @@ export async function processChannelFinalDeliveryMessage(
     }
 
     if (attemptCount >= CHANNEL_FINAL_DELIVERY_MAX_DELIVERY_ATTEMPTS) {
-      await markJobFailed({
+      await markJobDeliveryRetryExhausted({
         attemptCount,
         database: bindings.DB,
         errorCode,

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

@@ -7,6 +7,10 @@ import {
   processApiCommandDeadLetterMessage,
   processApiCommandMessage,
 } from "../../modules/api-command/application/api-command-processor";
+import {
+  redriveExhaustedChannelFinalDeliveryJobs,
+  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 +48,8 @@ export function createApiWorker(): ExportedHandler<ApiBindings> {
     },
     async scheduled(controller: ScheduledController, env: ApiBindings): Promise<void> {
       await redriveFailedApiCommandEnqueues(env);
+      await redriveFailedChannelFinalDeliveryEnqueues(env);
+      await redriveExhaustedChannelFinalDeliveryJobs(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,