Skip to content

fix(server): serialize scheduled task run transitions #21793

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 3 commits into from
+246 −83
Open

fix(server): serialize scheduled task run transitions #21793

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
303fddd
fix(server): serialize scheduled task run transitions
juliusmarminge Oct 19, 2026
d9e88c3
test(server): reject manual reservation after webhook conversion
juliusmarminge Oct 19, 2026
169c0f3
fix(server): recheck live triggers during manual reservation
juliusmarminge Oct 19, 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
115 changes: 115 additions & 0 deletions apps/server/src/scheduledTasks/ScheduledTaskService.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,3 +576,118 @@ it.effect("guarded mutations reject changed bindings and current or new bound-th
assert.deepEqual((yield* service.list()).tasks, [unbound]);
}).pipe(Effect.provide(atomicLayer)),
);

it.effect.each(["project move", "webhook conversion"] as const)(
"manual reservation rejects a %s after the initial load",
(change) =>
Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* insertRow(
sql,
{ id: atomicTaskId, next: "2026-09-09T12:01:00.000Z", enabled: 1, status: "never" },
"2026-09-09T12:00:00.000Z",
);
let changeBeforeReservation = false;
let launches = 0;
const serviceSql = new Proxy(sql, {
get(target, key) {
if (key !== "withTransaction") return Reflect.get(target, key);
return <A, E, R>(body: Effect.Effect<A, E, R>) =>
Effect.suspend(() => {
if (!changeBeforeReservation) return target.withTransaction(body);
changeBeforeReservation = false;
const mutation =
change === "project move"
? sql`UPDATE scheduled_tasks SET project_id = 'project:destination' WHERE task_id = ${atomicTaskId}`
: sql`UPDATE scheduled_tasks SET schedule_json = '{"type":"webhook","signature":null,"maxDeliveryAgeMinutes":null}', next_run_at = NULL WHERE task_id = ${atomicTaskId}`;
return mutation.pipe(Effect.andThen(target.withTransaction(body)));
});
},
});
const context = yield* Layer.build(
ScheduledTaskService.layer.pipe(
Layer.provide(
Layer.mergeAll(
atomicDependencies,
Layer.mock(ThreadLaunchService.ThreadLaunchService)({
launch: () =>
Effect.sync(() => {
launches++;
}).pipe(Effect.andThen(Effect.die("unexpected manual dispatch"))),
}),
),
),
Layer.provide(Layer.succeed(SqlClient.SqlClient, serviceSql)),
),
);
const service = yield* ScheduledTaskService.ScheduledTaskService.pipe(
Effect.provideContext(context),
);
changeBeforeReservation = true;
const rejected = yield* service
.runNow(atomicScope)
.pipe(Effect.match({ onFailure: (error) => error, onSuccess: () => undefined }));
assert.equal(launches, 0);
const task = (yield* service.list()).tasks[0]!;
assert.equal(
task.projectId,
change === "project move" ? "project:destination" : atomicProjectId,
);
assert.equal(task.schedule.type, change === "project move" ? "interval" : "webhook");
assert.equal(task.lastRunStatus, "never");
assert.equal(task.runCount, 0);
assert.isNull(task.lastRunAt);
assert.isTrue(isScheduledTaskError(rejected));
}).pipe(Effect.scoped, Effect.provide(SqlitePersistence.layerMemory)),
);

