From 2517b935da02888173c6ffee0cc988517dcc0620 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Thu, 22 Oct 2026 16:44:12 -0500 Subject: [PATCH 1/2] test(server): oversized coalesced windows stay within the suppression limit Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.webhook.test.ts | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 65cecbcf43286..f319fd05ed5fe 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1313,6 +1313,40 @@ it.effect.each([119_000, 120_000])( }, ); +it.effect("oversized coalesced windows stay within the suppression bucket's limit", () => { + const workers = new Map>(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + prompt: `{{body.absent}}${" ".repeat(10_000)}{{body.text}}`, + schedule: { + type: "webhook", + dedupe: { coalesceKey: "{{body.group}}", coalesceWindowSeconds: 10 }, + }, + }), + ); + const bodyText = JSON.stringify({ group: "B", text: "x".repeat(120_000) }); + for (let i = 0; i < 21; i++) { + const accepted = yield* service.triggerWebhook( + requestFor(task, { bodyText, body: new TextEncoder().encode(bodyText) }), + ); + assert.equal(accepted._tag === "accepted" && accepted.outcome, "accepted"); + yield* TestClock.adjust("10 seconds"); + yield* drainPendingDispatch(workers); + } + assert.equal(yield* Queue.size(launches), 0); + const counts = yield* sql<{ + outcome: string; + count: number; + }>`SELECT outcome, COUNT(*) AS count FROM scheduled_task_webhook_deliveries GROUP BY outcome`; + assert.deepEqual(counts, [{ outcome: "dispatch_failed", count: 20 }]); + }), + { workers }, + ); +}); + it.effect.each([1, 11])( "keeps relay claims for %i coalesced updates after their window closes", (count) => { From 92734be78ccc2c51b435edbb1f6163d9a8425de0 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Thu, 22 Oct 2026 16:47:12 -0500 Subject: [PATCH 2/2] fix(server): bound oversized coalesced webhook failure history A coalesced window whose rendered prompt is too long moves its logged delivery into the suppression bucket when it fails, but only insertion pruned a bucket, so these rows were never trimmed. Prune the row's bucket after a failure update too. Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.ts | 74 +++++++++++-------- 1 file changed, 45 insertions(+), 29 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 374d6b38027b8..d3957166cff2d 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1412,6 +1412,35 @@ export const layer = Layer.effect( WHERE delivery_id = ${deliveryId}`.pipe(Effect.asVoid, Effect.mapError(dedupeError)); }; + // Each suppression bucket retains its own rows, never real run history. + const deliveryBucket = (outcome: string, error: string | undefined) => + outcome === "filtered" + ? "filtered" + : outcome === "duplicate" || + outcome === "coalesced" || + outcome === "rate_limited" || + error === PROMPT_TOO_LONG + ? "dedupe" + : "main"; + const pruneDeliveries = ( + taskId: string, + retentionBucket: ReturnType, + ) => { + const rowBucket = sql.literal(`CASE WHEN outcome = 'filtered' THEN 'filtered' + WHEN outcome IN ('duplicate', 'coalesced', 'rate_limited') OR error = 'The filled-in prompt is too long.' THEN 'dedupe' + ELSE 'main' END`); + return sql` + DELETE FROM scheduled_task_webhook_deliveries + WHERE task_id = ${taskId} AND ${rowBucket} = ${retentionBucket} + AND delivery_id NOT IN ( + SELECT delivery_id FROM scheduled_task_webhook_deliveries + WHERE task_id = ${taskId} AND ${rowBucket} = ${retentionBucket} + ORDER BY received_at DESC, rowid DESC + LIMIT ${retentionBucket === "main" ? WEBHOOK_DELIVERY_RETENTION : retentionBucket === "filtered" ? WEBHOOK_FILTERED_DELIVERY_RETENTION : WEBHOOK_DEDUPE_DELIVERY_RETENTION} + ) + `; + }; + const recordDelivery = (input: { readonly id: string; readonly taskId: ScheduledTaskId; @@ -1454,30 +1483,7 @@ export const layer = Layer.effect( rendered_prompt = excluded.rendered_prompt, error = excluded.error `; - // Each suppression bucket retains its own rows, never real run history. - const bucket = (outcome: string, error: string | undefined) => - outcome === "filtered" - ? "filtered" - : outcome === "duplicate" || - outcome === "coalesced" || - outcome === "rate_limited" || - error === PROMPT_TOO_LONG - ? "dedupe" - : "main"; - const retentionBucket = bucket(input.outcome, input.error); - const rowBucket = sql.literal(`CASE WHEN outcome = 'filtered' THEN 'filtered' - WHEN outcome IN ('duplicate', 'coalesced', 'rate_limited') OR error = 'The filled-in prompt is too long.' THEN 'dedupe' - ELSE 'main' END`); - yield* sql` - DELETE FROM scheduled_task_webhook_deliveries - WHERE task_id = ${input.taskId} AND ${rowBucket} = ${retentionBucket} - AND delivery_id NOT IN ( - SELECT delivery_id FROM scheduled_task_webhook_deliveries - WHERE task_id = ${input.taskId} AND ${rowBucket} = ${retentionBucket} - ORDER BY received_at DESC, rowid DESC - LIMIT ${retentionBucket === "main" ? WEBHOOK_DELIVERY_RETENTION : retentionBucket === "filtered" ? WEBHOOK_FILTERED_DELIVERY_RETENTION : WEBHOOK_DEDUPE_DELIVERY_RETENTION} - ) - `; + yield* pruneDeliveries(input.taskId, deliveryBucket(input.outcome, input.error)); }), ) .pipe( @@ -1488,11 +1494,21 @@ export const layer = Layer.effect( }; const markDeliveryFailed = (deliveryId: string, message: string) => - sql` - UPDATE scheduled_task_webhook_deliveries - SET outcome = 'dispatch_failed', error = ${message} - WHERE delivery_id = ${deliveryId} - `.pipe(Effect.ignore); + sql + .withTransaction( + Effect.gen(function* () { + const [updated] = yield* sql<{ task_id: string }>` + UPDATE scheduled_task_webhook_deliveries + SET outcome = 'dispatch_failed', error = ${message} + WHERE delivery_id = ${deliveryId} + RETURNING task_id + `; + // An oversized prompt moves the row into the suppression bucket. + if (updated !== undefined) + yield* pruneDeliveries(updated.task_id, deliveryBucket("dispatch_failed", message)); + }), + ) + .pipe(Effect.ignore); /** Sliding one-minute window for deliveries that can start work. */ const takeRateSlot = (