From fda2b279a09d130afc59e9d7445e0b8179b55afd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 11:06:12 +0000 Subject: [PATCH 1/6] test(server): reproduce lost webhook deliveries after restart (#18810) An accepted, non-coalesced delivery that waits behind a running delivery for the same task is only held in memory. This test closes the service while it waits, starts a fresh service over the same database, and expects the delivery to launch. It fails on main. Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.webhook.test.ts | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 65cecbcf43286..f69680adc79af 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -14,6 +14,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as EffectScheduler from "effect/Scheduler"; import * as Schema from "effect/Schema"; @@ -1115,6 +1116,42 @@ it.effect( }, ); +it.effect("an accepted delivery queued behind a run still launches after a restart", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const gate = yield* Deferred.make(); + const { task, deliveryId } = yield* withService( + ({ service, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput()); + yield* service.triggerWebhook(requestFor(task)); + yield* Queue.take(launches); + const queued = yield* service.triggerWebhook(requestFor(task)); + assert.equal(queued._tag === "accepted" && queued.outcome, "accepted"); + return { task, deliveryId: queued._tag === "accepted" ? queued.deliveryId : "" }; + }), + { gate, sql }, + ); + const commandId = `scheduled-task:${task.id}:webhook:${deliveryId}`; + const workers = new Map>(); + yield* withService( + ({ launches }) => + Effect.gen(function* () { + const launched = yield* Effect.gen(function* () { + while ((yield* Queue.take(launches)).commandId !== commandId); + }).pipe(Effect.timeoutOption("1 minute"), Effect.forkChild); + yield* workers.get("webhook-coalescing")!; + yield* TestClock.adjust("1 minute"); + assert.isTrue( + Option.isSome(yield* Fiber.join(launched)), + "the accepted delivery was never launched after the restart", + ); + }), + { sql, workers }, + ); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)), +); + it.effect("guards bad-signature floods separately from valid run admission", () => withService(({ service, sql }) => Effect.gen(function* () { From b06e2f2f3ad1801b1b9360aa805f35d148cf3671 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 11:21:55 +0000 Subject: [PATCH 2/6] fix(server): accepted webhook deliveries survive a server restart An immediate (non-coalesced) delivery that the server answered with `accepted` lived only in a forked fiber waiting on the task's permit. A restart before its run started lost it, and a relay replay was then treated as a duplicate. The immediate path now writes the delivery into scheduled_task_webhook_pending in the same transaction as its `accepted` log, due now and ready to drain, and starts the existing per-task drain at once instead of forking the run directly. The scheduler's drain picks up anything left after a restart and reruns it with the same command id. The in-memory queued counter is gone: the persisted rows already count toward the per-task queue cap. Releasing a failed delivery now drops its row in the same transaction, so a relay redelivery the release admits never collides with or is deleted alongside the old row. Immediate rows use an empty coalesce key and are not reported as open windows. Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.ts | 174 ++++++++---------- .../ScheduledTaskService.webhook.test.ts | 89 ++++++--- .../src/scheduledTasks/webhookDedupe.ts | 21 ++- 3 files changed, 151 insertions(+), 133 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 374d6b38027b8..19d260203f647 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -410,9 +410,6 @@ export const layer = Layer.effect( const webhookPermits = yield* Ref.make>( new Map(), ); - // Keyed by task id and creation time, so deliveries of a deleted task that - // finish late release their own count, never a recreated task's. - const webhookQueued = yield* Ref.make>(new Map()); const webhookRateWindows = yield* Ref.make>(new Map()); const webhookRejectedRateWindows = yield* Ref.make>( new Map(), @@ -1347,7 +1344,7 @@ export const layer = Layer.effect( ) => sql` SELECT d.*, EXISTS (SELECT 1 FROM scheduled_task_webhook_pending p - WHERE p.delivery_id = d.delivery_id AND p.state IN ('pending', 'dispatching')) AS pending_window + WHERE p.delivery_id = d.delivery_id AND p.coalesce_key <> '' AND p.state IN ('pending', 'dispatching')) AS pending_window FROM scheduled_task_webhook_deliveries d WHERE task_id = ${input.id} ORDER BY received_at DESC, rowid DESC `.pipe( @@ -1368,7 +1365,7 @@ export const layer = Layer.effect( )(function* (input) { const rows = yield* sql` SELECT d.*, EXISTS (SELECT 1 FROM scheduled_task_webhook_pending p - WHERE p.delivery_id = d.delivery_id AND p.state IN ('pending', 'dispatching')) AS pending_window + WHERE p.delivery_id = d.delivery_id AND p.coalesce_key <> '' AND p.state IN ('pending', 'dispatching')) AS pending_window FROM scheduled_task_webhook_deliveries d WHERE task_id = ${input.id} AND delivery_id = ${input.deliveryId} `.pipe( Effect.mapError((cause) => @@ -1576,6 +1573,10 @@ export const layer = Layer.effect( yield* sql`DELETE FROM scheduled_task_webhook_relay_deliveries WHERE relay_delivery_id = ${deliveryId.slice("delivery:relay:".length)}`.pipe( Effect.mapError(dedupeError), ); + // A redelivery admitted by this release must not meet the old row. + yield* sql`DELETE FROM scheduled_task_webhook_pending WHERE delivery_id = ${deliveryId}`.pipe( + Effect.mapError(dedupeError), + ); }, (effect) => sql.withTransaction(effect).pipe(Effect.mapError(dedupeError)), ); @@ -1642,6 +1643,69 @@ export const layer = Layer.effect( ); }); + // Waits on the task's permit, then drains its persisted rows in order. + const drainPendingWebhooks = Effect.fnUntraced( + function* (taskId: ScheduledTaskId) { + const permit = yield* webhookPermit(taskId); + const dispatch = Effect.gen(function* () { + while (true) { + const [entry] = yield* sql<{ + delivery_id: string; + task_json: string; + request_json: string; + state: string; + }>` + SELECT delivery_id, task_json, request_json, state FROM scheduled_task_webhook_pending + WHERE task_id = ${taskId} AND state IN ('dispatching', 'running', 'completed', 'failed') ORDER BY due_at, rowid LIMIT 1 + `; + if (entry === undefined) return; + // Failures drop their row as they release it; a redelivery may already reuse its id. + const remove = sql`DELETE FROM scheduled_task_webhook_pending WHERE delivery_id = ${entry.delivery_id} AND state = 'completed'`; + if (entry.state === "completed") { + yield* remove; + continue; + } + if (entry.state === "failed") { + yield* markDeliveryFailed(entry.delivery_id, "The run failed to start."); + yield* releaseFailedDelivery(entry.delivery_id); + continue; + } + const prepared = yield* Effect.gen(function* () { + const task = yield* decodePendingTask(entry.task_json); + const request = yield* decodePendingRequest(entry.request_json); + const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); + return { task, rendered: renderWebhookPrompt(task.prompt, request, frameId) }; + }).pipe(Effect.exit); + if (Exit.isFailure(prepared)) { + if (Cause.hasInterruptsOnly(prepared.cause)) + return yield* Effect.failCause(prepared.cause); + yield* markDeliveryFailed( + entry.delivery_id, + "The pending webhook payload could not be prepared.", + ); + yield* releaseFailedDelivery(entry.delivery_id); + continue; + } + const { task, rendered } = prepared.value; + if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { + yield* markDeliveryFailed(entry.delivery_id, PROMPT_TOO_LONG); + yield* releaseFailedDelivery(entry.delivery_id); + } else { + yield* sql`UPDATE scheduled_task_webhook_deliveries SET outcome = 'accepted', error = NULL, rendered_prompt = ${rendered.prompt.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT)}, missing_fields_json = ${encodeMissingFieldsJson(rendered.missing)} WHERE delivery_id = ${entry.delivery_id}`; + if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) return; + } + yield* remove; + } + }); + yield* dispatch.pipe( + permit.withPermits(1), + Effect.catch((cause) => Effect.logWarning("Pending webhook dispatch failed", { cause })), + ); + }, + (effect) => + effect.pipe(Effect.withSpan("ScheduledTaskService.dispatchPendingWebhook", { root: true })), + ); + // Delivery attributes belong to the enclosing triggerWebhook span. const observeDelivery = Effect.fnUntraced(function* ( request: WebhookTriggerRequest, @@ -1875,16 +1939,12 @@ export const layer = Layer.effect( const reserve = takeRateSlot(task.id, DateTime.toEpochMillis(now)).pipe( Effect.flatMap((slot) => (slot === "allowed" ? Effect.void : Effect.fail(slot))), ); - const queueKey = `${task.id}\u0000${task.createdAt}`; const checkCapacity = Effect.gen(function* () { const [row] = yield* sql<{ count: number; }>`SELECT COUNT(*) AS count FROM scheduled_task_webhook_pending WHERE task_id = ${task.id} AND state IN ('pending', 'dispatching', 'running')`; - if ( - (row?.count ?? 0) + ((yield* Ref.get(webhookQueued)).get(queueKey) ?? 0) >= - WEBHOOK_MAX_QUEUED_PER_TASK - ) + if ((row?.count ?? 0) >= WEBHOOK_MAX_QUEUED_PER_TASK) return yield* Effect.fail("queue_full" as const); }).pipe( Effect.mapError((cause) => (cause === "queue_full" ? cause : dedupeError(cause))), @@ -1976,27 +2036,8 @@ export const layer = Layer.effect( // Bound the deliveries one task holds, so steady traffic to a stuck // task cannot pile up parked fibers. yield* refundRejectionSlot; - let queueReserved = false; - const release = Effect.suspend(() => { - if (!queueReserved) return Effect.void; - queueReserved = false; - return Ref.update(webhookQueued, (counts) => { - const next = new Map(counts); - const count = (next.get(queueKey) ?? 1) - 1; - if (count <= 0) next.delete(queueKey); - else next.set(queueKey, count); - return next; - }); - }); // Serialize immediate reservations with persisted window admission. - const reserveImmediate = Effect.gen(function* () { - yield* checkCapacity; - yield* reserve; - yield* Ref.update(webhookQueued, (counts) => - new Map(counts).set(queueKey, (counts.get(queueKey) ?? 0) + 1), - ); - queueReserved = true; - }); + const reserveImmediate = Effect.andThen(checkCapacity, reserve); const admitted = yield* admit( Effect.gen(function* () { const outcome = yield* acceptImmediateWebhookDedupe( @@ -2017,9 +2058,8 @@ export const layer = Layer.effect( }); return outcome; }), - ).pipe(Effect.onError(() => release)); + ); if (typeof admitted !== "string") { - yield* release; if (admitted.log) yield* log("rate_limited"); yield* observe(admitted.outcome); return yield* releaseClaim({ @@ -2028,7 +2068,6 @@ export const layer = Layer.effect( } as const); } if (admitted === "duplicate") { - yield* release; yield* observe("duplicate"); return yield* releaseClaim({ _tag: "accepted", @@ -2037,12 +2076,8 @@ export const layer = Layer.effect( } as const); } yield* observe("accepted"); - const permit = yield* webhookPermit(task.id); - yield* runWebhookDelivery(task, deliveryId, rendered.prompt).pipe( - permit.withPermits(1), - Effect.ensuring(release), - Effect.forkIn(serviceScope), - ); + // Start now instead of on the next tick; the permit keeps per-task order. + yield* drainPendingWebhooks(task.id).pipe(Effect.forkIn(serviceScope)); // Dispatch owns the accepted delivery now; retries must keep its claim. claimAcquired = false; return { _tag: "accepted" as const, deliveryId, outcome: "accepted" as const }; @@ -2076,64 +2111,8 @@ export const layer = Layer.effect( : ([true, new Set(active).add(taskId)] as const), ); if (!reserved) return; - const permit = yield* webhookPermit(taskId); - // One worker per task waits on its permit, then drains persisted rows in order. - // Other tasks remain independent and ticks never reload an active task's payloads. - const dispatch = Effect.gen(function* () { - while (true) { - const [entry] = yield* sql<{ - delivery_id: string; - task_json: string; - request_json: string; - state: string; - }>` - SELECT delivery_id, task_json, request_json, state FROM scheduled_task_webhook_pending - WHERE task_id = ${taskId} AND state IN ('dispatching', 'running', 'completed', 'failed') ORDER BY due_at, rowid LIMIT 1 - `; - if (entry === undefined) return; - const remove = sql`DELETE FROM scheduled_task_webhook_pending WHERE delivery_id = ${entry.delivery_id}`; - if (entry.state === "completed" || entry.state === "failed") { - if (entry.state === "failed") { - yield* markDeliveryFailed(entry.delivery_id, "The run failed to start."); - yield* releaseFailedDelivery(entry.delivery_id); - } - yield* remove; - continue; - } - const prepared = yield* Effect.gen(function* () { - const task = yield* decodePendingTask(entry.task_json); - const request = yield* decodePendingRequest(entry.request_json); - const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); - return { task, rendered: renderWebhookPrompt(task.prompt, request, frameId) }; - }).pipe(Effect.exit); - if (Exit.isFailure(prepared)) { - if (Cause.hasInterruptsOnly(prepared.cause)) - return yield* Effect.failCause(prepared.cause); - yield* markDeliveryFailed( - entry.delivery_id, - "The pending webhook payload could not be prepared.", - ); - yield* releaseFailedDelivery(entry.delivery_id); - yield* remove; - continue; - } - const { task, rendered } = prepared.value; - if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { - yield* markDeliveryFailed(entry.delivery_id, PROMPT_TOO_LONG); - yield* releaseFailedDelivery(entry.delivery_id); - } else { - yield* sql`UPDATE scheduled_task_webhook_deliveries SET outcome = 'accepted', error = NULL, rendered_prompt = ${rendered.prompt.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT)}, missing_fields_json = ${encodeMissingFieldsJson(rendered.missing)} WHERE delivery_id = ${entry.delivery_id}`; - if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) - return; - } - yield* remove; - } - }); - yield* dispatch.pipe( - permit.withPermits(1), - Effect.catch((cause) => - Effect.logWarning("Pending webhook dispatch failed", { cause }), - ), + // One tick worker per task; ticks never reload an active task's payloads. + yield* drainPendingWebhooks(taskId).pipe( Effect.ensuring( Ref.update(pendingInFlight, (active) => { const next = new Set(active); @@ -2141,7 +2120,6 @@ export const layer = Layer.effect( return next; }), ), - Effect.withSpan("ScheduledTaskService.dispatchPendingWebhook", { root: true }), Effect.forkIn(serviceScope), ); }), diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index f69680adc79af..ca7e8cec22ba9 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1152,6 +1152,30 @@ it.effect("an accepted delivery queued behind a run still launches after a resta }).pipe(Effect.provide(SqlitePersistence.layerMemory)), ); +it.effect("an immediate delivery queued behind a run is not shown as an open window", () => + Effect.gen(function* () { + const gate = yield* Deferred.make(); + yield* withService( + ({ service, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert(yield* webhookTaskInput()); + yield* service.triggerWebhook(requestFor(task)); + yield* Queue.take(launches); + const queued = yield* service.triggerWebhook(requestFor(task)); + const { delivery } = yield* service.getWebhookDelivery({ + id: task.id, + deliveryId: ScheduledTaskWebhookDeliveryId.make( + queued._tag === "accepted" ? queued.deliveryId : "", + ), + }); + assert.isFalse(delivery.pendingWindow); + yield* Deferred.succeed(gate, undefined); + }), + { gate }, + ); + }), +); + it.effect("guards bad-signature floods separately from valid run admission", () => withService(({ service, sql }) => Effect.gen(function* () { @@ -1190,9 +1214,10 @@ it.effect("start failures release both replay keys and relay claims for redelive yield* Queue.take(launches); assert.equal((yield* sql`SELECT * FROM scheduled_task_webhook_replay_keys`).length, 0); assert.equal((yield* sql`SELECT * FROM scheduled_task_webhook_relay_deliveries`).length, 0); + // The first drain may run the retry; the retry's own drain ends after it. const retry = yield* waitForSpan( service.triggerWebhook(request), - "ScheduledTaskService.runWebhookDelivery", + "ScheduledTaskService.dispatchPendingWebhook", ); assert.equal(retry._tag === "accepted" && retry.outcome, "accepted"); yield* Queue.take(launches); @@ -1278,33 +1303,40 @@ it.effect( }), ); -it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping fails", () => - withService(({ service, sql, launches }) => - Effect.gen(function* () { - const { task } = yield* service.upsert( - yield* webhookTaskInput({ - schedule: { - type: "webhook", - dedupe: { replayKey: "{{body}}" }, - }, - }), - ); - yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks +it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping fails", () => { + const workers = new Map>(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + schedule: { + type: "webhook", + dedupe: { replayKey: "{{body}}" }, + }, + }), + ); + yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks WHEN NEW.last_run_status = 'succeeded' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; - yield* waitForSpan( - service.triggerWebhook(requestFor(task)), - "ScheduledTaskService.runWebhookDelivery", - ); - yield* Queue.take(launches); - assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys`).length, 1); - yield* sql`DROP TRIGGER fail_completion`; - const retry = yield* service.triggerWebhook(requestFor(task)); - assert.equal(retry._tag === "accepted" && retry.outcome, "duplicate"); - assert.equal(yield* Queue.size(launches), 0); - assert.equal((yield* service.list()).tasks[0]!.runCount, 1); - }), - ), -); + yield* waitForSpan( + service.triggerWebhook(requestFor(task)), + "ScheduledTaskService.runWebhookDelivery", + ); + yield* Queue.take(launches); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys`).length, 1); + yield* sql`DROP TRIGGER fail_completion`; + const retry = yield* service.triggerWebhook(requestFor(task)); + assert.equal(retry._tag === "accepted" && retry.outcome, "duplicate"); + assert.equal((yield* service.list()).tasks[0]!.runCount, 0); + // The next pass finishes bookkeeping with the same command id. + yield* drainPendingDispatch(workers); + assert.equal(yield* Queue.size(launches), 0); + assert.equal((yield* service.list()).tasks[0]!.runCount, 1); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`).length, 0); + }), + { workers, completedLaunches: new Set() }, + ); +}); it.effect.each([119_000, 120_000])( "bounds coalesced prompt logs and checks trimmed length for %i body characters", @@ -1675,7 +1707,8 @@ it.effect.each([false, true])( before!.last_run_webhook_delivery_id, window._tag === "accepted" ? window.deliveryId : "", ); - assert.equal(before!.run_count, 0); + // The immediate delivery's drain first finishes the window's bookkeeping, in order. + assert.equal(before!.run_count, 1); if (corrupt) yield* sql`UPDATE scheduled_tasks SET model_selection_json = 'broken' WHERE task_id = ${task.id}`; const restartedWorkers = new Map>(); diff --git a/apps/server/src/scheduledTasks/webhookDedupe.ts b/apps/server/src/scheduledTasks/webhookDedupe.ts index 21d40b25ac5ed..dc15326ea4519 100644 --- a/apps/server/src/scheduledTasks/webhookDedupe.ts +++ b/apps/server/src/scheduledTasks/webhookDedupe.ts @@ -70,13 +70,7 @@ export const acceptWebhookDedupe = Effect.fn("ScheduledTaskService.acceptWebhook } // Coalesced requests retain their validation slot, bounding writes without dropping accepted payloads. yield* claimReplayKey(task.id, replayKey, pending[0]?.delivery_id ?? deliveryId, now); - const requestJson = yield* encodePendingRequest({ - method: request.method, - path: request.path, - query: request.query, - headers: request.headers, - bodyText: request.bodyText, - }); + const requestJson = yield* encodeRequest(request); if (pending.length > 0) { yield* sql`UPDATE scheduled_task_webhook_pending SET request_json = ${requestJson}, delivery_ids_json = json_insert(delivery_ids_json, '$[#]', ${deliveryId}) WHERE delivery_id = ${pending[0]!.delivery_id}`; @@ -107,6 +101,11 @@ export const acceptImmediateWebhookDedupe = Effect.fn( if (yield* hasReplayKey(task.id, replayKey, now)) return "duplicate" as const; yield* reserve; yield* claimReplayKey(task.id, replayKey, deliveryId, now); + // Persisted with the accepted log, so a restart before the run still drains it. + // The empty coalesce key marks it as no window; windows never key on ''. + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO scheduled_task_webhook_pending (delivery_id, task_id, coalesce_key, due_at, state, task_json, request_json, delivery_ids_json) + VALUES (${deliveryId}, ${task.id}, '', ${now}, 'dispatching', ${yield* encodePendingTask(task)}, ${yield* encodeRequest(request)}, json_array(${deliveryId}))`; return "run" as const; }); @@ -123,6 +122,14 @@ const PendingRequestJson = Schema.fromJsonString( }), ); const encodePendingRequest = Schema.encodeEffect(PendingRequestJson); +const encodeRequest = (request: WebhookRequest) => + encodePendingRequest({ + method: request.method, + path: request.path, + query: request.query, + headers: request.headers, + bodyText: request.bodyText, + }); export const decodePendingRequest = Schema.decodeUnknownEffect(PendingRequestJson); export const dedupeError = (cause: unknown) => new ScheduledTaskError({ message: "Could not persist webhook dedupe state.", cause }); From e3c21846c23fc6f412914a3590c73b94e3004e5e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 11:38:36 +0000 Subject: [PATCH 3/6] fix(server): a replaced task's old webhook drain no longer fails the new task's delivery Every accepted delivery now runs through the per-task drain, which selects pending rows by task id. When a task was deleted and recreated under the same id while a drain was mid-run, that old drain could loop, pick up a delivery the new incarnation's drain had already launched, hit the already-running guard, and mark it dispatch_failed, releasing its replay key and relay claim. A drain now stops before selecting another row once its permit is no longer the task's current one, so only the current incarnation's drain takes rows. Rows it leaves behind stay persisted for the next drain. Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.ts | 2 + .../ScheduledTaskService.webhook.test.ts | 88 ++++++++++++++++++- 2 files changed, 89 insertions(+), 1 deletion(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 19d260203f647..f222b64e1aad4 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -1649,6 +1649,8 @@ export const layer = Layer.effect( const permit = yield* webhookPermit(taskId); const dispatch = Effect.gen(function* () { while (true) { + // A delete-and-recreate swaps the permit; the new incarnation's drain owns its rows. + if ((yield* Ref.get(webhookPermits)).get(taskId) !== permit) return; const [entry] = yield* sql<{ delivery_id: string; task_json: string; diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index ca7e8cec22ba9..c1f2f084519a5 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -162,6 +162,7 @@ const withService = ( }) => Effect.Effect, options: { readonly onFrame?: () => void; + readonly beforeFrame?: () => Effect.Effect; readonly frameGate?: { readonly started: Deferred.Deferred; readonly release: Deferred.Deferred; @@ -179,7 +180,9 @@ const withService = ( const launches = yield* Queue.unbounded(); const secretsByRef = new Map(); const layerDependencies = Layer.mergeAll( - options.frameGate === undefined && options.onFrame === undefined + options.frameGate === undefined && + options.onFrame === undefined && + options.beforeFrame === undefined ? NodePlatformCrypto.layer : Layer.effect( Crypto.Crypto, @@ -190,6 +193,8 @@ const withService = ( randomBytes: (size: number) => Effect.suspend(() => { if (size === 6) options.onFrame?.(); + if (size === 6 && options.beforeFrame !== undefined) + return options.beforeFrame().pipe(Effect.andThen(crypto.randomBytes(size))); return size === 6 && options.frameGate !== undefined ? Deferred.succeed(options.frameGate.started, undefined).pipe( Effect.andThen(Deferred.await(options.frameGate.release)), @@ -1176,6 +1181,87 @@ it.effect("an immediate delivery queued behind a run is not shown as an open win }), ); +it.effect("a replaced task's old drain leaves the new incarnation's delivery alone", () => + Effect.gen(function* () { + const firstRun = yield* Deferred.make(); + const secondRun = yield* Deferred.make(); + const newDrainPreparing = yield* Deferred.make(); + const releaseNewDrain = yield* Deferred.make(); + const oldDrainDecided = yield* Deferred.make(); + const releaseOldDrain = yield* Deferred.make(); + const drainsDone = yield* Deferred.make(); + let frames = 0; + let drainEnds = 0; + let firstCommandId: string | undefined; + // Ends of the drain span: the old drain stops first, then the new one. + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (...args) => { + end(...args); + if (span.name !== "ScheduledTaskService.dispatchPendingWebhook") return; + drainEnds++; + if (drainEnds === 1) Deferred.doneUnsafe(oldDrainDecided, Effect.void); + if (drainEnds === 2) Deferred.doneUnsafe(drainsDone, Effect.void); + }; + return span; + }, + }); + yield* withService( + ({ service, launches }) => + Effect.gen(function* () { + const { task: first } = yield* service.upsert(yield* webhookTaskInput()); + yield* service.triggerWebhook(requestFor(first)); + yield* Queue.take(launches); + yield* service.delete({ id: first.id }); + const { task } = yield* service.upsert(yield* webhookTaskInput()); + const accepted = yield* service.triggerWebhook(requestFor(task)); + const deliveryId = accepted._tag === "accepted" ? accepted.deliveryId : ""; + yield* Deferred.await(newDrainPreparing); + // The old run ends while the new drain holds the second delivery. + yield* Deferred.succeed(firstRun, undefined); + yield* Deferred.await(oldDrainDecided); + yield* Deferred.succeed(releaseNewDrain, undefined); + assert.equal( + (yield* Queue.take(launches)).commandId, + `scheduled-task:${task.id}:webhook:${deliveryId}`, + ); + yield* Deferred.succeed(releaseOldDrain, undefined); + yield* Deferred.succeed(secondRun, undefined); + yield* Deferred.await(drainsDone); + const { delivery } = yield* service.getWebhookDelivery({ + id: task.id, + deliveryId: ScheduledTaskWebhookDeliveryId.make(deliveryId), + }); + assert.equal(delivery.outcome, "accepted"); + assert.isTrue(Option.isNone(yield* Queue.poll(launches))); + }).pipe(Effect.withTracer(tracer)), + { + gateForTask: (input) => { + firstCommandId ??= input.commandId; + return input.commandId === firstCommandId ? firstRun : secondRun; + }, + workers: new Map(), + beforeFrame: () => { + frames++; + if (frames === 4) + return Deferred.succeed(newDrainPreparing, undefined).pipe( + Effect.andThen(Deferred.await(releaseNewDrain)), + ); + // Trigger and drain each render a frame, so only an old drain that + // took the new delivery reaches a fifth. + if (frames === 5) + return Deferred.succeed(oldDrainDecided, undefined).pipe( + Effect.andThen(Deferred.await(releaseOldDrain)), + ); + return Effect.void; + }, + }, + ); + }), +); + it.effect("guards bad-signature floods separately from valid run admission", () => withService(({ service, sql }) => Effect.gen(function* () { From ad40333ff8ec3ef311aa1676798ed24c3a3f7baf Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:36:45 -0800 Subject: [PATCH 4/6] test(server): restart delivery test waits on the drain, not a timeout Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.webhook.test.ts | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 772024f19c0e2..1a6650c5ec03e 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1288,13 +1288,10 @@ it.effect("an accepted delivery queued behind a run still launches after a resta yield* withService( ({ launches }) => Effect.gen(function* () { - const launched = yield* Effect.gen(function* () { - while ((yield* Queue.take(launches)).commandId !== commandId); - }).pipe(Effect.timeoutOption("1 minute"), Effect.forkChild); - yield* workers.get("webhook-coalescing")!; - yield* TestClock.adjust("1 minute"); - assert.isTrue( - Option.isSome(yield* Fiber.join(launched)), + yield* drainPendingDispatch(workers); + assert.include( + (yield* Queue.clear(launches)).map((launch) => launch.commandId), + commandId, "the accepted delivery was never launched after the restart", ); }), From ac887cfcaffcd456b61c4d965abdb947f1581273 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:03:44 -0800 Subject: [PATCH 5/6] fix(server): keep at most one waiting webhook drain per task Co-Authored-By: Claude Opus 5.5 --- .../src/scheduledTasks/ScheduledTaskService.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index ab59d6afd3605..c2cd0ae8e77f9 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -457,6 +457,7 @@ export const layer = Layer.effect( const webhookPermits = yield* Ref.make>( new Map(), ); + const waitingDrains = yield* Ref.make>(new Map()); const webhookRateWindows = yield* Ref.make>(new Map()); const webhookRejectedRateWindows = yield* Ref.make>( new Map(), @@ -1869,7 +1870,22 @@ export const layer = Layer.effect( const drainPendingWebhooks = Effect.fnUntraced( function* (taskId: ScheduledTaskId) { const permit = yield* webhookPermit(taskId); + // One waiter per task is enough: it reads every row committed before it gets the permit. + const token = {}; + const queued = yield* Ref.modify(waitingDrains, (waiting) => + waiting.has(taskId) + ? ([false, waiting] as const) + : ([true, new Map(waiting).set(taskId, token)] as const), + ); + if (!queued) return; + const leaveQueue = Ref.update(waitingDrains, (waiting) => { + if (waiting.get(taskId) !== token) return waiting; + const next = new Map(waiting); + next.delete(taskId); + return next; + }); const dispatch = Effect.gen(function* () { + yield* leaveQueue; while (true) { // A delete-and-recreate swaps the permit; the new incarnation's drain owns its rows. if ((yield* Ref.get(webhookPermits)).get(taskId) !== permit) return; @@ -1936,6 +1952,7 @@ export const layer = Layer.effect( }); yield* dispatch.pipe( permit.withPermits(1), + Effect.ensuring(leaveQueue), Effect.catch((cause) => Effect.logWarning("Pending webhook dispatch failed", { cause })), ); }, From 8a80bf7ee17186d02f60ecfe51f9f185db176fc0 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:14:43 -0800 Subject: [PATCH 6/6] fix(server): scope the waiting webhook drain to the task's permit Co-Authored-By: Claude Opus 5.5 --- .../src/scheduledTasks/ScheduledTaskService.ts | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index c2cd0ae8e77f9..efcbb53fabc01 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -457,7 +457,7 @@ export const layer = Layer.effect( const webhookPermits = yield* Ref.make>( new Map(), ); - const waitingDrains = yield* Ref.make>(new Map()); + const waitingDrains = yield* Ref.make>(new Map()); const webhookRateWindows = yield* Ref.make>(new Map()); const webhookRejectedRateWindows = yield* Ref.make>( new Map(), @@ -1870,18 +1870,18 @@ export const layer = Layer.effect( const drainPendingWebhooks = Effect.fnUntraced( function* (taskId: ScheduledTaskId) { const permit = yield* webhookPermit(taskId); - // One waiter per task is enough: it reads every row committed before it gets the permit. + // One waiter per permit is enough: it reads every row committed before it gets the permit. const token = {}; const queued = yield* Ref.modify(waitingDrains, (waiting) => - waiting.has(taskId) + waiting.has(permit) ? ([false, waiting] as const) - : ([true, new Map(waiting).set(taskId, token)] as const), + : ([true, new Map(waiting).set(permit, token)] as const), ); if (!queued) return; const leaveQueue = Ref.update(waitingDrains, (waiting) => { - if (waiting.get(taskId) !== token) return waiting; + if (waiting.get(permit) !== token) return waiting; const next = new Map(waiting); - next.delete(taskId); + next.delete(permit); return next; }); const dispatch = Effect.gen(function* () {