it.effect("completion preserves a pause committed before its transaction", () =>
Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* insertRow(
sql,
{ id: atomicTaskId, next: "2026-09-09T12:01:00.000Z", enabled: 1, status: "never" },
"2026-09-09T12:00:00.000Z",
);
let pauseBeforeCompletion = false;
const serviceSql = new Proxy(sql, {
get(target, key) {
if (key !== "withTransaction") return Reflect.get(target, key);
return <A, E, R>(body: Effect.Effect<A, E, R>) =>
Effect.suspend(() => {
if (!pauseBeforeCompletion) return target.withTransaction(body);
pauseBeforeCompletion = false;
return sql`UPDATE scheduled_tasks SET enabled = 0, next_run_at = NULL WHERE task_id = ${atomicTaskId}`.pipe(
Effect.andThen(target.withTransaction(body)),
);
});
},
});
const context = yield* Layer.build(
ScheduledTaskService.layer.pipe(
Layer.provide(
Layer.mergeAll(
atomicDependencies,
Layer.mock(ThreadLaunchService.ThreadLaunchService)({
launch: () =>
Effect.sync(() => {
pauseBeforeCompletion = true;
}).pipe(Effect.andThen(Effect.die("test launch failure"))),
}),
),
),
Layer.provide(Layer.succeed(SqlClient.SqlClient, serviceSql)),
),
);
const service = yield* ScheduledTaskService.ScheduledTaskService.pipe(
Effect.provideContext(context),
);
const result = yield* service.runNow(atomicScope);
const task = (yield* service.list()).tasks[0]!;
assert.equal(task.enabled, false);
assert.isNull(task.nextRunAt);
assert.equal(task.lastRunStatus, "failed");
assert.equal(task.runCount, 1);
assert.deepEqual(result.task, task);
}).pipe(Effect.scoped, Effect.provide(SqlitePersistence.layerMemory)),
);
214 changes: 131 additions & 83 deletions apps/server/src/scheduledTasks/ScheduledTaskService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -710,21 +710,27 @@ export const layer = Layer.effect(
// the task forever (it filters out 'running' rows) nor re-fires it
// immediately: the dispatch may already have gone out, so next_run_at must
// advance and run_count must count the attempt.
const releaseStuckRun = (task: ScheduledTask, message: string) =>
const releaseStuckRun = (task: ScheduledTask, message: string, startedAtIso: string) =>
Effect.gen(function* () {
const now = yield* localNow;
const reread = yield* Effect.result(findTask(task.id));
if (Result.isSuccess(reread) && reread.success === null) return;
const source = Result.isSuccess(reread) && reread.success !== null ? reread.success : task;
yield* sql`
yield* sql.withTransaction(
Effect.gen(function* () {
const reread = yield* Effect.result(findTask(task.id));
if (Result.isSuccess(reread) && reread.success === null) return;
const source =
Result.isSuccess(reread) && reread.success !== null ? reread.success : task;
yield* sql`
UPDATE scheduled_tasks
SET last_run_status = 'failed',
last_run_error = ${message},
next_run_at = ${nextRunAt(source, now)},
updated_at = ${iso(now)},
run_count = run_count + 1
WHERE task_id = ${task.id} AND last_run_status = 'running'
AND last_run_at = ${startedAtIso}
`;
}),
);
yield* notifyChanged;
}).pipe(
Effect.catch((cause) =>
Expand Down Expand Up @@ -759,48 +765,60 @@ export const layer = Layer.effect(
const startedAt = yield* localNow;
const startedAtIso = iso(startedAt);

// The in-memory snapshot may be stale: re-read before touching run
// state. The task may have been deleted, paused, or postponed since
// the poll loaded it — none of those may fire.
const active = yield* findTask(task.id);
if (active === null || (projectId !== undefined && active.projectId !== projectId)) {
if (webhook !== undefined) {
// The due check and reservation share the write lock with definition edits.
const reservedTask = yield* sql
.withTransaction(
Effect.gen(function* () {
const active = yield* findTask(task.id);
if (active === null || (projectId !== undefined && active.projectId !== projectId))
return null;
const parsedNextRunAt =
active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt);
if (
trigger === "scheduled" &&
(!active.enabled ||
Option.isNone(parsedNextRunAt) ||
DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt))
)
return { task: active, running: false };
if (trigger === "manual" && active.schedule.type === "webhook") {
return yield* taskError("Webhook tasks run when their URL receives a request.", {
taskId: active.id,
});
}
if (webhook !== undefined) {
const reason =
active.createdAt !== task.createdAt
? "The task was replaced before this delivery ran."
: active.schedule.type !== "webhook"
? "The task's trigger changed before this delivery ran."
: !active.enabled
? "The task was paused before this delivery ran."
: null;
if (reason !== null) return yield* new WebhookDeliverySkipped({ reason });
}
yield* markRunning(active.id, active.projectId, startedAtIso);
reservedAt = startedAtIso;
return { task: active, running: true };
}),
)
.pipe(
Effect.catchTags({
SqlError: (cause) =>
taskError("Could not mark schedule task as running.", { taskId: task.id, cause }),
}),
);
if (reservedTask === null) {
if (webhook !== undefined)
return yield* new WebhookDeliverySkipped({
reason: "The task was deleted before this delivery ran.",
});
}
if (trigger !== "scheduled") {
if (trigger !== "scheduled")
return yield* taskError("Schedule task not found.", { taskId: task.id });
}
return task;
}
const parsedNextRunAt =
active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt);
if (
trigger === "scheduled" &&
(!active.enabled ||
Option.isNone(parsedNextRunAt) ||
DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt))
) {
return active;
}
// A queued delivery must not run a task that was paused, deleted and
// recreated under the same id, or switched to another trigger while
// it waited for its turn.
if (webhook !== undefined) {
const reason =
active.createdAt !== task.createdAt
? "The task was replaced before this delivery ran."
: active.schedule.type !== "webhook"
? "The task's trigger changed before this delivery ran."
: !active.enabled
? "The task was paused before this delivery ran."
: null;
if (reason !== null) return yield* new WebhookDeliverySkipped({ reason });
}

yield* markRunning(active.id, active.projectId, startedAtIso);
reservedAt = startedAtIso;
if (!reservedTask.running) return reservedTask.task;
const active = reservedTask.task;
yield* notifyChanged;

const fireKey =
Expand Down Expand Up @@ -857,35 +875,49 @@ export const layer = Layer.effect(
const runSucceeded = result._tag === "Success";
const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const);
const lastRunError = runSucceeded ? null : errorMessage(result.cause);
const current = yield* findTask(task.id);
const scheduleSource = current ?? task;
const completed: ScheduledTask = {
...scheduleSource,
updatedAt: iso(completedAt),
lastRunAt: startedAtIso,
nextRunAt: nextRunAt(scheduleSource, completedAt),
lastRunStatus,
lastRunError,
runCount: scheduleSource.runCount + 1,
};
if (current !== null) {
// startedAtIso in the guard ensures this writes only to the row this
// run marked as running — a task deleted mid-run and recreated with
// the same id (idempotent commandId replay) must not be stamped.
yield* markCompleted({
id: task.id,
completedAtIso: completed.updatedAt,
nextRunAtIso: completed.nextRunAt,
status: lastRunStatus,
error: lastRunError,
startedAtIso,
});
yield* notifyChanged;
}
const completed = yield* sql
.withTransaction(
Effect.gen(function* () {
const current = yield* findTask(task.id);
const scheduleSource = current ?? task;
const completed: ScheduledTask = {
...scheduleSource,
updatedAt: iso(completedAt),
lastRunAt: startedAtIso,
nextRunAt: nextRunAt(scheduleSource, completedAt),
lastRunStatus,
lastRunError,
runCount: scheduleSource.runCount + 1,
};
if (current !== null) {
// startedAtIso in the guard ensures this writes only to the row this
// run marked as running — a task deleted mid-run and recreated with
// the same id (idempotent commandId replay) must not be stamped.
yield* markCompleted({
id: task.id,
completedAtIso: completed.updatedAt,
nextRunAtIso: completed.nextRunAt,
status: lastRunStatus,
error: lastRunError,
startedAtIso,
});
}
return completed;
}),
)
.pipe(
Effect.catchTags({
SqlError: (cause) =>
taskError("Could not record schedule task run.", { taskId: task.id, cause }),
}),
);
yield* notifyChanged;
return completed;
}).pipe(
Effect.onError((cause) =>
reservedAt === undefined ? Effect.void : releaseStuckRun(task, errorMessage(cause)),
reservedAt === undefined
? Effect.void
: releaseStuckRun(task, errorMessage(cause), reservedAt),
),
Effect.ensuring(
Ref.update(activeRuns, (active) => {
Expand All @@ -901,22 +933,38 @@ export const layer = Layer.effect(
task: ScheduledTask,
now: DateTime.DateTime,
) {
const next = nextRunAt(task, now);
yield* Effect.logInfo("Skipping missed schedule task run", {
taskId: task.id,
missedRunAt: task.nextRunAt,
rescheduledTo: next,
});
yield* sql`
UPDATE scheduled_tasks
SET next_run_at = ${next},
updated_at = ${iso(now)}
WHERE task_id = ${task.id}
`.pipe(
Effect.mapError((cause) =>
taskError("Could not reschedule missed schedule task run.", { taskId: task.id, cause }),
),
);
yield* sql
.withTransaction(
Effect.gen(function* () {
const current = yield* findTask(task.id);
if (
current === null ||
!current.enabled ||
current.nextRunAt !== task.nextRunAt ||
!isSameSchedule(current.schedule, task.schedule)
)
return;
const next = nextRunAt(current, now);
yield* Effect.logInfo("Skipping missed schedule task run", {
taskId: task.id,
missedRunAt: task.nextRunAt,
rescheduledTo: next,
});
yield* sql`
UPDATE scheduled_tasks SET next_run_at = ${next}, updated_at = ${iso(now)}
WHERE task_id = ${task.id}
`;
}),
)
.pipe(
Effect.catchTags({
SqlError: (cause) =>
taskError("Could not reschedule missed schedule task run.", {
taskId: task.id,
cause,
}),
}),
);
yield* notifyChanged;
});

Expand Down
Loading