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..b8de2aae 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, 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 { + 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 { 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); + } +} + +async function clearRecoveredJobPendingMarker(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"), + eq( + channelFinalDeliveryJobsTable.lastErrorCode, + CHANNEL_FINAL_DELIVERY_RECOVERY_QUEUE_PENDING_CODE, + ), + ), + ) .run(); } +async function sendRecoveredJobMessage( + bindings: Pick, + jobId: ChannelFinalDeliveryJobId, +): Promise { + 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, +): Promise { + 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; } 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..dffe8e13 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 @@ -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, diff --git a/apps/api/src/platform/cloudflare/create-api-worker.ts b/apps/api/src/platform/cloudflare/create-api-worker.ts index 66caad6e..20b50b2f 100644 --- a/apps/api/src/platform/cloudflare/create-api-worker.ts +++ b/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 { }, async scheduled(controller: ScheduledController, env: ApiBindings): Promise { await redriveFailedApiCommandEnqueues(env); + await redriveFailedChannelFinalDeliveryEnqueues(env); + await redriveExhaustedChannelFinalDeliveryJobs(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..a33ad27c 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, @@ -525,16 +527,33 @@ describe("channel final delivery scheduling", () => { } }); - test("fails the durable job after the final delivery retry attempt", async () => { + test("recovers a final delivery job after the retry limit through the scheduled worker", async () => { const telegramBodies: unknown[] = []; + let providerAvailable = false; + let recoveryQueueAvailable = true; const restoreFetch = installTelegramFetch(telegramBodies, { async onSendMessage() { - throw new Error("provider unavailable"); + if (!providerAvailable) { + throw new Error("provider unavailable"); + } }, }); try { - const { bindings, database, queue } = await createTestEnvironment(); + const { bindings: baseBindings, database, queue } = await createTestEnvironment(); + const bindings: ApiBindings = { + ...baseBindings, + CHANNEL_FINAL_DELIVERY_QUEUE: { + sent: queue.sent, + async send(body, options): Promise { + if (!recoveryQueueAvailable) { + throw new Error("Queue unavailable during terminal recovery."); + } + + await queue.send(body, options); + }, + }, + } as ApiBindings; const seed = await createCompletedTelegramFinalDeliveryJob({ bindings, database, @@ -568,11 +587,61 @@ describe("channel final delivery scheduling", () => { expect(telegramBodies).toHaveLength(1); expect(job).toEqual({ attemptCount: 8, - lastErrorCode: "Error", + lastErrorCode: "delivery_retry_exhausted:Error", status: "failed", }); expect(recorded.recorded).toEqual([{ type: "ack" }]); expect(queue.sent).toHaveLength(1); + + providerAvailable = true; + recoveryQueueAvailable = false; + await createApiWorker().scheduled( + { scheduledTime: nowMsForTest() } as ScheduledController, + bindings, + ); + + const pendingRecovery = await database + .app() + .select({ + lastErrorCode: channelFinalDeliveryJobsTable.lastErrorCode, + status: channelFinalDeliveryJobsTable.status, + }) + .from(channelFinalDeliveryJobsTable) + .where(eq(channelFinalDeliveryJobsTable.id, seed.jobId)) + .get(); + + expect(pendingRecovery).toEqual({ + lastErrorCode: "channel_final_delivery_recovery_queue_pending", + status: "dispatched", + }); + recoveryQueueAvailable = true; + await createApiWorker().scheduled( + { scheduledTime: nowMsForTest() } as ScheduledController, + bindings, + ); + const redriven = createRecordedQueueMessage({ + body: takeQueuedMessageBody(queue, seed.jobId), + }); + await processChannelFinalDeliveryMessage(bindings, redriven.message, {}, nowMsForTest); + + const recovered = await database + .app() + .select({ + attemptCount: channelFinalDeliveryJobsTable.attemptCount, + lastErrorCode: channelFinalDeliveryJobsTable.lastErrorCode, + status: channelFinalDeliveryJobsTable.status, + }) + .from(channelFinalDeliveryJobsTable) + .where(eq(channelFinalDeliveryJobsTable.id, seed.jobId)) + .get(); + + expect(telegramBodies).toHaveLength(2); + expect(recovered).toEqual({ + attemptCount: 9, + lastErrorCode: null, + status: "delivered", + }); + expect(redriven.recorded).toEqual([{ type: "ack" }]); } finally { restoreFetch(); } @@ -666,6 +735,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);