Improved schedule engine #2174
Improved schedule engine #2174
Conversation
…ter that out in the lists
|
WalkthroughThis update introduces a comprehensive scheduling engine refactor and migration across the codebase. A new internal package, ✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
🔭 Outside diff range comments (5)
internal-packages/run-engine/src/engine/types.ts (1)
128-137: 💡 Verification agent🧩 Analysis chain
Temporal field proliferation – clarify semantics of
createdAtvsqueuedAt/queueTimestamp
TriggerParamsnow exposes three very similar timestamps:
createdAtqueuedAtqueueTimestampWithout a clear contract, callers may set only one of them or confuse their meaning, causing subtle ordering bugs in metrics or deduplication.
Suggest either:
- createdAt?: Date; + /** + * Absolute creation time of the *task run* record in the DB. + * Prefer `queuedAt` for queue-ing semantics and `queueTimestamp` for + * external-system hand-off. Documented here to avoid ambiguity. + */ + createdAt?: Date;and add JSDoc for the other two, or fold them into a single canonical field.
Run this to locate all writes to these properties:
🏁 Script executed:
#!/bin/bash rg -n --pretty $'(createdAt|queuedAt|queueTimestamp)\\s*:' | headLength of output: 296
🏁 Script executed:
#!/bin/bash set -euo pipefail # Check for queuedAt and queueTimestamp in TriggerParams type rg -n "queuedAt" internal-packages/run-engine/src/engine/types.ts || echo "no matches for queuedAt" rg -n "queueTimestamp" internal-packages/run-engine/src/engine/types.ts || echo "no matches for queueTimestamp"Length of output: 209
Enhance JSDoc for Temporal Fields in TriggerParams
The three timestamp fields in
TriggerParams—createdAt,queuedAt, andqueueTimestamp—currently lack documentation and have subtly different meanings. Please add clear JSDoc comments to each to prevent misuse:• File:
internal-packages/run-engine/src/engine/types.ts
Lines: ~108–137Suggested diff:
interface TriggerParams { + /** + * Absolute creation time of the task-run record in the database. + * Use only for DB-side ordering; not for queue sequencing. + */ createdAt?: Date; + /** + * Time when this run was enqueued in our internal scheduler. + * Use for internal queue ordering and deduplication. + */ queuedAt?: Date; + /** + * Time when the run was handed off to an external system or broker. + * Use for external-system metrics and hand-off tracing. + */ queueTimestamp?: Date; machine?: MachinePresetName; workerId?: string; runnerId?: string; releaseConcurrency?: boolean; runChainState?: RunChainState; scheduleId?: string; scheduleInstanceId?: string; }apps/webapp/app/v3/services/triggerTaskV1.server.ts (1)
367-372: 🛠️ Refactor suggestionSame validation gap for
options.queueTimestampDuplicate the sanity check here (or extract a shared util) to avoid divergent behaviours between V1 and the new engine path.
internal-packages/schedule-engine/tsconfig.test.json (1)
1-22:⚠️ Potential issue
includemisses the actual test folderIntegration tests live in
test/, but"include": ["src/**/*.test.ts"]only covers unit tests insidesrc/.
vitestwill still run the file, yet the compiler won’t type-check it, defeating the purpose of this config.- "include": ["src/**/*.test.ts"], + "include": ["src/**/*.test.ts", "test/**/*.ts"],Add the glob (or move the tests) to ensure they’re compiled.
apps/webapp/app/v3/services/timeoutDeployment.server.ts (1)
46-63: 🛠️ Refactor suggestion
txparameter is now unused – drop or use
enqueue()still acceptstx?: PrismaClientOrTransactionbut never references it after the refactor tocommonWorker.-static async enqueue( - deploymentId: string, - fromStatus: string, - errorMessage: string, - runAt: Date, - tx?: PrismaClientOrTransaction -) { +static async enqueue( + deploymentId: string, + fromStatus: string, + errorMessage: string, + runAt: Date, +) {Keeping an unused parameter risks confusion and lint warnings. If transactional enqueue is still required, wire it through
commonWorker.enqueue; otherwise remove it.apps/webapp/app/v3/services/cancelDevSessionRuns.server.ts (1)
93-106:⚠️ Potential issueBreaking change: parameter order flipped
enqueue()used to be(options, tx, runAt?).
The new signature(options, runAt?, tx?)silently swapsrunAtandtx. Existing call-sites will now pass a Prisma client where aDateis expected, resulting in an invalidavailableAtand no transactional context.Either restore the original ordering or add an overload while deprecating the old one:
-static async enqueue( - options: CancelDevSessionRunsServiceOptions, - runAt?: Date, - tx?: PrismaClientOrTransaction -) { +// Preserve old order +static async enqueue( + options: CancelDevSessionRunsServiceOptions, + tx?: PrismaClientOrTransaction, + runAt?: Date, +) {Remember to prefix
_txif it remains unused.
♻️ Duplicate comments (3)
apps/webapp/app/presenters/v3/RunListPresenter.server.ts (1)
182-186: SameclampToNowduplication as in NextRunListPresenterPlease pull this helper from a shared util to keep behaviour consistent across all call-sites and cut maintenance overhead.
internal-packages/schedule-engine/tsconfig.build.json (1)
6-8: Same DOM lib concern as in tsconfig.src.jsonConsider trimming the
libarray here as well to keep the build output consistent with the runtime environment.apps/webapp/app/v3/services/triggerTaskV1.server.ts (1)
442-443: Conditionally includecreatedAtApply the same conditional spread pattern here to avoid sending explicit
undefinedto Prisma.
🧹 Nitpick comments (31)
apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts (2)
192-195: Utility duplication – extractclampToNow
clampToNowis now copy-pasted in at least two presenters. Move it to a small shared util (e.g.~/utils/date.ts) to avoid divergence and ease testing.-function clampToNow(date: Date): Date { - const now = new Date(); - return date > now ? now : date; -} +import { clampToNow } from "~/utils/date";
206-208: Minor: avoid doubleDate→number→Dateconversions
time.from/time.toareDateobjects. You callclampToNow(time.to)(returnsDate) and then immediately.getTime(). Consider inlining to keep it obvious:- to: time.to ? clampToNow(time.to).getTime() : undefined, + to: time.to ? Math.min(time.to.getTime(), Date.now()) : undefined,Purely cosmetic; feel free to skip.
apps/webapp/app/presenters/v3/RunListPresenter.server.ts (1)
291-294: Performance & readability: calculate once
clampToNow(time.to)is evaluated during SQL template construction; for large batches this inside-template computation is negligible, but calculating once improves clarity:- time.to - ? Prisma.sql`AND tr."createdAt" <= ${clampToNow(time.to).toISOString()}::timestamp` + time.to + ? Prisma.sql`AND tr."createdAt" <= ${ + clampToNow(time.to).toISOString() + }::timestamp`Even better after extracting util as suggested above.
internal-packages/schedule-engine/test/setup.ts (1)
1-4: Consider moving global vitest config tovitest.config.*instead of runtime mutation
vi.setConfig({ testTimeout: 60_000 });works, but:
- It executes after test file import order, which can be flaky if other files rely on the timeout earlier.
- A
vitest.config.ts(or<root>/vitest.setup.tsreferenced viasetupFiles) is the canonical place.If you keep this, at least prefix the filename with
setupTests.to make its intent clearer.apps/webapp/app/v3/services/triggerTask.server.ts (1)
34-36: Consider surfacing docs / validation for the new options
queueTimestampandoverrideCreatedAtare accepted here, but nothing verifies that:
queueTimestamp≤overrideCreatedAt(or vice-versa depending on intent)- both are sensible (e.g. not in the future or > now + max skew)
A short JSDoc comment or runtime guard would prevent subtle clock-skew bugs.
apps/webapp/app/v3/services/createBackgroundWorker.server.ts (1)
28-29: Import is used only for side-effectful calls – keep tree-shaking friendly
scheduleEngineis imported at module top level but only referenced insidesyncDeclarativeSchedules.
For clarity / bundler friendliness you could move the import next to its usage:-import { scheduleEngine } from "../scheduleEngine.server"; +// defer import to keep the main bundle lean +import { scheduleEngine } from "../scheduleEngine.server";(Not critical; feel free to ignore if build size is not a concern.)
internal-packages/schedule-engine/tsconfig.json (1)
1-8: Minor: missing"strict": true&rootDircan hurt DXSince this is a new package, enabling compiler strictness early saves bugs later:
"compilerOptions": { + "strict": true, "moduleResolution": "Node16", "module": "Node16", "customConditions": ["@triggerdotdev/source"] }Also add
"rootDir": "./src"to keep emitted paths tidy.internal-packages/schedule-engine/tsconfig.src.json (1)
6-8: Unnecessary DOM libs increase type-check noiseThe library list includes
DOM*libs even though the package runs exclusively in Node. This can:
- pull in thousands of irrelevant types
- mask accidental browser-only API usage
- "lib": ["ES2020", "DOM", "DOM.Iterable", "DOM.AsyncIterable"], + "lib": ["ES2020"],Unless you intentionally reference browser globals inside the engine, dropping them will speed up builds and tighten type safety.
apps/webapp/app/runEngine/services/triggerTask.server.ts (1)
311-312: PassingcreatedAtasundefinedclutters the Prisma payload
createdAtis always present in the object, even when no override is supplied. Prisma acceptsundefined, but it still bloats query text and hides intent.Consider making the property conditional:
- createdAt: options.overrideCreatedAt, + ...(options.overrideCreatedAt + ? { createdAt: options.overrideCreatedAt } + : {}),Keeps the payload minimal and future-proof if Prisma tightens strict-null-checks.
internal-packages/schedule-engine/src/engine/workerCatalog.ts (1)
3-14: Explicit typing &as constwould safeguard the catalog
scheduleWorkerCatalogis currently an untyped object literal. Down-stream code will lose IntelliSense and accidental property changes won’t be caught.-export const scheduleWorkerCatalog = { +import type { WorkerCatalog } from "@trigger.dev/worker"; // hypothetical shared type + +export const scheduleWorkerCatalog: WorkerCatalog = { "schedule.triggerScheduledTask": { schema: z.object({ instanceId: z.string(), exactScheduleTime: z.coerce.date(), }),Optionally append
as constafter the object to freeze the keys.This minor change prevents typos (e.g.,
"schedule.triggerSchedueldTask") from compiling.apps/webapp/app/services/email.server.ts (1)
95-101: Return the enqueue promise to surface failures
scheduleEmailcurrently awaits the enqueue but returnsvoid, swallowing any value the worker library might provide (job id, etc.). Returning the promise enables callers to react or log the job id.-export async function scheduleEmail(data: DeliverEmail, delay?: { seconds: number }) { +export async function scheduleEmail( + data: DeliverEmail, + delay?: { seconds: number } +): Promise<void> { const availableAt = delay ? new Date(Date.now() + delay.seconds * 1000) : undefined; - await commonWorker.enqueue({ + return commonWorker.enqueue({ job: "scheduleEmail", payload: data, availableAt, }); }Also consider validating
delay.seconds >= 0to avoid accidental past dates.internal-packages/schedule-engine/src/engine/distributedScheduling.ts (1)
21-26: Small clarity nit
Math.pow(2, 31)can be replaced with the more idiomatic2 ** 31, which is easier to read and avoids an extra function call.apps/webapp/app/v3/commonWorker.server.ts (1)
48-141: Catalog definition is getting unwieldyHard-coding dozens of job specs with identical
visibilityTimeoutMsand very similar retry policies makes maintenance painful. Consider extracting helpers:const defaultJob = <T>(schema: T, retry = 5) => ({ schema, visibilityTimeoutMs: 60_000, retry: { maxAttempts: retry }, });Then:
catalog: { "v3.resumeBatchRun": defaultJob(z.object({ batchRunId: z.string() })), ... }This trims ~150 LOC and reduces copy-paste errors.
internal-packages/schedule-engine/README.md (2)
51-58: Minor wording & brevity nit“from executing at exactly the same moment”
-Prevents thundering herd issues by distributing executions across time windows +Prevents thundering-herd issues by spreading executions across a time window🧰 Tools
🪛 LanguageTool
[style] ~51-~51: ‘exactly the same’ might be wordy. Consider a shorter alternative.
Context: ...t all scheduled tasks from executing at exactly the same moment: ```typescript import { calcula...(EN_WORDINESS_PREMIUM_EXACTLY_THE_SAME)
101-105: Dangling colon renders as loose punctuationA colon at the start of a bullet occasionally shows up in some markdown renderers:
-- `redis`: Redis connection configuration ++ `redis` – Redis connection configuration🧰 Tools
🪛 LanguageTool
[uncategorized] ~101-~101: Loose punctuation mark.
Context: ...these configuration options: -prisma: PrismaClient instance -redis: Redis ...(UNLIKELY_OPENING_PUNCTUATION)
apps/webapp/app/v3/services/timeoutDeployment.server.ts (1)
53-62: ConsiderenqueueOnceto avoid duplicate timeout jobsTimeouts are idempotent; using plain
enqueuecan queue duplicates ifenqueueis called multiple times before the first job is processed.- await commonWorker.enqueue({ + await commonWorker.enqueueOnce({ id: `timeoutDeployment:${deploymentId}`,The unique
idis already provided, so switching toenqueueOnceprotects against accidental duplication at no extra cost.apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts (1)
107-116: UseenqueueOncefor id-based deduplicationAs with the timeout service, the job ID is deterministic (
v3.executeTasksWaitingForDeploy:${backgroundWorkerId}).
Switching toenqueueOnceguarantees exactly-once semantics during bursts (e.g., multiple deploy hooks firing).- return await commonWorker.enqueue({ + return await commonWorker.enqueueOnce({apps/webapp/app/v3/services/cancelTaskAttemptDependencies.server.ts (1)
60-69: Unusedtxparameter – silence linter or drop itThe
txargument is no longer forwarded tocommonWorker.enqueue, leaving it unused and triggering the@typescript-eslint/no-unused-varsrule in most code-bases.-static async enqueue(attemptId: string, tx: PrismaClientOrTransaction, runAt?: Date) { +static async enqueue( + attemptId: string, + _tx: PrismaClientOrTransaction, // keep positional compatibility but silence linter + runAt?: Date +) {If transactions are no longer required here, consider making the parameter optional and prefixing with an underscore (or removing it entirely) across call-sites to avoid confusion.
apps/webapp/app/runEngine/services/batchTrigger.server.ts (1)
316-321:txparameter kept but never used
#enqueueBatchTaskRun()still accepts atxparameter yet the value is ignored after the switch tocommonWorker.enqueue. Either remove the parameter or rename it to_txto clarify that it is intentionally unused:-async #enqueueBatchTaskRun(options: BatchProcessingOptions, tx?: PrismaClientOrTransaction) { +async #enqueueBatchTaskRun(options: BatchProcessingOptions, _tx?: PrismaClientOrTransaction) {Cleaning this up avoids false expectations that the call will be wrapped in a transaction.
apps/webapp/app/v3/services/resumeTaskDependency.server.ts (1)
157-171: UnusedtxargumentSimilar to other services,
txis accepted but not used after the move tocommonWorker.enqueue.-static async enqueue( - dependencyId: string, - sourceTaskAttemptId: string, - tx: PrismaClientOrTransaction, - runAt?: Date -) { +static async enqueue( + dependencyId: string, + sourceTaskAttemptId: string, + _tx: PrismaClientOrTransaction, + runAt?: Date +) {This prevents linter warnings and signals that transactional support has been removed.
apps/webapp/test/distributedScheduling.test.ts (1)
57-88: Potential flakiness in uniform-distribution assertionThe “should distribute work evenly” test uses 1 000 samples and fails if fewer than 80 % of 30 buckets receive at least one hit.
While unlikely, the deterministic hash could still cluster enough samples to dip below the 24-bucket threshold, making the test flaky.Consider loosening the assertion or increasing the sample size:
-const testCount = 1000; -… -expect(bucketsWithItems).toBeGreaterThan(distributionWindow * 0.8); +const testCount = 5_000; // reduces variance +… +expect(bucketsWithItems).toBeGreaterThan(distributionWindow * 0.7);This keeps the spirit of the check while minimising spurious CI failures.
apps/webapp/app/v3/services/resumeBatchRun.server.ts (1)
2-2: Unused import side-effectSwitching to
commonWorkermakes thePrismaClientOrTransactionargument inenqueue()obsolete. Consider pruning that parameter from the method signature (and its call-sites) to avoid dead code and confusion.apps/webapp/app/v3/scheduleEngine.server.ts (1)
81-116: Error is swallowed – add structured loggingThe
catchonly returns{ success:false }but drops stack traces; production debugging will be painful.- } catch (error) { - return { - success: false, - error: error instanceof Error ? error.message : String(error), - }; + } catch (error) { + logger.error("scheduleEngine.onTriggerScheduledTask failed", { error }); + return { + success: false, + error: error instanceof Error ? error.message : String(error), + }; }apps/webapp/app/v3/services/batchTriggerV3.server.ts (2)
478-540: Inconsistent logging & IDs reference “V2”This v3 service logs under
BatchTriggerV2and builds job IDs withBatchTriggerV2Service.process.
Rename toBatchTriggerV3to avoid future grep/debug headaches.- id: `BatchTriggerV2Service.process:${options.batchId}:${options.processingId}`, + id: `BatchTriggerV3Service.process:${options.batchId}:${options.processingId}`,Same for every logger.debug message using
[BatchTriggerV2] ….
895-899:txparameter is never used
#enqueueBatchTaskRunstill acceptstxbut ignores it. Drop the argument or forward it to a transactional queue helper; leaving it dangling is misleading.internal-packages/schedule-engine/test/scheduleEngine.test.ts (1)
121-123: Avoid brittle exact-equality on Date
expect(actualNextExecution).toEqual(expectedExecutionTime)can flake by a few ms.
Use a ±1 s window ortoBeCloseTowith a tolerance instead.internal-packages/schedule-engine/src/engine/types.ts (1)
27-30: ConstrainlogLevelvia a union
logLevel?: stringaccepts invalid values at runtime. ConsiderlogLevel?: "log" | "error" | "warn" | "info" | "debug";This mirrors the enum you use elsewhere and prevents typos.
apps/webapp/app/v3/services/upsertTaskSchedule.server.ts (2)
108-128: Parallelise instance creation / registrationThe loop issues a Prisma insert then immediately waits for
scheduleEngine.registerNextTaskScheduleInstance.
With many environments this becomes O(n) sequential I/O. Consider:await Promise.all( options.environments.map(async (environmentId) => { const instance = await this._prisma.taskScheduleInstance.create({ … }); await scheduleEngine.registerNextTaskScheduleInstance({ instanceId: instance.id }); }) );This keeps semantic order while cutting latency substantially.
206-218: Typo & readability
instancesToDeleted➜instancesToDelete. Minor, but avoiding grammatical slips helps future readers.internal-packages/schedule-engine/src/engine/index.ts (1)
96-106: Side-effects in constructor hinder testabilitySpawning the worker inside the constructor (
this.worker.start()) means merely importing / instantiating the class connects to Redis and spawns threads, complicating unit tests and life-cycle management.
Expose an explicitstart()method instead; call it from your singleton wrapper.apps/webapp/app/env.server.ts (1)
686-695: Clarify overlapping concurrency env-varsThe trio
SCHEDULE_WORKER_CONCURRENCY_WORKERS,
SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER, and
SCHEDULE_WORKER_CONCURRENCY_LIMIT
overlap but the codebase only readsconcurrency(workers × tasks). Document which flags are authoritative or drop the unused ones to avoid misconfiguration.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (2)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yamlreferences/hello-world/src/trigger/schedule.tsis excluded by!references/**
📒 Files selected for processing (49)
apps/webapp/app/env.server.ts(1 hunks)apps/webapp/app/presenters/v3/NextRunListPresenter.server.ts(2 hunks)apps/webapp/app/presenters/v3/RunListPresenter.server.ts(2 hunks)apps/webapp/app/runEngine/services/batchTrigger.server.ts(2 hunks)apps/webapp/app/runEngine/services/triggerTask.server.ts(1 hunks)apps/webapp/app/services/email.server.ts(2 hunks)apps/webapp/app/services/runsRepository.server.ts(1 hunks)apps/webapp/app/services/worker.server.ts(19 hunks)apps/webapp/app/v3/commonWorker.server.ts(3 hunks)apps/webapp/app/v3/scheduleEngine.server.ts(1 hunks)apps/webapp/app/v3/services/batchTriggerV3.server.ts(3 hunks)apps/webapp/app/v3/services/cancelDevSessionRuns.server.ts(2 hunks)apps/webapp/app/v3/services/cancelTaskAttemptDependencies.server.ts(2 hunks)apps/webapp/app/v3/services/changeCurrentDeployment.server.ts(1 hunks)apps/webapp/app/v3/services/createBackgroundWorker.server.ts(3 hunks)apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV3.server.ts(1 hunks)apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts(2 hunks)apps/webapp/app/v3/services/registerNextTaskScheduleInstance.server.ts(0 hunks)apps/webapp/app/v3/services/resumeBatchRun.server.ts(2 hunks)apps/webapp/app/v3/services/resumeTaskDependency.server.ts(2 hunks)apps/webapp/app/v3/services/resumeTaskRunDependencies.server.ts(2 hunks)apps/webapp/app/v3/services/retryAttempt.server.ts(2 hunks)apps/webapp/app/v3/services/timeoutDeployment.server.ts(2 hunks)apps/webapp/app/v3/services/triggerScheduledTask.server.ts(0 hunks)apps/webapp/app/v3/services/triggerTask.server.ts(1 hunks)apps/webapp/app/v3/services/triggerTaskV1.server.ts(2 hunks)apps/webapp/app/v3/services/upsertTaskSchedule.server.ts(4 hunks)apps/webapp/app/v3/utils/distributedScheduling.server.ts(1 hunks)apps/webapp/package.json(1 hunks)apps/webapp/test/distributedScheduling.test.ts(1 hunks)internal-packages/run-engine/src/engine/index.ts(2 hunks)internal-packages/run-engine/src/engine/types.ts(1 hunks)internal-packages/schedule-engine/README.md(1 hunks)internal-packages/schedule-engine/package.json(1 hunks)internal-packages/schedule-engine/src/engine/distributedScheduling.ts(1 hunks)internal-packages/schedule-engine/src/engine/index.ts(1 hunks)internal-packages/schedule-engine/src/engine/scheduleCalculation.ts(1 hunks)internal-packages/schedule-engine/src/engine/types.ts(1 hunks)internal-packages/schedule-engine/src/engine/workerCatalog.ts(1 hunks)internal-packages/schedule-engine/src/index.ts(1 hunks)internal-packages/schedule-engine/test/scheduleEngine.test.ts(1 hunks)internal-packages/schedule-engine/test/setup.ts(1 hunks)internal-packages/schedule-engine/tsconfig.build.json(1 hunks)internal-packages/schedule-engine/tsconfig.json(1 hunks)internal-packages/schedule-engine/tsconfig.src.json(1 hunks)internal-packages/schedule-engine/tsconfig.test.json(1 hunks)internal-packages/schedule-engine/vitest.config.ts(1 hunks)packages/redis-worker/src/worker.ts(1 hunks)packages/rsc/src/package.json(1 hunks)
💤 Files with no reviewable changes (2)
- apps/webapp/app/v3/services/registerNextTaskScheduleInstance.server.ts
- apps/webapp/app/v3/services/triggerScheduledTask.server.ts
🧰 Additional context used
🧬 Code Graph Analysis (15)
apps/webapp/app/v3/services/changeCurrentDeployment.server.ts (1)
apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts (1)
ExecuteTasksWaitingForDeployService(8-117)
apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV3.server.ts (1)
apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts (1)
ExecuteTasksWaitingForDeployService(8-117)
apps/webapp/app/v3/services/resumeTaskRunDependencies.server.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
apps/webapp/app/services/email.server.ts (2)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)internal-packages/emails/src/index.tsx (1)
data(80-122)
apps/webapp/app/v3/services/executeTasksWaitingForDeploy.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
apps/webapp/app/v3/services/timeoutDeployment.server.ts (3)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)apps/webapp/app/db.server.ts (1)
PrismaClientOrTransaction(21-21)apps/webapp/app/services/worker.server.ts (1)
workerQueue(373-373)
apps/webapp/app/v3/services/cancelTaskAttemptDependencies.server.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
packages/redis-worker/src/worker.ts (2)
internal-packages/schedule-engine/src/engine/index.ts (1)
JobHandlerParams(217-225)internal-packages/zod-worker/src/index.ts (1)
K(581-660)
apps/webapp/app/v3/services/retryAttempt.server.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
internal-packages/schedule-engine/src/engine/types.ts (4)
internal-packages/schedule-engine/src/index.ts (3)
TriggerScheduledTaskCallback(5-5)ScheduleEngineOptions(3-3)TriggerScheduleParams(4-4)apps/webapp/app/db.server.ts (1)
PrismaClient(228-228)internal-packages/redis/src/index.ts (1)
RedisOptions(4-4)internal-packages/tracing/src/index.ts (2)
Tracer(14-14)Meter(16-16)
apps/webapp/app/v3/services/resumeTaskDependency.server.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
apps/webapp/test/distributedScheduling.test.ts (1)
apps/webapp/app/v3/utils/distributedScheduling.server.ts (1)
calculateDistributedExecutionTime(6-31)
apps/webapp/app/v3/services/cancelDevSessionRuns.server.ts (2)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)apps/webapp/app/v3/services/batchTriggerV3.server.ts (1)
options(894-900)
apps/webapp/app/v3/services/resumeBatchRun.server.ts (1)
apps/webapp/app/v3/commonWorker.server.ts (1)
commonWorker(279-279)
apps/webapp/app/v3/services/createBackgroundWorker.server.ts (1)
apps/webapp/app/v3/scheduleEngine.server.ts (1)
scheduleEngine(11-11)
🪛 LanguageTool
internal-packages/schedule-engine/README.md
[style] ~51-~51: ‘exactly the same’ might be wordy. Consider a shorter alternative.
Context: ...t all scheduled tasks from executing at exactly the same moment: ```typescript import { calcula...
(EN_WORDINESS_PREMIUM_EXACTLY_THE_SAME)
[uncategorized] ~101-~101: Loose punctuation mark.
Context: ...these configuration options: - prisma: PrismaClient instance - redis: Redis ...
(UNLIKELY_OPENING_PUNCTUATION)
⏰ Context from checks skipped due to timeout of 90000ms (25)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (1, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (8, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (10, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (7, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (4, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (6, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (9, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (3, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (5, 10)
- GitHub Check: units / webapp / 🧪 Unit Tests: Webapp (2, 10)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (8, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (7, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (4, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (3, 8)
- GitHub Check: units / packages / 🧪 Unit Tests: Packages (1, 1)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (2, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (1, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (6, 8)
- GitHub Check: units / internal / 🧪 Unit Tests: Internal (5, 8)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (ubuntu-latest - npm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - pnpm)
- GitHub Check: e2e / 🧪 CLI v3 tests (windows-latest - npm)
- GitHub Check: typecheck / typecheck
- GitHub Check: Analyze (javascript-typescript)
🔇 Additional comments (16)
packages/rsc/src/package.json (1)
1-3: Minimalpackage.jsonLGTM, but double-check directory purposeDeclaring
"type": "module"is fine and consistent with the rest of the repo.
Make sure:
- No other
package.jsonhigher in thepackages/rsctree overrides this.- Tooling that consumes this folder (ts-node, vitest, etc.) expects ESM.
If both are already covered, nothing else to do here.
apps/webapp/package.json (1)
53-57: Workspace dependency added – ensure pipeline & lockfile are updated
"@internal/schedule-engine": "workspace:*"looks correct.
Please run the package-manager install (pnpm install/npm install -w) and commit the updated lockfile so CI picks it up.apps/webapp/app/v3/services/changeCurrentDeployment.server.ts (1)
87-87: Call site now matches newenqueuesignature – all good
ExecuteTasksWaitingForDeployService.enqueuenow only accepts(backgroundWorkerId, runAt?).
This update removes the stale Prisma argument and compiles cleanly.apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV3.server.ts (1)
164-166: LGTM – signature change reflected correctly
ExecuteTasksWaitingForDeployService.enqueueno longer requires the Prisma client, and the call site has been updated accordingly.apps/webapp/app/v3/services/createBackgroundWorker.server.ts (1)
571-573: Error propagation unchanged but worth revisiting
scheduleEngine.registerNextTaskScheduleInstanceis awaited inside this tight loop.
If that call throws (e.g. transient Redis issue) the entire worker-creation flow aborts and rolls back the DB transaction – the same behaviour the removed service had, but now the failure surface is broader (Redis, distributed locks…).Consider wrapping in a retry helper or catching and logging non-fatal errors to avoid making deployments brittle.
Also applies to: 603-604
internal-packages/schedule-engine/vitest.config.ts (1)
11-13: Confirm the esbuild target is aligned with runtime
target: "node18"is fine today, but if the runtime matrix includes Node 20 (current LTS) the emitted code will miss out on native improvements (e.g., fetch, structuredClone). Worth double-checking the production containers.internal-packages/schedule-engine/src/index.ts (1)
1-6: Public entry-point looks goodThe minimalist barrel file is clear and follows the Node16/ESM convention that the compiler will emit
.jsfiles, so the explicit.jsextension is appropriate. No further concerns.apps/webapp/app/v3/services/resumeTaskRunDependencies.server.ts (1)
8-9: Import change looks correctSwitching to the new
commonWorkeraligns with the broader refactor.apps/webapp/app/services/email.server.ts (1)
7-8: Import update acknowledgedMigrating to
commonWorkeris consistent with the rest of the PR.internal-packages/schedule-engine/package.json (1)
8-14: Non-standard export key
"@triggerdotdev/source"insideexportsisn’t a recognised condition and may break resolution in Node ≥16. Verify that tools relying on this key actually support it; otherwise use a documented condition such as"source"or"development".apps/webapp/app/v3/commonWorker.server.ts (1)
188-195: Ensure numeric env values
pollIntervalMs,immediatePollIntervalMs, worker counts, etc., are passed straight fromenv.*. Confirm the Zod schema parses them to numbers; if they are still strings, RedisWorker will treat"5000"as a string and mis-handle arithmetic/timeouts.If parsing isn’t already done in
env.server.ts, coerce here:pollIntervalMs: Number(env.COMMON_WORKER_POLL_INTERVAL),apps/webapp/app/services/worker.server.ts (1)
238-246: ConfirmfinalAttemptsemantics against worker implementation
finalAttemptis derived fromjob.attempts === job.max_attempts.
Ingraphile-workertheattemptscounter already includes the current attempt, so the flag will betruewhile the job is still running, not after it has finally failed.
IfscheduleEngine.triggerScheduledTask()internally interpretsfinalAttemptas “no more retries will follow”, this could lead to premature deletion / alerting.await scheduleEngine.triggerScheduledTask({ instanceId: payload.instanceId, finalAttempt: job.attempts === job.max_attempts, // ← verify semantics });Please double-check the contract and adjust to
job.attempts + 1 === job.max_attempts(or similar) if the engine expects the next attempt to be the last.apps/webapp/app/v3/services/timeoutDeployment.server.ts (1)
66-69: Acknowledge both queues symmetrically
dequeueis invoked onworkerQueue, whileackis invoked oncommonWorker.
IfworkerQueue.dequeue()fails (e.g., job not found) thecommonWorker.ack()will still run, silently discarding the job in the new queue. Consider guarding with an existence check or catching errors to keep behaviour predictable during the migration window.packages/redis-worker/src/worker.ts (1)
37-48: Type extraction looks goodExtracting
JobHandlerParamsmakes handler definitions clearer and reusable; compile-time inference also improves.apps/webapp/app/v3/services/resumeBatchRun.server.ts (1)
337-344: VerifyavailableAtsemantics
commonWorker.enqueueoften expects a Unix epoch (ms) whereas you’re passing aDate | undefined.
Double-check the worker’s API – if it expects a number, wrap with.getTime(); if it acceptsDate, ignore this note.internal-packages/schedule-engine/src/engine/scheduleCalculation.ts (1)
22-30: LGTM – cron calculation is concise and correct
| testTimeout: 30000, | ||
| hookTimeout: 30000, | ||
| }, |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Avoid conflicting time-out definitions
testTimeout/hookTimeout are set to 30 s here, but ./test/setup.ts later overrides the default timeout to 60 s for container-heavy tests. The lower value configured here will win and may randomly fail the longer integration suite.
- testTimeout: 30000,
- hookTimeout: 30000,
+ // Allow the 60 s override in ./test/setup.ts to take effect
+ // testTimeout: 60000,
+ // hookTimeout: 60000,Either bump these to 60 s or drop them entirely so the per-test override is the single source of truth.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| testTimeout: 30000, | |
| hookTimeout: 30000, | |
| }, | |
| // Allow the 60 s override in ./test/setup.ts to take effect | |
| // testTimeout: 60000, | |
| // hookTimeout: 60000, | |
| }, |
🤖 Prompt for AI Agents
In internal-packages/schedule-engine/vitest.config.ts around lines 8 to 10, the
testTimeout and hookTimeout are set to 30000 ms (30 s), which conflicts with the
60000 ms (60 s) timeout set later in ./test/setup.ts for container-heavy tests.
To fix this, either increase both testTimeout and hookTimeout here to 60000 ms
to match the later override or remove these timeout settings entirely so that
the per-test override in setup.ts is the single source of truth.
| options.queueTimestamp ?? | ||
| (parentRun && body.options?.resumeParentOnCompletion | ||
| ? parentRun.queueTimestamp ?? undefined | ||
| : undefined, | ||
| : undefined), | ||
| runChainState, |
There was a problem hiding this comment.
options.queueTimestamp needs validation
The new override is accepted blindly. If a caller passes an invalid date (string, future > now + years, etc.) it will persist to the DB and break chronological queries & metrics.
Add a narrow guard before assignment:
- queueTimestamp:
- options.queueTimestamp ??
+ queueTimestamp:
+ (options.queueTimestamp instanceof Date
+ ? options.queueTimestamp
+ : undefined) ??
(parentRun && body.options?.resumeParentOnCompletion
? parentRun.queueTimestamp ?? undefined
: undefined),or run it through a zod schema shared with the client.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| options.queueTimestamp ?? | |
| (parentRun && body.options?.resumeParentOnCompletion | |
| ? parentRun.queueTimestamp ?? undefined | |
| : undefined, | |
| : undefined), | |
| runChainState, | |
| queueTimestamp: | |
| (options.queueTimestamp instanceof Date | |
| ? options.queueTimestamp | |
| : undefined) ?? | |
| (parentRun && body.options?.resumeParentOnCompletion | |
| ? parentRun.queueTimestamp ?? undefined | |
| : undefined), | |
| runChainState, |
🤖 Prompt for AI Agents
In apps/webapp/app/runEngine/services/triggerTask.server.ts around lines 304 to
308, the code assigns options.queueTimestamp without validation, risking invalid
dates being saved and breaking chronological logic. Add a validation step before
assignment to ensure options.queueTimestamp is a valid date and within an
acceptable range (not a future date far beyond now). You can implement this by
checking the type and range explicitly or by validating it through a shared zod
schema with the client, then only assign if valid; otherwise, fallback to the
existing logic.
| schema: z.object({ | ||
| instanceId: z.string(), | ||
| exactScheduleTime: z.coerce.date(), | ||
| }), |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Potential silent date parsing failures
z.coerce.date() will accept almost anything that new Date(value) doesn’t throw on, including ambiguous or invalid ISO strings that silently become "Invalid Date". Consider a stricter guard:
exactScheduleTime: z
.string()
.datetime({ offset: true }) // stricter ISO-8601 with timezone
.transform((v) => new Date(v)),to fail fast on malformed payloads.
🤖 Prompt for AI Agents
In internal-packages/schedule-engine/src/engine/workerCatalog.ts around lines 5
to 8, replace the use of z.coerce.date() for exactScheduleTime with a stricter
validation by first validating the value as a string with z.string().datetime({
offset: true }) to enforce ISO-8601 format with timezone, then transform it to a
Date object. This change ensures that invalid or ambiguous date strings cause
validation to fail immediately instead of silently producing invalid Date
objects.
| static async enqueue(runId: string, tx: PrismaClientOrTransaction, runAt?: Date) { | ||
| return await workerQueue.enqueue( | ||
| "v3.retryAttempt", | ||
| { | ||
| return await commonWorker.enqueue({ | ||
| id: `retryAttempt:${runId}`, | ||
| job: "v3.retryAttempt", | ||
| payload: { | ||
| runId, | ||
| }, | ||
| { | ||
| tx, | ||
| runAt, | ||
| jobKey: `retryAttempt:${runId}`, | ||
| } | ||
| ); | ||
| availableAt: runAt, | ||
| }); |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Unused tx parameter
enqueue keeps the tx: PrismaClientOrTransaction argument but never uses it after the worker refactor, which will raise linter/TS warnings and confuses callers.
- static async enqueue(runId: string, tx: PrismaClientOrTransaction, runAt?: Date) {
+ static async enqueue(runId: string, runAt?: Date) {Update call-sites accordingly.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| static async enqueue(runId: string, tx: PrismaClientOrTransaction, runAt?: Date) { | |
| return await workerQueue.enqueue( | |
| "v3.retryAttempt", | |
| { | |
| return await commonWorker.enqueue({ | |
| id: `retryAttempt:${runId}`, | |
| job: "v3.retryAttempt", | |
| payload: { | |
| runId, | |
| }, | |
| { | |
| tx, | |
| runAt, | |
| jobKey: `retryAttempt:${runId}`, | |
| } | |
| ); | |
| availableAt: runAt, | |
| }); | |
| static async enqueue(runId: string, runAt?: Date) { | |
| return await commonWorker.enqueue({ | |
| id: `retryAttempt:${runId}`, | |
| job: "v3.retryAttempt", | |
| payload: { | |
| runId, | |
| }, | |
| availableAt: runAt, | |
| }); |
🤖 Prompt for AI Agents
In apps/webapp/app/v3/services/retryAttempt.server.ts between lines 26 and 34,
the enqueue method has an unused parameter tx of type PrismaClientOrTransaction
that is no longer needed after the worker refactor. Remove the tx parameter from
the method signature and update all call-sites to stop passing this argument to
avoid linter and TypeScript warnings and reduce confusion.
| const engine = new ScheduleEngine({ | ||
| prisma, | ||
| logLevel: env.SCHEDULE_ENGINE_LOG_LEVEL, | ||
| redis: { | ||
| host: env.SCHEDULE_WORKER_REDIS_HOST ?? "localhost", | ||
| port: env.SCHEDULE_WORKER_REDIS_PORT ?? 6379, | ||
| username: env.SCHEDULE_WORKER_REDIS_USERNAME, | ||
| password: env.SCHEDULE_WORKER_REDIS_PASSWORD, | ||
| keyPrefix: "schedule:", | ||
| enableAutoPipelining: true, | ||
| ...(env.SCHEDULE_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }), | ||
| }, | ||
| worker: { | ||
| concurrency: env.SCHEDULE_WORKER_CONCURRENCY_LIMIT, | ||
| pollIntervalMs: env.SCHEDULE_WORKER_POLL_INTERVAL, | ||
| shutdownTimeoutMs: env.SCHEDULE_WORKER_SHUTDOWN_TIMEOUT_MS, | ||
| disabled: env.SCHEDULE_WORKER_ENABLED === "0", | ||
| }, |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Coerce env values to numbers
env.SCHEDULE_WORKER_REDIS_PORT, SCHEDULE_WORKER_CONCURRENCY_LIMIT, etc. come from process.env and will be strings.
Pass them through Number() (or zod coerce.number()) before feeding them to redis / worker configs to prevent silent mis-configuration.
🤖 Prompt for AI Agents
In apps/webapp/app/v3/scheduleEngine.server.ts around lines 50 to 67,
environment variables like SCHEDULE_WORKER_REDIS_PORT and
SCHEDULE_WORKER_CONCURRENCY_LIMIT are used directly but they are strings from
process.env. Convert these string values to numbers using Number() before
assigning them to the redis and worker configuration properties to ensure
correct types and prevent misconfiguration.
| environment: any; // The full environment object from the database with all relationships | ||
| payload: { |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Avoid any – model the Environment shape
environment: any defeats the purpose of the strongly-typed contracts you’re introducing. Define a EnvironmentPayload interface (even if partial) or use unknown + explicit narrowing at call-sites.
🤖 Prompt for AI Agents
In internal-packages/schedule-engine/src/engine/types.ts at lines 8 to 9,
replace the type of the environment property from any to a properly defined
interface named EnvironmentPayload that models the expected shape of the
environment object. Define this interface with the relevant fields you expect
from the database, or alternatively use unknown and perform explicit type checks
where environment is used. This will enforce strong typing and improve type
safety.
| this.tracer = options.tracer ?? (startSpan as any).tracer; | ||
| this.meter = options.meter ?? getMeter("schedule-engine"); | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Unsafe tracer fallback
Casting startSpan to any to pluck .tracer bypasses type-safety and will break if the util implementation changes.
Prefer passing an explicit getTracer() helper or making tracer a required option in production builds.
🤖 Prompt for AI Agents
In internal-packages/schedule-engine/src/engine/index.ts around lines 42 to 44,
avoid using an unsafe fallback by casting startSpan to any to access .tracer, as
this bypasses type safety and risks breakage if the implementation changes.
Instead, refactor the code to require an explicit tracer option or provide a
dedicated getTracer() helper function to obtain the tracer, ensuring type safety
and reliability in production builds.
| try { | ||
| await this.worker.enqueue({ | ||
| id: `scheduled-task-instance:${instanceId}`, | ||
| job: "schedule.triggerScheduledTask", | ||
| payload: { | ||
| instanceId, | ||
| exactScheduleTime, | ||
| }, | ||
| availableAt: distributedExecutionTime, | ||
| }); |
There was a problem hiding this comment.
Job-ID collision risk
id: \scheduled-task-instance:${instanceId}`is reused for every run of the same schedule instance. If the previous run’s job hasn’t been pruned yet,enqueue` may reject or overwrite, silently dropping executions.
Include the scheduled timestamp (or an incrementing counter) in the ID:
- id: `scheduled-task-instance:${instanceId}`,
+ id: `scheduled-task-instance:${instanceId}:${exactScheduleTime.getTime()}`,📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try { | |
| await this.worker.enqueue({ | |
| id: `scheduled-task-instance:${instanceId}`, | |
| job: "schedule.triggerScheduledTask", | |
| payload: { | |
| instanceId, | |
| exactScheduleTime, | |
| }, | |
| availableAt: distributedExecutionTime, | |
| }); | |
| try { | |
| await this.worker.enqueue({ | |
| - id: `scheduled-task-instance:${instanceId}`, | |
| + id: `scheduled-task-instance:${instanceId}:${exactScheduleTime.getTime()}`, | |
| job: "schedule.triggerScheduledTask", | |
| payload: { | |
| instanceId, | |
| exactScheduleTime, | |
| }, | |
| availableAt: distributedExecutionTime, | |
| }); |
🤖 Prompt for AI Agents
In internal-packages/schedule-engine/src/engine/index.ts around lines 606 to
615, the job ID uses only the instanceId, causing potential collisions if
multiple runs of the same instance are enqueued before pruning. To fix this,
modify the job ID to include the scheduled timestamp or a unique incrementing
counter along with the instanceId to ensure each job ID is unique and prevent
silent overwrites or rejections.
No description provided.