feat(schedule-engine): stop persisting per-tick schedule state (#3476)
## Summary
Each scheduled-task tick previously issued **3 Prisma `UPDATE`s**
against
`TaskSchedule.lastRunTriggeredAt`,
`TaskScheduleInstance.lastScheduledTimestamp`,
and `TaskScheduleInstance.nextScheduledTimestamp`. All three were pure
denormalization — every value can be derived without persisting.
After this PR `TaskSchedule` and `TaskScheduleInstance` become **near
read-only**:
writes happen only on schedule create / update / delete (rare admin
actions),
so the per-tick autovacuum churn on these hot tables disappears.
## Design
The previous fire time travels forward through the **schedule worker
payload**,
not through the database. Concretely:
- The `schedule.triggerScheduledTask` worker payload gains an optional
`lastScheduleTime: z.coerce.date().optional()` field.
- When the engine fires a schedule, it re-enqueues the next tick with
`lastScheduleTime = scheduleTimestamp` (the just-fired time).
- When the next tick dequeues, `payload.lastTimestamp` is sourced from
`params.lastScheduleTime` directly. No DB round-trip, no cron-derivation
drift across DST boundaries, no caveats around recently-edited cron
expressions.
`payload.lastTimestamp` keeps its `Date | undefined` SDK shape.
First-ever
fires still report `undefined`, so customer `if
(!payload.lastTimestamp)`
first-run patterns keep working.
For Redis jobs that were enqueued **before** this change (which lack
`lastScheduleTime` in their payload), the engine falls back to
`instance.lastScheduledTimestamp` once. Once those drain, the column is
never read again. Revert is code-only; the columns stay in place and can
be dropped in a follow-up once the rollout is stable.
## Files
- `internal-packages/schedule-engine/*` — engine refactor,
`workerCatalog`
schema field, `TriggerScheduleParams` extension, tests updated to assert
on the worker-payload flow rather than DB readbacks.
- `internal-packages/database/prisma/schema.prisma` — `/// @deprecated`
triple-slash docstrings on the three columns. No migration.
- `apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts` —
drops
the `lastRunTriggeredAt` Prisma select; "Last run" cell is approximated
from the cron expression's previous slot, gated on `schedule.createdAt`
so brand-new schedules show "–". UI is best-effort; the runs page is the
source of truth.
- `apps/webapp/app/v3/utils/calculateNextSchedule.server.ts` — adds a
`previousScheduledTimestamp` helper for the UI cell above. Public API
responses (`api.v1.schedules.*`) already compute `nextRun` from cron and
don't expose `lastTimestamp` — no public API change.
- `references/scheduled-tasks/` — new reference project with declarative
schedules at multiple cadences and three throw-on-fail validators
(`first-fire-detector`, `interval-validator`, `upcoming-validator`) for
E2E-verifying the worker-payload flow.
Refs TRI-8891
## Test plan
- [x] `pnpm run typecheck --filter @internal/schedule-engine --filter
webapp`
- [x] `pnpm run build --filter @trigger.dev/core`
- [x] `pnpm run test --filter @internal/schedule-engine` — integration
test
asserts first-fire `lastTimestamp === undefined`, second fire carries
the previous fire's timestamp exactly.
- [x] E2E against local webapp via `references/scheduled-tasks`:
- Fresh schedules attached → all three deprecated columns stay `NULL`
after
multiple fires.
- Redis payload at second fire contains
`"lastScheduleTime":"<previous fire timestamp>"`.
- `TaskRun.payload` and the every-minute task's returned output both
confirm
`lastTimestamp = null` on first fire and `lastTimestamp = <prev fire>`
on
second fire, exactly 60s apart.
- All three throw-on-FAIL validators completed successfully on every
non-first fire.
- [x] Schedules REST API end-to-end (`POST` / `GET` / `PUT` / `activate`
/
`deactivate` / `DELETE`) — `nextRun` recomputed live from cron + tz on
every response, no reads of deprecated columns.
This commit is contained in:
Eric Allam
@@ -0,0 +1,6 @@
|
||||
---
|
||||
area: webapp
|
||||
type: improvement
|
||||
---
|
||||
|
||||
Stop writing per-tick state (`lastScheduledTimestamp`, `nextScheduledTimestamp`, `lastRunTriggeredAt`) on `TaskSchedule` and `TaskScheduleInstance`. The schedule engine now carries the previous fire time forward via the worker queue payload, eliminating ~270K dead-tuple-driven autovacuums per year on these hot tables and the associated `IO:XactSync` mini-spikes on the writer. Customer-facing `payload.lastTimestamp` semantics are unchanged.
|
||||
@@ -6,7 +6,10 @@ import { getLimit } from "~/services/platform.v3.server";
|
||||
import { findCurrentWorkerFromEnvironment } from "~/v3/models/workerDeployment.server";
|
||||
import { ServiceValidationError } from "~/v3/services/baseService.server";
|
||||
import { CheckScheduleService } from "~/v3/services/checkSchedule.server";
|
||||
import { calculateNextScheduledTimestampFromNow } from "~/v3/utils/calculateNextSchedule.server";
|
||||
import {
|
||||
calculateNextScheduledTimestampFromNow,
|
||||
previousScheduledTimestamp,
|
||||
} from "~/v3/utils/calculateNextSchedule.server";
|
||||
import { BasePresenter } from "./basePresenter.server";
|
||||
|
||||
type ScheduleListOptions = {
|
||||
@@ -193,8 +196,8 @@ export class ScheduleListPresenter extends BasePresenter {
|
||||
},
|
||||
},
|
||||
active: true,
|
||||
lastRunTriggeredAt: true,
|
||||
createdAt: true,
|
||||
updatedAt: true,
|
||||
},
|
||||
where: {
|
||||
projectId: project.id,
|
||||
@@ -244,6 +247,29 @@ export class ScheduleListPresenter extends BasePresenter {
|
||||
});
|
||||
|
||||
const schedules: ScheduleListItem[] = rawSchedules.map((schedule) => {
|
||||
// Approximate "last run" from the cron's previous slot. Skip inactive
|
||||
// schedules — the cron's previous slot reflects what *would* have
|
||||
// fired, but a deactivated schedule didn't actually fire there. Skip
|
||||
// when the cron's previous slot predates `updatedAt`: any config
|
||||
// change (cron edited, timezone changed, deactivate/reactivate)
|
||||
// bumps updatedAt, and a slot from before the most recent change
|
||||
// didn't fire under the current configuration. cron-parser throws
|
||||
// on malformed expressions, so degrade to undefined per-row rather
|
||||
// than failing the whole list. UI is best-effort; the runs page is
|
||||
// the source of truth.
|
||||
let lastRun: Date | undefined;
|
||||
if (schedule.active) {
|
||||
try {
|
||||
const cronPrev = previousScheduledTimestamp(
|
||||
schedule.generatorExpression,
|
||||
schedule.timezone
|
||||
);
|
||||
lastRun = cronPrev.getTime() > schedule.updatedAt.getTime() ? cronPrev : undefined;
|
||||
} catch {
|
||||
lastRun = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
id: schedule.id,
|
||||
type: schedule.type,
|
||||
@@ -256,7 +282,7 @@ export class ScheduleListPresenter extends BasePresenter {
|
||||
timezone: schedule.timezone,
|
||||
active: schedule.active,
|
||||
externalId: schedule.externalId,
|
||||
lastRun: schedule.lastRunTriggeredAt ?? undefined,
|
||||
lastRun,
|
||||
nextRun: calculateNextScheduledTimestampFromNow(
|
||||
schedule.generatorExpression,
|
||||
schedule.timezone
|
||||
|
||||
@@ -22,6 +22,20 @@ function calculateNextStep(schedule: string, timezone: string | null, currentDat
|
||||
.toDate();
|
||||
}
|
||||
|
||||
export function previousScheduledTimestamp(
|
||||
schedule: string,
|
||||
timezone: string | null,
|
||||
fromTimestamp: Date = new Date()
|
||||
) {
|
||||
return parseExpression(schedule, {
|
||||
currentDate: fromTimestamp,
|
||||
utc: timezone === null,
|
||||
tz: timezone ?? undefined,
|
||||
})
|
||||
.prev()
|
||||
.toDate();
|
||||
}
|
||||
|
||||
export function nextScheduledTimestamps(
|
||||
cron: string,
|
||||
timezone: string | null,
|
||||
|
||||
@@ -2128,6 +2128,7 @@ model TaskSchedule {
|
||||
///Instances of the schedule that are active
|
||||
instances TaskScheduleInstance[]
|
||||
|
||||
/// @deprecated stop writing 2026-04-30; reads moved out of code (UI now derives from cron's previous slot). Drop in follow-up.
|
||||
lastRunTriggeredAt DateTime?
|
||||
|
||||
project Project @relation(fields: [projectId], references: [id], onDelete: Cascade, onUpdate: Cascade)
|
||||
@@ -2173,7 +2174,9 @@ model TaskScheduleInstance {
|
||||
|
||||
active Boolean @default(true)
|
||||
|
||||
/// @deprecated stop writing 2026-04-30; engine derives from cron + exactScheduleTime. Drop in follow-up.
|
||||
lastScheduledTimestamp DateTime?
|
||||
/// @deprecated stop writing 2026-04-30; engine derives from cron + now(). Drop in follow-up.
|
||||
nextScheduledTimestamp DateTime?
|
||||
|
||||
//you can only have a schedule attached to each environment once
|
||||
|
||||
@@ -11,7 +11,11 @@ import { Logger } from "@trigger.dev/core/logger";
|
||||
import { PrismaClient } from "@trigger.dev/database";
|
||||
import { Worker, type JobHandlerParams } from "@trigger.dev/redis-worker";
|
||||
import { calculateDistributedExecutionTime } from "./distributedScheduling.js";
|
||||
import { calculateNextScheduledTimestamp, nextScheduledTimestamps } from "./scheduleCalculation.js";
|
||||
import {
|
||||
calculateNextScheduledTimestamp,
|
||||
nextScheduledTimestamps,
|
||||
previousScheduledTimestamp,
|
||||
} from "./scheduleCalculation.js";
|
||||
import {
|
||||
RegisterScheduleInstanceParams,
|
||||
ScheduleEngineOptions,
|
||||
@@ -171,13 +175,13 @@ export class ScheduleEngine {
|
||||
instance.taskSchedule.generatorExpression
|
||||
);
|
||||
|
||||
const lastScheduledTimestamp = instance.lastScheduledTimestamp ?? new Date();
|
||||
span.setAttribute("last_scheduled_timestamp", lastScheduledTimestamp.toISOString());
|
||||
const fromTimestamp = params.fromTimestamp ?? new Date();
|
||||
span.setAttribute("from_timestamp", fromTimestamp.toISOString());
|
||||
|
||||
const nextScheduledTimestamp = calculateNextScheduledTimestamp(
|
||||
instance.taskSchedule.generatorExpression,
|
||||
instance.taskSchedule.timezone,
|
||||
lastScheduledTimestamp
|
||||
fromTimestamp
|
||||
);
|
||||
|
||||
span.setAttribute("next_scheduled_timestamp", nextScheduledTimestamp.toISOString());
|
||||
@@ -185,7 +189,7 @@ export class ScheduleEngine {
|
||||
const schedulingDelayMs = nextScheduledTimestamp.getTime() - Date.now();
|
||||
span.setAttribute("scheduling_delay_ms", schedulingDelayMs);
|
||||
|
||||
this.logger.info("Calculated next schedule timestamp", {
|
||||
this.logger.debug("Calculated next schedule timestamp", {
|
||||
instanceId: params.instanceId,
|
||||
taskIdentifier: instance.taskSchedule.taskIdentifier,
|
||||
nextScheduledTimestamp: nextScheduledTimestamp.toISOString(),
|
||||
@@ -194,17 +198,44 @@ export class ScheduleEngine {
|
||||
timezone: instance.taskSchedule.timezone,
|
||||
});
|
||||
|
||||
await this.prisma.taskScheduleInstance.update({
|
||||
where: {
|
||||
id: params.instanceId,
|
||||
},
|
||||
data: {
|
||||
nextScheduledTimestamp,
|
||||
},
|
||||
});
|
||||
// Determine the lastScheduleTime to embed in the next worker job's
|
||||
// payload. If the caller passed it explicitly (the after-fire path
|
||||
// does this with the just-fired timestamp, the after-skip path
|
||||
// carries the existing value forward), use that. Otherwise — every
|
||||
// external caller (deploy sync, schedule upsert, recovery) — derive
|
||||
// from the cron expression's previous slot.
|
||||
//
|
||||
// Without this fallback, every deploy / cron edit would clobber the
|
||||
// existing in-flight job's lastScheduleTime with `undefined`, and
|
||||
// the next fire would surface a frozen DB-column value to the
|
||||
// customer (since this PR stops writing that column). Pure cron
|
||||
// math, no DB read on top of the existing instance load — the
|
||||
// recovery loop already pays the cost of loading the instance.
|
||||
let lastScheduleTime = params.lastScheduleTime;
|
||||
if (lastScheduleTime === undefined) {
|
||||
try {
|
||||
const cronPrev = previousScheduledTimestamp(
|
||||
instance.taskSchedule.generatorExpression,
|
||||
instance.taskSchedule.timezone
|
||||
);
|
||||
// Guarded against the cron's previous slot predating the
|
||||
// instance itself — for a brand-new schedule, the slot is from
|
||||
// before the schedule existed, so `undefined` is the honest
|
||||
// answer (preserves the `if (!payload.lastTimestamp)` first-run
|
||||
// sentinel customers rely on).
|
||||
if (cronPrev.getTime() > instance.createdAt.getTime()) {
|
||||
lastScheduleTime = cronPrev;
|
||||
}
|
||||
} catch {
|
||||
// Malformed cron — leave undefined.
|
||||
}
|
||||
}
|
||||
|
||||
// Enqueue the scheduled task
|
||||
await this.enqueueScheduledTask(params.instanceId, nextScheduledTimestamp);
|
||||
await this.enqueueScheduledTask(
|
||||
params.instanceId,
|
||||
nextScheduledTimestamp,
|
||||
lastScheduleTime
|
||||
);
|
||||
|
||||
// Record metrics
|
||||
this.scheduleRegistrationCounter.add(1, {
|
||||
@@ -244,6 +275,7 @@ export class ScheduleEngine {
|
||||
instanceId: payload.instanceId,
|
||||
finalAttempt: false, // TODO: implement retry logic
|
||||
exactScheduleTime: payload.exactScheduleTime,
|
||||
lastScheduleTime: payload.lastScheduleTime,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -350,14 +382,6 @@ export class ScheduleEngine {
|
||||
skipReason = "schedule_inactive";
|
||||
}
|
||||
|
||||
if (!instance.nextScheduledTimestamp) {
|
||||
this.logger.debug("No next scheduled timestamp", {
|
||||
instanceId: params.instanceId,
|
||||
});
|
||||
shouldTrigger = false;
|
||||
skipReason = "no_next_timestamp";
|
||||
}
|
||||
|
||||
// For development environments, check if there's an active session
|
||||
if (instance.environment.type === "DEVELOPMENT") {
|
||||
this.devEnvironmentCheckCounter.add(1, {
|
||||
@@ -396,15 +420,26 @@ export class ScheduleEngine {
|
||||
}
|
||||
|
||||
// Calculate the schedule timestamp that will be used (regardless of whether we trigger or not)
|
||||
const scheduleTimestamp =
|
||||
params.exactScheduleTime ?? instance.nextScheduledTimestamp ?? new Date();
|
||||
const scheduleTimestamp = params.exactScheduleTime ?? new Date();
|
||||
|
||||
if (shouldTrigger) {
|
||||
// payload.lastTimestamp is the actual previous fire time. Sources, in
|
||||
// order:
|
||||
// 1. params.lastScheduleTime — populated by the engine when this
|
||||
// job was enqueued. Always present for jobs enqueued post-deploy.
|
||||
// 2. instance.lastScheduledTimestamp — backward-compat fallback for
|
||||
// in-flight Redis jobs enqueued by older engines that didn't
|
||||
// include lastScheduleTime in the payload. Once those drain
|
||||
// this fallback never triggers and we can drop the column.
|
||||
// 3. undefined — first-ever fire (no previous fire to point at).
|
||||
const lastTimestamp =
|
||||
params.lastScheduleTime ?? instance.lastScheduledTimestamp ?? undefined;
|
||||
|
||||
const payload = {
|
||||
scheduleId: instance.taskSchedule.friendlyId,
|
||||
type: instance.taskSchedule.type as "DECLARATIVE" | "IMPERATIVE",
|
||||
timestamp: scheduleTimestamp,
|
||||
lastTimestamp: instance.lastScheduledTimestamp ?? undefined,
|
||||
lastTimestamp,
|
||||
externalId: instance.taskSchedule.externalId ?? undefined,
|
||||
timezone: instance.taskSchedule.timezone,
|
||||
upcoming: nextScheduledTimestamps(
|
||||
@@ -422,13 +457,13 @@ export class ScheduleEngine {
|
||||
span.setAttribute("scheduling_accuracy_ms", schedulingAccuracyMs);
|
||||
span.setAttribute("actual_execution_time", actualExecutionTime.toISOString());
|
||||
|
||||
this.logger.info("Triggering scheduled task", {
|
||||
this.logger.debug("Triggering scheduled task", {
|
||||
instanceId: params.instanceId,
|
||||
taskIdentifier: instance.taskSchedule.taskIdentifier,
|
||||
scheduleTimestamp: scheduleTimestamp.toISOString(),
|
||||
actualExecutionTime: actualExecutionTime.toISOString(),
|
||||
schedulingAccuracyMs,
|
||||
lastTimestamp: instance.lastScheduledTimestamp?.toISOString(),
|
||||
lastTimestamp: lastTimestamp?.toISOString(),
|
||||
});
|
||||
|
||||
const triggerStartTime = Date.now();
|
||||
@@ -473,17 +508,7 @@ export class ScheduleEngine {
|
||||
);
|
||||
} else if (result) {
|
||||
if (result.success) {
|
||||
// Update the last run triggered timestamp
|
||||
await this.prisma.taskSchedule.update({
|
||||
where: {
|
||||
id: instance.taskSchedule.id,
|
||||
},
|
||||
data: {
|
||||
lastRunTriggeredAt: new Date(),
|
||||
},
|
||||
});
|
||||
|
||||
this.logger.info("Successfully triggered scheduled task", {
|
||||
this.logger.debug("Successfully triggered scheduled task", {
|
||||
instanceId: params.instanceId,
|
||||
taskIdentifier: instance.taskSchedule.taskIdentifier,
|
||||
durationMs: triggerDuration,
|
||||
@@ -542,20 +567,23 @@ export class ScheduleEngine {
|
||||
});
|
||||
}
|
||||
|
||||
// Always update the last scheduled timestamp and register next run
|
||||
await this.prisma.taskScheduleInstance.update({
|
||||
where: {
|
||||
id: params.instanceId,
|
||||
},
|
||||
data: {
|
||||
lastScheduledTimestamp: scheduleTimestamp,
|
||||
},
|
||||
});
|
||||
// Register the next run. `fromTimestamp` advances on every tick so
|
||||
// the next cron slot keeps marching forward even through skips.
|
||||
// `lastScheduleTime` is the actual previous fire time the next job
|
||||
// will report as `payload.lastTimestamp` — only advance it when we
|
||||
// actually triggered, otherwise carry forward the existing value so
|
||||
// a long pause/disconnect doesn't quietly overwrite the real
|
||||
// last-fire timestamp with a series of skipped slots.
|
||||
const carriedLastScheduleTime = shouldTrigger
|
||||
? scheduleTimestamp
|
||||
: params.lastScheduleTime ?? instance.lastScheduledTimestamp ?? undefined;
|
||||
|
||||
// Register the next run
|
||||
// Rewritten try/catch to use tryCatch utility
|
||||
const [nextRunError] = await tryCatch(
|
||||
this.registerNextTaskScheduleInstance({ instanceId: params.instanceId })
|
||||
this.registerNextTaskScheduleInstance({
|
||||
instanceId: params.instanceId,
|
||||
fromTimestamp: scheduleTimestamp,
|
||||
lastScheduleTime: carriedLastScheduleTime,
|
||||
})
|
||||
);
|
||||
if (nextRunError) {
|
||||
this.logger.error("Failed to schedule next run after execution", {
|
||||
@@ -610,10 +638,17 @@ export class ScheduleEngine {
|
||||
/**
|
||||
* Enqueues a scheduled task with distributed execution timing
|
||||
*/
|
||||
private async enqueueScheduledTask(instanceId: string, exactScheduleTime: Date) {
|
||||
private async enqueueScheduledTask(
|
||||
instanceId: string,
|
||||
exactScheduleTime: Date,
|
||||
lastScheduleTime?: Date
|
||||
) {
|
||||
return startSpan(this.tracer, "enqueueScheduledTask", async (span) => {
|
||||
span.setAttribute("instanceId", instanceId);
|
||||
span.setAttribute("exactScheduleTime", exactScheduleTime.toISOString());
|
||||
if (lastScheduleTime) {
|
||||
span.setAttribute("lastScheduleTime", lastScheduleTime.toISOString());
|
||||
}
|
||||
|
||||
const distributedExecutionTime = calculateDistributedExecutionTime(
|
||||
exactScheduleTime,
|
||||
@@ -646,6 +681,7 @@ export class ScheduleEngine {
|
||||
payload: {
|
||||
instanceId,
|
||||
exactScheduleTime,
|
||||
lastScheduleTime,
|
||||
},
|
||||
availableAt: distributedExecutionTime,
|
||||
});
|
||||
@@ -694,12 +730,12 @@ export class ScheduleEngine {
|
||||
select: {
|
||||
id: true,
|
||||
generatorExpression: true,
|
||||
timezone: true,
|
||||
instances: {
|
||||
select: {
|
||||
id: true,
|
||||
environmentId: true,
|
||||
lastScheduledTimestamp: true,
|
||||
nextScheduledTimestamp: true,
|
||||
createdAt: true,
|
||||
},
|
||||
},
|
||||
},
|
||||
@@ -733,7 +769,7 @@ export class ScheduleEngine {
|
||||
} as { recovered: string[]; skipped: string[] };
|
||||
|
||||
for (const { instance, schedule } of instancesWithSchedule) {
|
||||
this.logger.info("Recovering schedule", {
|
||||
this.logger.debug("Recovering schedule", {
|
||||
schedule,
|
||||
instance,
|
||||
});
|
||||
@@ -774,16 +810,15 @@ export class ScheduleEngine {
|
||||
instance: {
|
||||
id: string;
|
||||
environmentId: string;
|
||||
lastScheduledTimestamp: Date | null;
|
||||
nextScheduledTimestamp: Date | null;
|
||||
createdAt: Date;
|
||||
};
|
||||
schedule: { id: string; generatorExpression: string };
|
||||
schedule: { id: string; generatorExpression: string; timezone: string | null };
|
||||
}) {
|
||||
// inspect the schedule worker to see if there is a job for this instance
|
||||
const job = await this.worker.getJob(`scheduled-task-instance:${instance.id}`);
|
||||
|
||||
if (job) {
|
||||
this.logger.info("Job already exists for instance", {
|
||||
this.logger.debug("Job already exists for instance", {
|
||||
instanceId: instance.id,
|
||||
job,
|
||||
schedule,
|
||||
@@ -792,12 +827,15 @@ export class ScheduleEngine {
|
||||
return "skipped";
|
||||
}
|
||||
|
||||
this.logger.info("No job found for instance, registering next run", {
|
||||
this.logger.debug("No job found for instance, registering next run", {
|
||||
instanceId: instance.id,
|
||||
schedule,
|
||||
});
|
||||
|
||||
// If the job does not exist, register the next run
|
||||
// No `lastScheduleTime` passed — `registerNextTaskScheduleInstance`
|
||||
// will derive it from the cron's previous slot (with a createdAt
|
||||
// guard) so the post-recovery fire reports an accurate
|
||||
// `payload.lastTimestamp`.
|
||||
await this.registerNextTaskScheduleInstance({ instanceId: instance.id });
|
||||
|
||||
return "recovered";
|
||||
|
||||
@@ -29,6 +29,26 @@ function calculateNextStep(schedule: string, timezone: string | null, currentDat
|
||||
.toDate();
|
||||
}
|
||||
|
||||
/**
|
||||
* Cron's previous slot relative to `fromTimestamp`. For a continuously-
|
||||
* running schedule this equals the actual last fire time; for paused or
|
||||
* DST-edge cases it's an approximation. Used only on the recovery path
|
||||
* where the actual last fire isn't recoverable from in-flight worker state.
|
||||
*/
|
||||
export function previousScheduledTimestamp(
|
||||
cron: string,
|
||||
timezone: string | null,
|
||||
fromTimestamp: Date = new Date()
|
||||
) {
|
||||
return parseExpression(cron, {
|
||||
currentDate: fromTimestamp,
|
||||
utc: timezone === null,
|
||||
tz: timezone ?? undefined,
|
||||
})
|
||||
.prev()
|
||||
.toDate();
|
||||
}
|
||||
|
||||
export function nextScheduledTimestamps(
|
||||
cron: string,
|
||||
timezone: string | null,
|
||||
|
||||
@@ -74,8 +74,22 @@ export interface TriggerScheduleParams {
|
||||
instanceId: string;
|
||||
finalAttempt: boolean;
|
||||
exactScheduleTime?: Date;
|
||||
lastScheduleTime?: Date;
|
||||
}
|
||||
|
||||
export interface RegisterScheduleInstanceParams {
|
||||
instanceId: string;
|
||||
/**
|
||||
* Anchor for computing the next cron slot. Defaults to now() when omitted.
|
||||
* This advances on every tick (fired or skipped) so the next slot keeps
|
||||
* marching forward regardless of skip reasons.
|
||||
*/
|
||||
fromTimestamp?: Date;
|
||||
/**
|
||||
* The actual previous fire time to embed in the next worker job's payload,
|
||||
* which becomes that job's `payload.lastTimestamp` on dequeue. Distinct
|
||||
* from `fromTimestamp` so that skipped ticks (inactive schedule, dev env
|
||||
* disconnected, etc.) do NOT advance this — only real fires do.
|
||||
*/
|
||||
lastScheduleTime?: Date;
|
||||
}
|
||||
|
||||
@@ -5,6 +5,11 @@ export const scheduleWorkerCatalog = {
|
||||
schema: z.object({
|
||||
instanceId: z.string(),
|
||||
exactScheduleTime: z.coerce.date(),
|
||||
// Optional for backward compat with in-flight jobs enqueued by older
|
||||
// engines. After deploy, every newly-enqueued job populates this with
|
||||
// the just-fired schedule time so the next dequeue can report
|
||||
// payload.lastTimestamp accurately without a DB round-trip.
|
||||
lastScheduleTime: z.coerce.date().optional(),
|
||||
}),
|
||||
visibilityTimeoutMs: 60_000,
|
||||
retry: {
|
||||
|
||||
@@ -98,30 +98,13 @@ describe("ScheduleEngine Integration", () => {
|
||||
},
|
||||
});
|
||||
|
||||
// Calculate the expected next execution time (next minute boundary)
|
||||
const now = new Date();
|
||||
const expectedExecutionTime = new Date(now);
|
||||
expectedExecutionTime.setMinutes(now.getMinutes() + 1, 0, 0); // Next minute, 0 seconds, 0 milliseconds
|
||||
|
||||
// Calculate the expected upcoming execution times (next 10 minutes after the first execution)
|
||||
const expectedUpcoming = [];
|
||||
for (let i = 1; i <= 10; i++) {
|
||||
const upcoming = new Date(expectedExecutionTime);
|
||||
upcoming.setMinutes(expectedExecutionTime.getMinutes() + i);
|
||||
expectedUpcoming.push(upcoming);
|
||||
}
|
||||
|
||||
// Manually enqueue the first scheduled task to kick off the lifecycle
|
||||
// Manually enqueue the first scheduled task to kick off the lifecycle.
|
||||
// Anchor expectations to the first observed `exactScheduleTime` rather
|
||||
// than a precomputed wall-clock value — registration that happens to
|
||||
// straddle a minute boundary used to flake tests asserting against a
|
||||
// pre-baked "next minute".
|
||||
await engine.registerNextTaskScheduleInstance({ instanceId: scheduleInstance.id });
|
||||
|
||||
// Get the actual nextScheduledTimestamp that was calculated by the engine
|
||||
const instanceAfterRegistration = await prisma.taskScheduleInstance.findFirst({
|
||||
where: { id: scheduleInstance.id },
|
||||
});
|
||||
const actualNextExecution = instanceAfterRegistration?.nextScheduledTimestamp;
|
||||
expect(actualNextExecution).toBeDefined();
|
||||
expect(actualNextExecution).toEqual(expectedExecutionTime);
|
||||
|
||||
// Wait for the first execution
|
||||
console.log("Waiting for first execution...");
|
||||
const startTime = Date.now();
|
||||
@@ -181,6 +164,17 @@ describe("ScheduleEngine Integration", () => {
|
||||
}
|
||||
}
|
||||
|
||||
// Anchor all expectations to what the engine actually fired with, so
|
||||
// the test stays deterministic regardless of when within a minute it
|
||||
// started.
|
||||
const firstScheduledTime = firstExecution.params.exactScheduleTime;
|
||||
const secondScheduledTime = secondExecution.params.exactScheduleTime;
|
||||
expect(firstScheduledTime).toBeDefined();
|
||||
expect(secondScheduledTime).toBeDefined();
|
||||
|
||||
// Each cron slot for "* * * * *" is exactly 60s apart.
|
||||
expect(secondScheduledTime!.getTime() - firstScheduledTime!.getTime()).toBe(60_000);
|
||||
|
||||
// Verify the first execution parameters
|
||||
expect(firstExecution.params).toEqual({
|
||||
taskIdentifier: "test-task",
|
||||
@@ -201,68 +195,152 @@ describe("ScheduleEngine Integration", () => {
|
||||
payload: {
|
||||
scheduleId: "sched_abc123",
|
||||
type: "DECLARATIVE",
|
||||
timestamp: actualNextExecution,
|
||||
lastTimestamp: undefined, // First run has no lastTimestamp
|
||||
timestamp: firstScheduledTime,
|
||||
// First-ever fire: no `lastScheduleTime` carried in the worker
|
||||
// payload and `instance.lastScheduledTimestamp` is null on a
|
||||
// fresh instance, so lastTimestamp is undefined. This preserves
|
||||
// the `if (!payload.lastTimestamp)` first-run sentinel customers
|
||||
// rely on.
|
||||
lastTimestamp: undefined,
|
||||
externalId: "ext-123",
|
||||
timezone: "UTC",
|
||||
upcoming: expect.arrayContaining([expect.any(Date)]),
|
||||
},
|
||||
scheduleInstanceId: scheduleInstance.id,
|
||||
scheduleId: taskSchedule.id,
|
||||
exactScheduleTime: actualNextExecution,
|
||||
exactScheduleTime: firstScheduledTime,
|
||||
});
|
||||
|
||||
// Verify the second execution parameters
|
||||
if (actualNextExecution) {
|
||||
const expectedSecondExecution = new Date(actualNextExecution);
|
||||
expectedSecondExecution.setMinutes(actualNextExecution.getMinutes() + 1);
|
||||
|
||||
expect(secondExecution.params).toEqual({
|
||||
taskIdentifier: "test-task",
|
||||
environment: expect.objectContaining({
|
||||
id: environment.id,
|
||||
type: "PRODUCTION",
|
||||
}),
|
||||
payload: {
|
||||
scheduleId: "sched_abc123",
|
||||
type: "DECLARATIVE",
|
||||
timestamp: expectedSecondExecution,
|
||||
lastTimestamp: actualNextExecution, // Second run should have the first execution time as lastTimestamp
|
||||
externalId: "ext-123",
|
||||
timezone: "UTC",
|
||||
upcoming: expect.arrayContaining([expect.any(Date)]),
|
||||
},
|
||||
scheduleInstanceId: scheduleInstance.id,
|
||||
scheduleId: taskSchedule.id,
|
||||
exactScheduleTime: expectedSecondExecution,
|
||||
});
|
||||
}
|
||||
|
||||
// Verify database updates occurred after both executions
|
||||
const updatedSchedule = await prisma.taskSchedule.findFirst({
|
||||
where: { id: taskSchedule.id },
|
||||
expect(secondExecution.params).toEqual({
|
||||
taskIdentifier: "test-task",
|
||||
environment: expect.objectContaining({
|
||||
id: environment.id,
|
||||
type: "PRODUCTION",
|
||||
}),
|
||||
payload: {
|
||||
scheduleId: "sched_abc123",
|
||||
type: "DECLARATIVE",
|
||||
timestamp: secondScheduledTime,
|
||||
// The previous fire's exactScheduleTime is carried through the
|
||||
// worker payload as `lastScheduleTime` and surfaced here.
|
||||
lastTimestamp: firstScheduledTime,
|
||||
externalId: "ext-123",
|
||||
timezone: "UTC",
|
||||
upcoming: expect.arrayContaining([expect.any(Date)]),
|
||||
},
|
||||
scheduleInstanceId: scheduleInstance.id,
|
||||
scheduleId: taskSchedule.id,
|
||||
exactScheduleTime: secondScheduledTime,
|
||||
});
|
||||
expect(updatedSchedule?.lastRunTriggeredAt).toBeTruthy();
|
||||
expect(updatedSchedule?.lastRunTriggeredAt).toBeInstanceOf(Date);
|
||||
|
||||
const finalInstance = await prisma.taskScheduleInstance.findFirst({
|
||||
where: { id: scheduleInstance.id },
|
||||
});
|
||||
|
||||
// After two executions, lastScheduledTimestamp should be the second execution time
|
||||
if (actualNextExecution && secondExecution.params.exactScheduleTime) {
|
||||
const secondExecutionTime = secondExecution.params.exactScheduleTime;
|
||||
expect(finalInstance?.lastScheduledTimestamp).toEqual(secondExecutionTime);
|
||||
|
||||
// The next scheduled timestamp should be 1 minute after the second execution
|
||||
const expectedThirdExecution = new Date(secondExecutionTime);
|
||||
expectedThirdExecution.setMinutes(secondExecutionTime.getMinutes() + 1);
|
||||
expect(finalInstance?.nextScheduledTimestamp).toEqual(expectedThirdExecution);
|
||||
}
|
||||
} finally {
|
||||
// Clean up: stop the worker
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// Deploy-moment backward compatibility. At deploy time, in-flight Redis jobs
|
||||
// were enqueued by the old engine — their payload has no `lastScheduleTime`
|
||||
// field — and `instance.lastScheduledTimestamp` is still populated (last
|
||||
// written by the old engine pre-deploy). The new engine must report that DB
|
||||
// value as `payload.lastTimestamp` so customers don't see a transient
|
||||
// `undefined` for the one fire per schedule that drains the legacy queue.
|
||||
containerTest(
|
||||
"should fall back to instance.lastScheduledTimestamp when payload lacks lastScheduleTime",
|
||||
{ timeout: 30_000 },
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const triggerCalls: TriggerScheduledTaskParams[] = [];
|
||||
const engine = new ScheduleEngine({
|
||||
prisma,
|
||||
redis: redisOptions,
|
||||
distributionWindow: { seconds: 10 },
|
||||
worker: {
|
||||
concurrency: 1,
|
||||
disabled: true, // Don't actually run the worker — calling triggerScheduledTask directly
|
||||
pollIntervalMs: 1000,
|
||||
},
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
onTriggerScheduledTask: async (params) => {
|
||||
triggerCalls.push(params);
|
||||
return { success: true };
|
||||
},
|
||||
isDevEnvironmentConnectedHandler: vi.fn().mockResolvedValue(true),
|
||||
});
|
||||
|
||||
try {
|
||||
const organization = await prisma.organization.create({
|
||||
data: { title: "Legacy Payload Org", slug: "legacy-payload-org" },
|
||||
});
|
||||
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "Legacy Payload Project",
|
||||
slug: "legacy-payload-project",
|
||||
externalRef: "legacy-payload-ref",
|
||||
organizationId: organization.id,
|
||||
},
|
||||
});
|
||||
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "legacy-payload-env",
|
||||
type: "PRODUCTION",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "tr_legacy_1234",
|
||||
pkApiKey: "pk_legacy_1234",
|
||||
shortcode: "legacy-short",
|
||||
},
|
||||
});
|
||||
|
||||
const taskSchedule = await prisma.taskSchedule.create({
|
||||
data: {
|
||||
friendlyId: "sched_legacy_payload",
|
||||
taskIdentifier: "legacy-payload-task",
|
||||
projectId: project.id,
|
||||
deduplicationKey: "legacy-payload-dedup",
|
||||
userProvidedDeduplicationKey: false,
|
||||
generatorExpression: "*/5 * * * *",
|
||||
generatorDescription: "Every 5 minutes",
|
||||
timezone: "UTC",
|
||||
type: "DECLARATIVE",
|
||||
active: true,
|
||||
externalId: "legacy-ext",
|
||||
},
|
||||
});
|
||||
|
||||
// Pre-populate lastScheduledTimestamp on the instance — simulates the
|
||||
// value the old engine wrote to the DB before this PR deployed.
|
||||
const preDeployLastFire = new Date("2026-04-30T10:00:00.000Z");
|
||||
const scheduleInstance = await prisma.taskScheduleInstance.create({
|
||||
data: {
|
||||
taskScheduleId: taskSchedule.id,
|
||||
environmentId: environment.id,
|
||||
projectId: project.id,
|
||||
active: true,
|
||||
lastScheduledTimestamp: preDeployLastFire,
|
||||
},
|
||||
});
|
||||
|
||||
// Call triggerScheduledTask directly without lastScheduleTime,
|
||||
// simulating an in-flight Redis job enqueued by the old engine.
|
||||
const exactScheduleTime = new Date("2026-04-30T10:05:00.000Z");
|
||||
await engine.triggerScheduledTask({
|
||||
instanceId: scheduleInstance.id,
|
||||
finalAttempt: false,
|
||||
exactScheduleTime,
|
||||
// lastScheduleTime intentionally omitted — legacy payload shape
|
||||
});
|
||||
|
||||
expect(triggerCalls.length).toBe(1);
|
||||
expect(triggerCalls[0].payload.timestamp).toEqual(exactScheduleTime);
|
||||
// Falls back to instance.lastScheduledTimestamp from the DB rather
|
||||
// than reporting undefined for this one transitional fire.
|
||||
expect(triggerCalls[0].payload.lastTimestamp).toEqual(preDeployLastFire);
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
@@ -93,18 +93,14 @@ describe("Schedule Recovery", () => {
|
||||
// Perform recovery
|
||||
await engine.recoverSchedulesInEnvironment(project.id, environment.id);
|
||||
|
||||
// Verify that a job was created
|
||||
// Verify that a job was created. The engine no longer persists
|
||||
// nextScheduledTimestamp; correctness is now determined entirely by the
|
||||
// job sitting in the worker queue.
|
||||
const jobAfterRecovery = await engine.getJob(
|
||||
`scheduled-task-instance:${scheduleInstance.id}`
|
||||
);
|
||||
expect(jobAfterRecovery).not.toBeNull();
|
||||
expect(jobAfterRecovery?.job).toBe("schedule.triggerScheduledTask");
|
||||
|
||||
// Verify the instance was updated with next scheduled timestamp
|
||||
const updatedInstance = await prisma.taskScheduleInstance.findFirst({
|
||||
where: { id: scheduleInstance.id },
|
||||
});
|
||||
expect(updatedInstance?.nextScheduledTimestamp).toBeDefined();
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
@@ -313,20 +309,14 @@ describe("Schedule Recovery", () => {
|
||||
// Perform recovery
|
||||
await engine.recoverSchedulesInEnvironment(project.id, environment.id);
|
||||
|
||||
// Verify that jobs were created for all instances
|
||||
// Verify that jobs were created for all instances. The engine no longer
|
||||
// persists nextScheduledTimestamp — the worker-queue presence is the
|
||||
// source of truth.
|
||||
for (const instance of instances) {
|
||||
const job = await engine.getJob(`scheduled-task-instance:${instance.id}`);
|
||||
expect(job).not.toBeNull();
|
||||
expect(job?.job).toBe("schedule.triggerScheduledTask");
|
||||
}
|
||||
|
||||
// Verify all instances were updated
|
||||
for (const instance of instances) {
|
||||
const updatedInstance = await prisma.taskScheduleInstance.findFirst({
|
||||
where: { id: instance.id },
|
||||
});
|
||||
expect(updatedInstance?.nextScheduledTimestamp).toBeDefined();
|
||||
}
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
@@ -396,4 +386,194 @@ describe("Schedule Recovery", () => {
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// External-caller backward-compat. Deploy sync (`syncDeclarativeSchedules`)
|
||||
// and schedule upsert both call `registerNextTaskScheduleInstance` with no
|
||||
// `lastScheduleTime`. They run on every app deploy and on every cron edit.
|
||||
// For an existing-and-firing schedule, the call must NOT clobber the
|
||||
// worker payload's `lastScheduleTime` with `undefined` — otherwise the
|
||||
// next fire would surface a stale frozen DB-column value to the customer
|
||||
// (since this PR stops writing that column). The function must derive a
|
||||
// sensible `lastScheduleTime` from the cron expression's previous slot
|
||||
// when the caller doesn't pass one.
|
||||
containerTest(
|
||||
"should derive lastScheduleTime from cron when external callers omit it",
|
||||
{ timeout: 30_000 },
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const engine = new ScheduleEngine({
|
||||
prisma,
|
||||
redis: redisOptions,
|
||||
distributionWindow: { seconds: 10 },
|
||||
worker: { concurrency: 1, disabled: true, pollIntervalMs: 1000 },
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
onTriggerScheduledTask: async () => ({ success: true }),
|
||||
isDevEnvironmentConnectedHandler: vi.fn().mockResolvedValue(true),
|
||||
});
|
||||
|
||||
try {
|
||||
const organization = await prisma.organization.create({
|
||||
data: { title: "External Caller Org", slug: "external-caller-org" },
|
||||
});
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "External Caller Project",
|
||||
slug: "external-caller-project",
|
||||
externalRef: "external-caller-ref",
|
||||
organizationId: organization.id,
|
||||
},
|
||||
});
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "external-caller-env",
|
||||
type: "PRODUCTION",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "tr_external_1234",
|
||||
pkApiKey: "pk_external_1234",
|
||||
shortcode: "external-short",
|
||||
},
|
||||
});
|
||||
const taskSchedule = await prisma.taskSchedule.create({
|
||||
data: {
|
||||
friendlyId: "sched_external_caller",
|
||||
taskIdentifier: "external-caller-task",
|
||||
projectId: project.id,
|
||||
deduplicationKey: "external-caller-dedup",
|
||||
userProvidedDeduplicationKey: false,
|
||||
generatorExpression: "*/5 * * * *",
|
||||
generatorDescription: "Every 5 minutes",
|
||||
timezone: "UTC",
|
||||
type: "DECLARATIVE",
|
||||
active: true,
|
||||
},
|
||||
});
|
||||
|
||||
// Backdate the instance so the cron's previous slot postdates
|
||||
// createdAt — this simulates a long-running schedule, the case
|
||||
// Devin flagged (deploy clobbers lastScheduleTime, post-deploy fire
|
||||
// would otherwise read from a frozen DB column).
|
||||
const longAgo = new Date(Date.now() - 24 * 60 * 60 * 1000);
|
||||
const scheduleInstance = await prisma.taskScheduleInstance.create({
|
||||
data: {
|
||||
taskScheduleId: taskSchedule.id,
|
||||
environmentId: environment.id,
|
||||
projectId: project.id,
|
||||
active: true,
|
||||
},
|
||||
});
|
||||
await prisma.taskScheduleInstance.update({
|
||||
where: { id: scheduleInstance.id },
|
||||
data: { createdAt: longAgo },
|
||||
});
|
||||
|
||||
// External-caller pattern — no lastScheduleTime.
|
||||
await engine.registerNextTaskScheduleInstance({
|
||||
instanceId: scheduleInstance.id,
|
||||
});
|
||||
|
||||
const job = await engine.getJob(`scheduled-task-instance:${scheduleInstance.id}`);
|
||||
expect(job).not.toBeNull();
|
||||
// The function should have derived lastScheduleTime from cron,
|
||||
// putting a real timestamp into the worker payload rather than
|
||||
// undefined. The Redis worker stores payloads as JSON, so the value
|
||||
// is a string when read back here — Zod re-coerces it to Date on
|
||||
// dequeue (workerCatalog uses `z.coerce.date()`).
|
||||
const enqueuedLastScheduleTime = (job?.item as { lastScheduleTime?: string })
|
||||
.lastScheduleTime;
|
||||
expect(enqueuedLastScheduleTime).toBeDefined();
|
||||
const derived = new Date(enqueuedLastScheduleTime!);
|
||||
// The derived value should match the cron's previous slot — for
|
||||
// `*/5 * * * *`, a 5-minute boundary in the recent past.
|
||||
expect(derived.getTime()).toBeLessThan(Date.now());
|
||||
expect(derived.getUTCSeconds()).toBe(0);
|
||||
expect(derived.getUTCMinutes() % 5).toBe(0);
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
|
||||
// Brand-new schedules must NOT receive a cron-derived lastScheduleTime —
|
||||
// the cron's previous slot predates the instance, so it's not a real
|
||||
// previous fire. The first-run sentinel (`if (!payload.lastTimestamp)`)
|
||||
// must keep working.
|
||||
containerTest(
|
||||
"should leave lastScheduleTime undefined for brand-new schedules",
|
||||
{ timeout: 30_000 },
|
||||
async ({ prisma, redisOptions }) => {
|
||||
const engine = new ScheduleEngine({
|
||||
prisma,
|
||||
redis: redisOptions,
|
||||
distributionWindow: { seconds: 10 },
|
||||
worker: { concurrency: 1, disabled: true, pollIntervalMs: 1000 },
|
||||
tracer: trace.getTracer("test", "0.0.0"),
|
||||
onTriggerScheduledTask: async () => ({ success: true }),
|
||||
isDevEnvironmentConnectedHandler: vi.fn().mockResolvedValue(true),
|
||||
});
|
||||
|
||||
try {
|
||||
const organization = await prisma.organization.create({
|
||||
data: { title: "Brand New Org", slug: "brand-new-org" },
|
||||
});
|
||||
const project = await prisma.project.create({
|
||||
data: {
|
||||
name: "Brand New Project",
|
||||
slug: "brand-new-project",
|
||||
externalRef: "brand-new-ref",
|
||||
organizationId: organization.id,
|
||||
},
|
||||
});
|
||||
const environment = await prisma.runtimeEnvironment.create({
|
||||
data: {
|
||||
slug: "brand-new-env",
|
||||
type: "PRODUCTION",
|
||||
projectId: project.id,
|
||||
organizationId: organization.id,
|
||||
apiKey: "tr_brandnew_1234",
|
||||
pkApiKey: "pk_brandnew_1234",
|
||||
shortcode: "brandnew-short",
|
||||
},
|
||||
});
|
||||
const taskSchedule = await prisma.taskSchedule.create({
|
||||
data: {
|
||||
friendlyId: "sched_brand_new",
|
||||
taskIdentifier: "brand-new-task",
|
||||
projectId: project.id,
|
||||
deduplicationKey: "brand-new-dedup",
|
||||
userProvidedDeduplicationKey: false,
|
||||
// Hourly cron — the previous slot is plausibly minutes-to-an-hour
|
||||
// ago, comfortably predating an instance just created.
|
||||
generatorExpression: "0 * * * *",
|
||||
generatorDescription: "Hourly",
|
||||
timezone: "UTC",
|
||||
type: "DECLARATIVE",
|
||||
active: true,
|
||||
},
|
||||
});
|
||||
const scheduleInstance = await prisma.taskScheduleInstance.create({
|
||||
data: {
|
||||
taskScheduleId: taskSchedule.id,
|
||||
environmentId: environment.id,
|
||||
projectId: project.id,
|
||||
active: true,
|
||||
},
|
||||
});
|
||||
|
||||
await engine.registerNextTaskScheduleInstance({
|
||||
instanceId: scheduleInstance.id,
|
||||
});
|
||||
|
||||
const job = await engine.getJob(`scheduled-task-instance:${scheduleInstance.id}`);
|
||||
expect(job).not.toBeNull();
|
||||
const enqueuedLastScheduleTime = (job?.item as { lastScheduleTime?: Date }).lastScheduleTime;
|
||||
// Brand-new schedule: cron's previous slot predates instance.createdAt,
|
||||
// so the function leaves lastScheduleTime undefined — the first fire
|
||||
// will report `payload.lastTimestamp: undefined` and customer first-run
|
||||
// sentinel patterns keep working.
|
||||
expect(enqueuedLastScheduleTime).toBeUndefined();
|
||||
} finally {
|
||||
await engine.quit();
|
||||
}
|
||||
}
|
||||
);
|
||||
});
|
||||
|
||||
Generated
+19
@@ -2828,6 +2828,25 @@ importers:
|
||||
specifier: 5.5.4
|
||||
version: 5.5.4
|
||||
|
||||
references/scheduled-tasks:
|
||||
dependencies:
|
||||
'@trigger.dev/sdk':
|
||||
specifier: workspace:*
|
||||
version: link:../../packages/trigger-sdk
|
||||
zod:
|
||||
specifier: 3.25.76
|
||||
version: 3.25.76
|
||||
devDependencies:
|
||||
'@trigger.dev/build':
|
||||
specifier: workspace:*
|
||||
version: link:../../packages/build
|
||||
trigger.dev:
|
||||
specifier: workspace:*
|
||||
version: link:../../packages/cli-v3
|
||||
typescript:
|
||||
specifier: 5.5.4
|
||||
version: 5.5.4
|
||||
|
||||
references/seed:
|
||||
dependencies:
|
||||
'@sinclair/typebox':
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
# scheduled-tasks reference
|
||||
|
||||
E2E test bed for the schedule engine. Designed to verify the worker-payload flow
|
||||
that carries `payload.lastTimestamp` forward across fires (no DB round-trip,
|
||||
no cron-derivation drift).
|
||||
|
||||
## Setup
|
||||
|
||||
1. Create a project in the local webapp at http://localhost:3030 and copy the
|
||||
project ref from Project Settings.
|
||||
2. Replace `proj_REPLACE_ME` in `trigger.config.ts` with that ref.
|
||||
3. From this directory:
|
||||
|
||||
```bash
|
||||
pnpm exec trigger login -a http://localhost:3030 --profile local
|
||||
pnpm exec trigger dev --profile local
|
||||
```
|
||||
|
||||
4. Open the project in the dashboard and visit each schedule. Click "Attach to
|
||||
environment" → dev. The first fire happens at the next cron slot.
|
||||
|
||||
## What to look for
|
||||
|
||||
- **`first-fire-detector`** — first run of a freshly-attached schedule should
|
||||
log `first-fire-detector PASS (first fire)` with `lastTimestamp: null`.
|
||||
Subsequent runs log `PASS (Nth fire)` with a Date.
|
||||
- **`interval-validator`** — every non-first fire of an every-minute task
|
||||
should log `interval-validator PASS` with `actualIntervalMs: 60000`. A
|
||||
`FAIL` here means the worker payload isn't carrying the previous fire time
|
||||
correctly.
|
||||
- **`upcoming-validator`** — every fire should log `upcoming-validator PASS`
|
||||
with 10 strictly-increasing slots, each 60s apart.
|
||||
- **`every-minute`**, **`every-five-minutes`**, **`hourly-utc`** — sanity
|
||||
checks across cadences. Inspect `payload` in the dashboard to confirm
|
||||
`timestamp` and `lastTimestamp` look right.
|
||||
- **`daily-*`** schedules — won't fire during a short dev session, but the
|
||||
attach action should enqueue a Redis job for the next slot in the listed
|
||||
timezone. Worth checking that next-fire time matches the expected wall-clock
|
||||
in that tz.
|
||||
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"name": "references-scheduled-tasks",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"dev": "trigger dev",
|
||||
"deploy": "trigger deploy"
|
||||
},
|
||||
"dependencies": {
|
||||
"@trigger.dev/sdk": "workspace:*",
|
||||
"zod": "3.25.76"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@trigger.dev/build": "workspace:*",
|
||||
"trigger.dev": "workspace:*",
|
||||
"typescript": "^5.5.4"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,216 @@
|
||||
import { logger, schedules } from "@trigger.dev/sdk";
|
||||
|
||||
/**
|
||||
* Reference project for E2E-testing the schedule engine.
|
||||
*
|
||||
* The schedule engine carries the previous fire time forward via the worker
|
||||
* queue payload (no DB round-trip). These tasks exercise the customer-visible
|
||||
* surface so we can verify the flow end-to-end:
|
||||
*
|
||||
* - First-ever fire reports `payload.lastTimestamp === undefined`.
|
||||
* - Subsequent fires report `payload.lastTimestamp` equal to the previous
|
||||
* `payload.timestamp` exactly (no cron-derivation drift).
|
||||
* - `payload.upcoming` is a strictly-increasing array of 10 future slots.
|
||||
* - Multiple cron syntaxes and timezones all behave consistently.
|
||||
*
|
||||
* Validators (every-minute, interval, upcoming) emit explicit PASS / FAIL
|
||||
* log lines so you can grep `trigger dev` output for regressions.
|
||||
*/
|
||||
|
||||
// --- Basic recurring tasks ----------------------------------------------------
|
||||
|
||||
export const everyMinute = schedules.task({
|
||||
id: "every-minute",
|
||||
cron: "* * * * *",
|
||||
run: async (payload) => {
|
||||
logger.info("every-minute fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
upcomingCount: payload.upcoming.length,
|
||||
timezone: payload.timezone,
|
||||
scheduleId: payload.scheduleId,
|
||||
});
|
||||
|
||||
return {
|
||||
timestamp: payload.timestamp,
|
||||
lastTimestamp: payload.lastTimestamp,
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
export const everyFiveMinutes = schedules.task({
|
||||
id: "every-five-minutes",
|
||||
cron: "*/5 * * * *",
|
||||
run: async (payload) => {
|
||||
logger.info("every-five-minutes fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const hourlyUtc = schedules.task({
|
||||
id: "hourly-utc",
|
||||
cron: "0 * * * *",
|
||||
run: async (payload) => {
|
||||
logger.info("hourly-utc fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
timezone: payload.timezone,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
// --- Timezone coverage --------------------------------------------------------
|
||||
//
|
||||
// These exercise the engine's tz handling. They fire infrequently, so they're
|
||||
// mostly useful for inspecting enqueued jobs in the dashboard rather than for
|
||||
// short dev sessions.
|
||||
|
||||
export const dailyNewYorkMorning = schedules.task({
|
||||
id: "daily-new-york-morning",
|
||||
cron: { pattern: "0 9 * * *", timezone: "America/New_York" },
|
||||
run: async (payload) => {
|
||||
logger.info("daily-new-york-morning fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
timezone: payload.timezone,
|
||||
// For DST-observing tz the cron interpretation may shift twice/year —
|
||||
// worth eyeballing the timestamp lines up with 09:00 NY local.
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const dailyLondonEvening = schedules.task({
|
||||
id: "daily-london-evening",
|
||||
cron: { pattern: "0 18 * * *", timezone: "Europe/London" },
|
||||
run: async (payload) => {
|
||||
logger.info("daily-london-evening fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
timezone: payload.timezone,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const dailyTokyoMidnight = schedules.task({
|
||||
id: "daily-tokyo-midnight",
|
||||
cron: { pattern: "0 0 * * *", timezone: "Asia/Tokyo" },
|
||||
run: async (payload) => {
|
||||
logger.info("daily-tokyo-midnight fired", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
timezone: payload.timezone,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
// --- Validators ---------------------------------------------------------------
|
||||
//
|
||||
// These explicitly check invariants and emit PASS / FAIL log lines. Running
|
||||
// `trigger dev` and watching these for several fires gives us the E2E signal
|
||||
// that the worker-payload flow is correct.
|
||||
|
||||
/**
|
||||
* Verifies that the very first fire reports `lastTimestamp === undefined` and
|
||||
* subsequent fires report a real Date. This is the customer-visible surface
|
||||
* for the "first-run sentinel" pattern: `if (!payload.lastTimestamp) initOnce()`.
|
||||
*/
|
||||
export const firstFireDetector = schedules.task({
|
||||
id: "first-fire-detector",
|
||||
cron: "* * * * *",
|
||||
run: async (payload) => {
|
||||
const isFirstFire = payload.lastTimestamp === undefined;
|
||||
logger.info(
|
||||
isFirstFire ? "first-fire-detector PASS (first fire)" : "first-fire-detector PASS (Nth fire)",
|
||||
{
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp?.toISOString() ?? null,
|
||||
isFirstFire,
|
||||
}
|
||||
);
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* Verifies that for non-first fires, `timestamp - lastTimestamp` equals exactly
|
||||
* the cron interval (60s for every-minute). This is the key invariant of the
|
||||
* workerCatalog approach — the value is carried through the Redis payload, so
|
||||
* it should be exact, not a cron-derived approximation.
|
||||
*/
|
||||
export const intervalValidator = schedules.task({
|
||||
id: "interval-validator",
|
||||
cron: "* * * * *",
|
||||
run: async (payload) => {
|
||||
if (payload.lastTimestamp === undefined) {
|
||||
logger.info("interval-validator skipped (first fire — no lastTimestamp)", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const expectedIntervalMs = 60_000;
|
||||
const actualIntervalMs = payload.timestamp.getTime() - payload.lastTimestamp.getTime();
|
||||
const passed = actualIntervalMs === expectedIntervalMs;
|
||||
|
||||
logger.info(passed ? "interval-validator PASS" : "interval-validator FAIL", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
lastTimestamp: payload.lastTimestamp.toISOString(),
|
||||
expectedIntervalMs,
|
||||
actualIntervalMs,
|
||||
driftMs: actualIntervalMs - expectedIntervalMs,
|
||||
});
|
||||
|
||||
if (!passed) {
|
||||
throw new Error(
|
||||
`interval-validator FAIL: expected ${expectedIntervalMs}ms between fires, got ${actualIntervalMs}ms`
|
||||
);
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* Verifies that `payload.upcoming` is a strictly-increasing array of 10 future
|
||||
* slots, each 60s apart for `* * * * *`.
|
||||
*/
|
||||
export const upcomingValidator = schedules.task({
|
||||
id: "upcoming-validator",
|
||||
cron: "* * * * *",
|
||||
run: async (payload) => {
|
||||
const issues: string[] = [];
|
||||
|
||||
if (payload.upcoming.length !== 10) {
|
||||
issues.push(`expected upcoming.length === 10, got ${payload.upcoming.length}`);
|
||||
}
|
||||
|
||||
for (let i = 0; i < payload.upcoming.length; i++) {
|
||||
const slot = payload.upcoming[i];
|
||||
if (slot.getTime() <= payload.timestamp.getTime()) {
|
||||
issues.push(`upcoming[${i}] (${slot.toISOString()}) is not strictly after timestamp`);
|
||||
}
|
||||
if (i > 0) {
|
||||
const prev = payload.upcoming[i - 1];
|
||||
const gapMs = slot.getTime() - prev.getTime();
|
||||
if (gapMs !== 60_000) {
|
||||
issues.push(
|
||||
`upcoming[${i}] - upcoming[${i - 1}] = ${gapMs}ms, expected 60000ms (every-minute cron)`
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (issues.length === 0) {
|
||||
logger.info("upcoming-validator PASS", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
upcoming: payload.upcoming.map((d) => d.toISOString()),
|
||||
});
|
||||
} else {
|
||||
logger.error("upcoming-validator FAIL", {
|
||||
timestamp: payload.timestamp.toISOString(),
|
||||
issues,
|
||||
upcoming: payload.upcoming.map((d) => d.toISOString()),
|
||||
});
|
||||
throw new Error(`upcoming-validator FAIL: ${issues.join("; ")}`);
|
||||
}
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,21 @@
|
||||
import { defineConfig } from "@trigger.dev/sdk";
|
||||
|
||||
export default defineConfig({
|
||||
project: "proj_qcibamtpklwidqfzzyir",
|
||||
runtime: "node",
|
||||
machine: "small-1x",
|
||||
maxDuration: 60,
|
||||
dirs: ["./src/trigger"],
|
||||
logLevel: "debug",
|
||||
compatibilityFlags: ["run_engine_v2"],
|
||||
retries: {
|
||||
enabledInDev: true,
|
||||
default: {
|
||||
maxAttempts: 3,
|
||||
minTimeoutInMs: 1000,
|
||||
maxTimeoutInMs: 10000,
|
||||
factor: 2,
|
||||
randomize: true,
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,14 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2023",
|
||||
"module": "Node16",
|
||||
"moduleResolution": "Node16",
|
||||
"esModuleInterop": true,
|
||||
"strict": true,
|
||||
"skipLibCheck": true,
|
||||
"customConditions": ["@triggerdotdev/source"],
|
||||
"lib": ["DOM", "DOM.Iterable"],
|
||||
"noEmit": true
|
||||
},
|
||||
"include": ["./src/**/*.ts", "trigger.config.ts"]
|
||||
}
|
||||
Reference in New Issue
Block a user