Skip to content

fix(server): accepted webhook deliveries survive a server restart #23579

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
juliusmarminge wants to merge 10 commits into
base: main
Choose a base branch
from
+308 −155
Open

fix(server): accepted webhook deliveries survive a server restart #23579

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
fda2b27
test(server): reproduce lost webhook deliveries after restart (#18810)
juliusmarminge Oct 22, 2026
b06e2f2
fix(server): accepted webhook deliveries survive a server restart
juliusmarminge Oct 22, 2026
e3c2184
fix(server): a replaced task's old webhook drain no longer fails the …
juliusmarminge Oct 22, 2026
8277000
Merge remote-tracking branch 'origin/main' into HEAD
juliusmarminge Oct 30, 2026
2449967
Merge remote-tracking branch 'github/main' into HEAD
juliusmarminge Nov 3, 2026
34f2e68
Merge branch 'main' into fix/issue-18810
juliusmarminge Nov 3, 2026
ad40333
test(server): restart delivery test waits on the drain, not a timeout
juliusmarminge Nov 3, 2026
3781b55
Merge remote-tracking branch 'github/main' into HEAD
juliusmarminge Nov 3, 2026
ac887cf
fix(server): keep at most one waiting webhook drain per task
juliusmarminge Nov 3, 2026
8a80bf7
fix(server): scope the waiting webhook drain to the task's permit
juliusmarminge Nov 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 82 additions & 106 deletions apps/server/src/scheduledTasks/ScheduledTaskService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -456,8 +456,6 @@ export const layer = Layer.effect(
const webhookPermits = yield* Ref.make<ReadonlyMap<ScheduledTaskId, Semaphore.Semaphore>>(
new Map(),
);
// Include creation time so late deliveries cannot decrement a recreated task's count.
const webhookQueued = yield* Ref.make<ReadonlyMap<string, number>>(new Map());
const webhookRateWindows = yield* Ref.make<ReadonlyMap<ScheduledTaskId, RateWindow>>(new Map());
const webhookRejectedRateWindows = yield* Ref.make<ReadonlyMap<ScheduledTaskId, RateWindow>>(
new Map(),
Expand Down Expand Up @@ -1548,7 +1546,7 @@ export const layer = Layer.effect(
) =>
sql<WebhookDeliveryRow>`
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(
Expand All @@ -1570,7 +1568,7 @@ export const layer = Layer.effect(
)(function* (input) {
const rows = yield* sql<WebhookDeliveryRow>`
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(
NodeSqliteClient.readOnly,
Expand Down Expand Up @@ -1797,6 +1795,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)),
);
Expand Down Expand Up @@ -1863,6 +1865,71 @@ 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) {
// 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;
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
`.pipe(NodeSqliteClient.readOnly);
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* random.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,
Expand Down Expand Up @@ -2104,16 +2171,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))),
Expand Down Expand Up @@ -2205,27 +2268,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(
Expand All @@ -2246,9 +2290,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({
Expand All @@ -2257,7 +2300,6 @@ export const layer = Layer.effect(
} as const);
}
if (admitted === "duplicate") {
yield* release;
yield* observe("duplicate");
return yield* releaseClaim({
_tag: "accepted",
Expand All @@ -2266,12 +2308,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));
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
Resolved
// Dispatch owns the accepted delivery now; retries must keep its claim.
claimAcquired = false;
return { _tag: "accepted" as const, deliveryId, outcome: "accepted" as const };
Expand Down Expand Up @@ -2306,77 +2344,16 @@ export const layer = Layer.effect(
}
yield* Effect.forEach(
tasks,
({ task_id: 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
`.pipe(NodeSqliteClient.readOnly);
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* random.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;
}
});
// The worker owns its reservation, so an interrupt anywhere still releases it.
return Effect.acquireUseRelease(
({ task_id: taskId }) =>
// One tick worker per task owns its reservation, so an interrupt anywhere still releases it.
// Ticks never reload an active task's payloads.
Effect.acquireUseRelease(
Ref.modify(pendingInFlight, (active) =>
active.has(taskId)
? ([false, active] as const)
: ([true, new Set(active).add(taskId)] as const),
),
(reserved) =>
reserved
? webhookPermit(taskId).pipe(
Effect.flatMap((permit) => dispatch.pipe(permit.withPermits(1))),
Effect.catch((cause) =>
Effect.logWarning("Pending webhook dispatch failed", { cause }),
),
Effect.withSpan("ScheduledTaskService.dispatchPendingWebhook", {
root: true,
}),
)
: Effect.void,
(reserved) => (reserved ? drainPendingWebhooks(taskId) : Effect.void),
(reserved) =>
reserved
? Ref.update(pendingInFlight, (active) => {
Expand All @@ -2385,8 +2362,7 @@ export const layer = Layer.effect(
return next;
})
: Effect.void,
).pipe(Effect.forkIn(serviceScope));
},
).pipe(Effect.forkIn(serviceScope)),
{ discard: true },
);
});
Expand Down
Loading
Loading