From 7b0bccfe27786c14420c838de55649eb733f5bf4 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 13 Sep 2026 22:43:35 +1000 Subject: [PATCH 1/2] fix(server): skip corrupt scheduled-task rows instead of stopping the scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A malformed persisted task_id threw a schema-constructor defect inside decodeRow, which escaped Effect.result and killed the due-task poll fiber — one corrupt row permanently stopped all scheduled work. The id now decodes through the ScheduledTask schema, and diagnostics carry a typed taskId only when the stored id itself decodes, so the row is skipped with a warning while healthy due tasks still run. Two same-class corrupt-row defects are fixed alongside: an unparseable stored next_run_at would defect the poll via DateTime.makeUnsafe, and a schema-valid interval that overflows the representable DateTime range threw inside nextRunAt — both are now treated as rows with no runnable next occurrence. Startup recovery also used `task_id = ...`, which never matches a NULL id, so those rows stayed stuck in 'running'; it now uses `IS`. Done by swe-2-high through Devin CLI. --- .../ScheduledTaskService.test.ts | 254 +++++++++++++++--- .../scheduledTasks/ScheduledTaskService.ts | 46 +++- 2 files changed, 256 insertions(+), 44 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 52c79619c663b..44d338d466540 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -1,59 +1,115 @@ +import * as NodeUtil from "node:util"; + +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import { assert, it } from "@effect/vitest"; +import { ScheduledTaskError } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Deferred from "effect/Deferred"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; + +import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; -import { listDueTasks } from "./ScheduledTaskService.ts"; +import { layer as scheduledTaskServiceLayer, listDueTasks } from "./ScheduledTaskService.ts"; + +const isScheduledTaskError = Schema.is(ScheduledTaskError); + +const insertRow = ( + sql: SqlClient.SqlClient, + row: { + id: string | null; + next: string | null; + enabled: number; + status: string; + scheduleJson?: string; + prompt?: string; + }, + now: string, +) => + sql`INSERT INTO scheduled_tasks ${sql.insert({ + task_id: row.id, + title: "task", + prompt: row.prompt ?? "Run task", + enabled: row.enabled, + schedule_json: row.scheduleJson ?? '{"type":"interval","everyMs":60000}', + project_id: "project:test", + thread_id: null, + workspace_strategy_json: '{"type":"root"}', + model_selection_json: '{"instanceId":"codex","model":"gpt-5"}', + runtime_mode: "full-access", + interaction_mode: "default", + created_by: "user", + creation_source: "web", + created_at: now, + updated_at: now, + next_run_at: row.next, + last_run_at: null, + last_run_status: row.status, + last_run_error: null, + run_count: 0, + })}`; -it.effect("loads only due tasks and skips a corrupt due row without decoding settled tasks", () => +it.effect("loads only due tasks and skips corrupt due rows without decoding settled tasks", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const now = "2026-09-09T12:00:00.000Z"; + const secret = "PROMPT-SECRET-DO-NOT-LOG"; + const credential = "CREDENTIAL-SECRET-DO-NOT-LOG"; for (const row of [ - { id: "due-now", next: now, enabled: 1, status: "never", corrupt: false }, + { id: "due-now", next: now, enabled: 1, status: "never" }, + // Ids that cannot decode as ScheduledTaskId — empty, whitespace-only, + // and NULL (SQLite's TEXT primary key admits it). + { id: "", next: now, enabled: 1, status: "never" }, + { id: " ", next: now, enabled: 1, status: "never" }, + { id: null, next: now, enabled: 1, status: "never" }, { id: "due-earlier", next: "2026-09-09T11:00:00.000Z", enabled: 1, status: "failed", - corrupt: false, }, - { id: "due-corrupt", next: now, enabled: 1, status: "never", corrupt: true }, - { id: "disabled", next: now, enabled: 0, status: "never", corrupt: true }, + // An unparseable next_run_at sorts before every real timestamp, so it + // passes the due filter and would defect the poll if not skipped. + { id: "due-bad-date", next: "", enabled: 1, status: "never" }, + { + id: "due-corrupt", + next: now, + enabled: 1, + status: "never", + scheduleJson: "broken", + }, + { + id: "due-cred", + next: now, + enabled: 1, + status: "never", + scheduleJson: `{"type":"interval","everyMs":"${credential}"}`, + }, + { id: "disabled", next: now, enabled: 0, status: "never", scheduleJson: "broken" }, { id: "future", next: "2026-09-09T12:00:00.001Z", enabled: 1, status: "never", - corrupt: true, + scheduleJson: "broken", + }, + { + id: "unscheduled", + next: null, + enabled: 1, + status: "never", + scheduleJson: "broken", }, - { id: "unscheduled", next: null, enabled: 1, status: "never", corrupt: true }, - { id: "running", next: now, enabled: 1, status: "running", corrupt: true }, + { id: "running", next: now, enabled: 1, status: "running", scheduleJson: "broken" }, ]) { - yield* sql`INSERT INTO scheduled_tasks ${sql.insert({ - task_id: row.id, - title: row.id, - prompt: "Run task", - enabled: row.enabled, - schedule_json: row.corrupt ? "broken" : '{"type":"interval","everyMs":60000}', - project_id: "project:test", - thread_id: null, - workspace_strategy_json: '{"type":"root"}', - model_selection_json: '{"instanceId":"codex","model":"gpt-5"}', - runtime_mode: "full-access", - interaction_mode: "default", - created_by: "user", - creation_source: "web", - created_at: now, - updated_at: now, - next_run_at: row.next, - last_run_at: null, - last_run_status: row.status, - last_run_error: null, - run_count: 0, - })}`; + yield* insertRow(sql, { ...row, prompt: secret }, now); } const warnings: unknown[] = []; const tasks = yield* listDueTasks(DateTime.makeUnsafe(now)).pipe( @@ -69,10 +125,144 @@ it.effect("loads only due tasks and skips a corrupt due row without decoding set tasks.map((task) => task.id), ["due-earlier", "due-now"], ); - assert.equal(warnings.length, 1); + // Corrupt rows sort between the healthy due rows; each is skipped with a + // warning instead of stopping the poll. + assert.equal(warnings.length, 6); + const annotations = warnings.map( + (warning) => (warning as ReadonlyArray)[1] as Record, + ); + assert.deepEqual( + annotations.map((annotation) => annotation.taskId), + ["due-bad-date", null, "", " ", "due-corrupt", "due-cred"], + ); + // The typed diagnostic carries a taskId only when the stored id itself + // decodes; the corrupt-id rows are still skipped, not fatal. + const causes = annotations.map((annotation) => annotation.cause); + assert.deepEqual( + causes.map((cause) => (isScheduledTaskError(cause) ? cause.taskId : undefined)), + [undefined, undefined, undefined, undefined, "due-corrupt", "due-cred"], + ); + for (const cause of causes) { + if (cause !== undefined) assert.isTrue(isScheduledTaskError(cause)); + } + const rendered = NodeUtil.inspect(annotations, { depth: null }); + assert.isFalse(rendered.includes(secret)); + assert.isFalse(rendered.includes(credential)); const running = yield* sql<{ last_run_status: string; }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'running'`; assert.equal(running[0]?.last_run_status, "running"); }).pipe(Effect.provide(SqlitePersistenceMemory)), ); + +it.effect( + "releases interrupted runs on startup and still executes due tasks, skipping corrupt rows", + () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-09-09T12:00:00.000Z"; + // An interval that decodes but overflows the representable DateTime + // range must not defect recovery or the poll. + const hugeInterval = '{"type":"interval","everyMs":9000000000000000}'; + for (const row of [ + { id: "stuck-valid", next: now, status: "running" }, + { id: "stuck-corrupt", next: now, status: "running", scheduleJson: "broken" }, + { id: "stuck-huge", next: now, status: "running", scheduleJson: hugeInterval }, + { id: null, next: now, status: "running" }, + // A due row with an unparseable next_run_at must not defect the poll + // before the healthy due tasks run. + { id: "due-bad-date", next: "", status: "never" }, + { id: "due-healthy", next: now, status: "never" }, + { id: "due-huge", next: now, status: "never", scheduleJson: hugeInterval }, + { id: "due-second", next: now, status: "never" }, + { id: "due-third", next: now, status: "never" }, + ]) { + yield* insertRow(sql, { ...row, enabled: 1 }, now); + } + + // it.effect freezes the clock; pin it just past the seeded due time so + // the poll sees the rows as due. + yield* TestClock.setTime(Date.parse(now) + 1_000); + // Runs are dispatched serially, so the fourth dispatch can only happen + // after the first three runs' completions were recorded — a drain + // receipt for the poll fiber, never a sleep. The fourth launch is held + // open so its in-flight state is observed deterministically. + const dispatched = yield* Ref.make(0); + const lastDispatched = yield* Deferred.make(); + const releaseLast = yield* Deferred.make(); + yield* Effect.scoped( + Effect.gen(function* () { + yield* Layer.build( + Layer.provideMerge( + scheduledTaskServiceLayer, + Layer.mergeAll( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.updateAndGet(dispatched, (n) => n + 1).pipe( + Effect.andThen((n) => + n === 4 + ? Deferred.succeed(lastDispatched, undefined).pipe( + Effect.andThen(Deferred.await(releaseLast)), + ) + : Effect.void, + ), + Effect.andThen(Effect.die(new Error("test launch failure"))), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + NodeCrypto.layer, + ), + ), + ); + // The poll loop delays before its first pass, so the frozen test + // clock must advance once to trigger it. + yield* TestClock.adjust("6 seconds"); + yield* Deferred.await(lastDispatched); + // The fourth dispatch implies the first three completions landed; + // markRunning commits before dispatch, so due-third is 'running'. + const inflight = yield* sql<{ + last_run_status: string; + }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'due-third'`; + assert.equal(inflight[0]?.last_run_status, "running"); + yield* Deferred.succeed(releaseLast, undefined); + }), + ); + + const rows = yield* sql<{ + task_id: string | null; + last_run_status: string; + last_run_error: string | null; + next_run_at: string | null; + run_count: number; + }>`SELECT task_id, last_run_status, last_run_error, next_run_at, run_count + FROM scheduled_tasks ORDER BY task_id`; + const byId = new Map(rows.map((row) => [row.task_id, row])); + assert.equal(rows.length, 9); + // The decodable stuck run is rescheduled for its next occurrence; the + // undecodable and unrepresentable rows are released without a next run. + // A NULL id must still match its own row — `= NULL` never does. + assert.equal(byId.get("stuck-valid")?.last_run_status, "failed"); + assert.isNotNull(byId.get("stuck-valid")?.next_run_at); + assert.equal(byId.get("stuck-corrupt")?.last_run_status, "failed"); + assert.equal(byId.get(null)?.last_run_status, "failed"); + const stuckHuge = byId.get("stuck-huge"); + assert.equal(stuckHuge?.last_run_status, "failed"); + assert.isNull(stuckHuge?.next_run_at); + for (const id of ["stuck-valid", "stuck-corrupt", "stuck-huge", null]) { + const row = byId.get(id); + assert.equal(row?.last_run_error, "Run was interrupted by a server restart."); + assert.equal(row?.run_count, 1); + } + // The healthy due tasks were dispatched and recorded failed runs; the + // corrupt due row was skipped without running. + for (const id of ["due-healthy", "due-huge", "due-second"]) { + const row = byId.get(id); + assert.equal(row?.last_run_status, "failed"); + assert.isTrue(row?.last_run_error?.includes("test launch failure") === true); + assert.equal(row?.run_count, 1); + } + assert.isNull(byId.get("due-huge")?.next_run_at); + assert.equal(byId.get("due-bad-date")?.last_run_status, "never"); + assert.equal(byId.get("due-bad-date")?.run_count, 0); + }).pipe(Effect.provide(SqlitePersistenceMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index af2f7f7dea980..590a464d108b2 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -21,6 +21,7 @@ import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; @@ -33,6 +34,7 @@ import * as ThreadManagementService from "../orchestration-v2/ThreadManagementSe import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts"; const decodeTask = Schema.decodeUnknownEffect(ScheduledTask); +const decodeTaskId = Schema.decodeUnknownOption(ScheduledTaskId); const decodeScheduleJson = Schema.decodeUnknownEffect( Schema.fromJsonString(ScheduledTask.fields.schedule), ); @@ -107,8 +109,14 @@ function nextRunAt( from: DateTime.DateTime, ): string | null { if (!task.enabled) return null; - const next = nextScheduledRunAt(task.schedule, from); - return next === null ? null : iso(next); + // A stored interval can decode yet overflow the representable DateTime + // range; an unrepresentable occurrence means the task has no next run. + try { + const next = nextScheduledRunAt(task.schedule, from); + return next !== null && Number.isFinite(DateTime.toEpochMillis(next)) ? iso(next) : null; + } catch { + return null; + } } function errorMessage(error: unknown): string { @@ -119,12 +127,13 @@ function errorMessage(error: unknown): string { const decodeRow = (row: ScheduledTaskRow) => Effect.gen(function* () { - const id = ScheduledTaskId.make(row.task_id); const schedule = yield* decodeScheduleJson(row.schedule_json); const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json); const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json); return yield* decodeTask({ - id, + // The stored id decodes through the task schema so a corrupt value fails + // as a typed parse error, not a `ScheduledTaskId.make` defect. + id: row.task_id, title: row.title, prompt: row.prompt, enabled: row.enabled === 1, @@ -146,12 +155,15 @@ const decodeRow = (row: ScheduledTaskRow) => runCount: row.run_count, }); }).pipe( - Effect.mapError((cause) => - taskError("Could not decode schedule task row.", { - taskId: ScheduledTaskId.make(row.task_id), + Effect.mapError((cause) => { + // The typed diagnostic can only carry an id that itself decodes; a + // corrupt stored id is omitted rather than re-thrown as a defect. + const taskId = decodeTaskId(row.task_id); + return taskError("Could not decode schedule task row.", { + ...(Option.isSome(taskId) ? { taskId: taskId.value } : {}), cause, - }), - ), + }); + }), ); /** Select poll candidates before decoding their schedules or other JSON payloads. */ @@ -169,7 +181,17 @@ export const listDueTasks = Effect.fn("ScheduledTaskService.listDueTasks")(funct for (const row of rows) { const decoded = yield* Effect.result(decodeRow(row)); if (Result.isSuccess(decoded)) { - tasks.push(decoded.success); + const task = decoded.success; + // next_run_at is a freeform string at the schema level; a stored value + // that cannot parse as a DateTime would defect the poll below, so the + // row is skipped here like any other corrupt row. + if (task.nextRunAt === null || Option.isSome(DateTime.make(task.nextRunAt))) { + tasks.push(task); + } else { + yield* Effect.logWarning("Skipping schedule task row with invalid next_run_at", { + taskId: row.task_id, + }); + } } else { yield* Effect.logWarning("Skipping undecodable schedule task row", { taskId: row.task_id, @@ -633,7 +655,7 @@ export const layer = Layer.effect( next_run_at = ${nextRunAt(decoded.success, now)}, updated_at = ${iso(now)}, run_count = run_count + 1 - WHERE task_id = ${row.task_id} AND last_run_status = 'running' + WHERE task_id IS ${row.task_id} AND last_run_status = 'running' `; return; } @@ -650,7 +672,7 @@ export const layer = Layer.effect( last_run_error = 'Run was interrupted by a server restart.', updated_at = ${iso(now)}, run_count = run_count + 1 - WHERE task_id = ${row.task_id} AND last_run_status = 'running' + WHERE task_id IS ${row.task_id} AND last_run_status = 'running' `; }), { concurrency: 1, discard: true }, From 1ba0b37355efff7766a18c813ee2f2a405ab3e14 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 14 Sep 2026 07:11:40 +1000 Subject: [PATCH 2/2] fix(server): treat corrupt re-read next_run_at as not due in scheduled tasks The dispatch-time stale check re-read the stored next_run_at and parsed it with DateTime.makeUnsafe, so a row corrupted between the poll SELECT and the re-read defected and killed the poll fiber. Parse through the Option-returning DateTime.make and treat an unparseable value as not due, mirroring the listDueTasks guard. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../ScheduledTaskService.test.ts | 79 +++++++++++++++++++ .../scheduledTasks/ScheduledTaskService.ts | 9 ++- 2 files changed, 85 insertions(+), 3 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts index 44d338d466540..700a32a97a222 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.test.ts @@ -266,3 +266,82 @@ it.effect( assert.equal(byId.get("due-bad-date")?.run_count, 0); }).pipe(Effect.provide(SqlitePersistenceMemory)), ); + +it.effect( + "treats a next_run_at corrupted before the dispatch re-read as not due instead of defecting the poll", + () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-09-09T12:00:00.000Z"; + // Equal next_run_at means dispatch order follows task_id order. + for (const row of [ + { id: "run-a", next: now, status: "never" }, + { id: "run-b-victim", next: now, status: "never" }, + { id: "run-c", next: now, status: "never" }, + { id: "run-d", next: now, status: "never" }, + ]) { + yield* insertRow(sql, { ...row, enabled: 1 }, now); + } + yield* TestClock.setTime(Date.parse(now) + 1_000); + const dispatched = yield* Ref.make(0); + const lastDispatched = yield* Deferred.make(); + const releaseLast = yield* Deferred.make(); + yield* Effect.scoped( + Effect.gen(function* () { + yield* Layer.build( + Layer.provideMerge( + scheduledTaskServiceLayer, + Layer.mergeAll( + Layer.mock(ThreadLaunchService.ThreadLaunchService)({ + launch: () => + Ref.updateAndGet(dispatched, (n) => n + 1).pipe( + // A concurrent writer (the CLI shares the SQLite file) + // corrupts the next row between the poll read and its + // dispatch re-read. Serial dispatch makes the + // interleaving deterministic. + Effect.andThen((n) => + n === 1 + ? sql`UPDATE scheduled_tasks + SET next_run_at = '' + WHERE task_id = 'run-b-victim'`.pipe(Effect.asVoid, Effect.orDie) + : n === 3 + ? Deferred.succeed(lastDispatched, undefined).pipe( + Effect.andThen(Deferred.await(releaseLast)), + ) + : Effect.void, + ), + Effect.andThen(Effect.die(new Error("test launch failure"))), + ), + }), + Layer.mock(ThreadManagementService.ThreadManagementService)({}), + NodeCrypto.layer, + ), + ), + ); + yield* TestClock.adjust("6 seconds"); + // The third dispatch (run-d) can only happen after the victim's + // re-read was handled — a receipt that the poll fiber survived. + yield* Deferred.await(lastDispatched); + assert.equal(yield* Ref.get(dispatched), 3); + const rows = yield* sql<{ + task_id: string; + last_run_status: string; + next_run_at: string | null; + run_count: number; + }>`SELECT task_id, last_run_status, next_run_at, run_count + FROM scheduled_tasks ORDER BY task_id`; + const byId = new Map(rows.map((row) => [row.task_id, row])); + const victim = byId.get("run-b-victim"); + assert.equal(victim?.last_run_status, "never"); + assert.equal(victim?.run_count, 0); + assert.equal(victim?.next_run_at, ""); + for (const id of ["run-a", "run-c"]) { + const row = byId.get(id); + assert.equal(row?.last_run_status, "failed"); + assert.equal(row?.run_count, 1); + } + yield* Deferred.succeed(releaseLast, undefined); + }), + ); + }).pipe(Effect.provide(SqlitePersistenceMemory)), +); diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 590a464d108b2..b2eea7b490719 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -478,12 +478,15 @@ export const layer = Layer.effect( } return task; } + // A next_run_at corrupted between the poll read and this re-read must + // not defect the poll; an unparseable value is treated as not due. + const parsedNextRunAt = + active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt); if ( trigger === "scheduled" && (!active.enabled || - active.nextRunAt === null || - DateTime.toEpochMillis(DateTime.makeUnsafe(active.nextRunAt)) > - DateTime.toEpochMillis(startedAt)) + Option.isNone(parsedNextRunAt) || + DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt)) ) { return active; }