diff --git a/apps/api/src/modules/channels/application/channel-final-delivery-jobs.ts b/apps/api/src/modules/channels/application/channel-final-delivery-jobs.ts index 97869d8e..d1fecf26 100644 --- a/apps/api/src/modules/channels/application/channel-final-delivery-jobs.ts +++ b/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 { 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 { + 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, + jobId: ChannelFinalDeliveryJobId, +): Promise { + 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, +): Promise { + 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; } diff --git a/apps/api/src/modules/channels/application/channel-final-delivery.service.ts b/apps/api/src/modules/channels/application/channel-final-delivery.service.ts index a2d51ff2..a41e01c1 100644 --- a/apps/api/src/modules/channels/application/channel-final-delivery.service.ts +++ b/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, diff --git a/apps/api/src/platform/cloudflare/create-api-worker.ts b/apps/api/src/platform/cloudflare/create-api-worker.ts index 66caad6e..be6c1c6a 100644 --- a/apps/api/src/platform/cloudflare/create-api-worker.ts +++ b/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 { }, async scheduled(controller: ScheduledController, env: ApiBindings): Promise { await redriveFailedApiCommandEnqueues(env); + await redriveFailedChannelFinalDeliveryEnqueues(env); await enqueueScheduledMaintenanceCommand(env, { scheduledTime: controller.scheduledTime, }); diff --git a/apps/api/tests/app-deployment-service.test.ts b/apps/api/tests/app-deployment-service.test.ts index 6f15a426..64810e29 100644 --- a/apps/api/tests/app-deployment-service.test.ts +++ b/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, diff --git a/apps/api/tests/channel-final-delivery-scheduling.test.ts b/apps/api/tests/channel-final-delivery-scheduling.test.ts index 148aaa4d..9b965bd5 100644 --- a/apps/api/tests/channel-final-delivery-scheduling.test.ts +++ b/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 { + 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({ + 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 { + 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({ 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);