diff --git a/.changeset/chat-route-spurious-wakes.md b/.changeset/chat-route-spurious-wakes.md new file mode 100644 index 00000000000..cd247801525 --- /dev/null +++ b/.changeset/chat-route-spurious-wakes.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/sdk": patch +--- + +Chat agents now return to a durable wait when a session wake does not deliver a matching message. This prevents resumed runs from staying active until their maximum duration and preserves the configured turn timeout across repeated wakes. diff --git a/.changeset/void-ipc-acks.md b/.changeset/void-ipc-acks.md new file mode 100644 index 00000000000..85caba20d35 --- /dev/null +++ b/.changeset/void-ipc-acks.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Fix a ~6-second delay between a task finishing and its run completing (and a ~31-second delay when cancelling a run) in projects that use zod 4.4 or newer. Run cost and billed usage was not impacted by this issue. diff --git a/.changeset/windows-deploy-paths-with-spaces.md b/.changeset/windows-deploy-paths-with-spaces.md new file mode 100644 index 00000000000..54fae1d759d --- /dev/null +++ b/.changeset/windows-deploy-paths-with-spaces.md @@ -0,0 +1,5 @@ +--- +"trigger.dev": patch +--- + +Fix Windows deploys failing at indexing with `Cannot find module` on a percent-encoded path when the project directory contains spaces or non-ASCII characters. diff --git a/.github/workflows/e2e-webapp.yml b/.github/workflows/e2e-webapp.yml index 48cc1d6df40..f9b6b79f311 100644 --- a/.github/workflows/e2e-webapp.yml +++ b/.github/workflows/e2e-webapp.yml @@ -102,7 +102,7 @@ jobs: redis:7.2 \ testcontainers/ryuk:0.14.0 \ ghcr.io/s2-streamstore/s2:0.40.0@sha256:b26249e2ede0949755f5af8028185dc2bcfc3aa2db21eb9610543d144eb6ee9d \ - quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b do pull "$image" & pids+=("$!") diff --git a/.github/workflows/unit-tests-webapp.yml b/.github/workflows/unit-tests-webapp.yml index f15f9991a49..45d15be4015 100644 --- a/.github/workflows/unit-tests-webapp.yml +++ b/.github/workflows/unit-tests-webapp.yml @@ -109,7 +109,7 @@ jobs: redis:7.2 \ testcontainers/ryuk:0.14.0 \ electricsql/electric:1.2.4@sha256:20da3d0b0e74926c5623392db67fd56698b9e374c4aeb6cb5cadeb8fea171c36 \ - quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b do pull "$image" & pids+=("$!") diff --git a/.server-changes/batch-completion-recovery.md b/.server-changes/batch-completion-recovery.md new file mode 100644 index 00000000000..405625627de --- /dev/null +++ b/.server-changes/batch-completion-recovery.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Prevent batch waits from remaining suspended when batch completion is briefly interrupted diff --git a/.server-changes/declarative-schedule-quota.md b/.server-changes/declarative-schedule-quota.md new file mode 100644 index 00000000000..3e63083cff5 --- /dev/null +++ b/.server-changes/declarative-schedule-quota.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Allow deployments to replace declarative schedules at the schedule limit when the resulting set stays within quota. diff --git a/.server-changes/otlp-worker-pool-stale-task-reap.md b/.server-changes/otlp-worker-pool-stale-task-reap.md new file mode 100644 index 00000000000..f5da2aa269d --- /dev/null +++ b/.server-changes/otlp-worker-pool-stale-task-reap.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Fixed a bug where a burst of telemetry could leave OpenTelemetry ingest rejecting every batch for a long time. Batches that wait too long are now dropped individually instead of restarting the processing workers, so ingest recovers as soon as the burst passes. diff --git a/.server-changes/schedule-last-run.md b/.server-changes/schedule-last-run.md new file mode 100644 index 00000000000..b1a05c63e41 --- /dev/null +++ b/.server-changes/schedule-last-run.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: fix +--- + +Keep scheduled task "Last run" times stable across unchanged deployments and align them with configured schedule windows. diff --git a/.server-changes/simplify-onboarding-questions.md b/.server-changes/simplify-onboarding-questions.md new file mode 100644 index 00000000000..6f0153d5b51 --- /dev/null +++ b/.server-changes/simplify-onboarding-questions.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Creating a project now asks only for its name. diff --git a/apps/webapp/app/components/onboarding/TechnologyPicker.tsx b/apps/webapp/app/components/onboarding/TechnologyPicker.tsx deleted file mode 100644 index e70af0d7030..00000000000 --- a/apps/webapp/app/components/onboarding/TechnologyPicker.tsx +++ /dev/null @@ -1,386 +0,0 @@ -import * as Ariakit from "@ariakit/react"; -import { - XMarkIcon, - PlusIcon, - CubeIcon, - MagnifyingGlassIcon, - ChevronDownIcon, -} from "@heroicons/react/20/solid"; -import { useCallback, useMemo, useRef, useState } from "react"; -import { CheckboxIndicator } from "~/components/primitives/CheckboxIndicator"; -import { cn } from "~/utils/cn"; -import { matchSorter } from "match-sorter"; -import { ShortcutKey } from "~/components/primitives/ShortcutKey"; - -const pillColors = [ - "bg-green-800/40 border-green-600/50", - "bg-teal-800/40 border-teal-600/50", - "bg-blue-800/40 border-blue-600/50", - "bg-indigo-800/40 border-indigo-600/50", - "bg-violet-800/40 border-violet-600/50", - "bg-purple-800/40 border-purple-600/50", - "bg-fuchsia-800/40 border-fuchsia-600/50", - "bg-pink-800/40 border-pink-600/50", - "bg-rose-800/40 border-rose-600/50", - "bg-orange-800/40 border-orange-600/50", - "bg-amber-800/40 border-amber-600/50", - "bg-yellow-800/40 border-yellow-600/50", - "bg-lime-800/40 border-lime-600/50", - "bg-emerald-800/40 border-emerald-600/50", - "bg-cyan-800/40 border-cyan-600/50", - "bg-sky-800/40 border-sky-600/50", -]; - -function getPillColor(value: string): string { - let hash = 0; - for (let i = 0; i < value.length; i++) { - hash = (hash << 5) - hash + value.charCodeAt(i); - hash |= 0; - } - return pillColors[Math.abs(hash) % pillColors.length]; -} - -const TECHNOLOGY_OPTIONS = [ - "Airflow", - "Angular", - "Anthropic", - "Astro", - "Auth0", - "AWS", - "AWS SQS", - "Azure", - "BigQuery", - "BullMQ", - "Bun", - "Cassandra", - "Celery", - "ClickHouse", - "Clerk", - "Cloudflare", - "CockroachDB", - "Cohere", - "Convex", - "Databricks", - "Datadog", - "DeepSeek", - "Deno", - "DigitalOcean", - "Django", - "Docker", - "Drizzle", - "DynamoDB", - "Elasticsearch", - "Electron", - "Elevenlabs", - "Expo", - "Express", - "FastAPI", - "Fastify", - "Firebase", - "Flask", - "Fly.io", - "Gatsby", - "GCP", - "Go", - "Google Cloud Tasks", - "Google Gemini", - "GraphQL", - "Groq", - "Heroku", - "Hono", - "htmx", - "Hugging Face", - "Inngest", - "Kafka", - "Kubernetes", - "LangChain", - "Laravel", - "LlamaIndex", - "MariaDB", - "Midjourney", - "Mistral", - "MongoDB", - "Mongoose", - "MySQL", - "Neo4j", - "Neon", - "Nest.js", - "Netlify", - "Next.js", - "Node.js", - "Nuxt", - "Ollama", - "OpenAI", - "Perplexity", - "PHP", - "Pinecone", - "PlanetScale", - "Python", - "PostHog", - "PostgreSQL", - "Prisma", - "Pulumi", - "RabbitMQ", - "Railway", - "React", - "React Native", - "Redis", - "Redshift", - "Remix", - "Render", - "Replicate", - "Resend", - "Ruby on Rails", - "Rust", - "SendGrid", - "Sentry", - "Sidekiq", - "Snowflake", - "Solid.js", - "Spring Boot", - "SQLite", - "Stability AI", - "Stripe", - "Supabase", - "Svelte", - "SvelteKit", - "Tailwind CSS", - "Temporal", - "Terraform", - "Together AI", - "tRPC", - "Turso", - "Twilio", - "TypeORM", - "Upstash", - "Vercel", - "Vercel AI SDK", - "Vite", - "Vue", - "Weaviate", -] as const; - -type TechnologyPickerProps = { - value: string[]; - onChange: (value: string[]) => void; - customValues: string[]; - onCustomValuesChange: (values: string[]) => void; -}; - -export function TechnologyPicker({ - value, - onChange, - customValues, - onCustomValuesChange, -}: TechnologyPickerProps) { - const [open, setOpen] = useState(false); - const [searchValue, setSearchValue] = useState(""); - const [otherInputValue, setOtherInputValue] = useState(""); - const [showOtherInput, setShowOtherInput] = useState(false); - const otherInputRef = useRef(null); - - const allSelected = useMemo(() => [...value, ...customValues], [value, customValues]); - - const filteredOptions = useMemo(() => { - if (!searchValue) return TECHNOLOGY_OPTIONS; - return matchSorter([...TECHNOLOGY_OPTIONS], searchValue); - }, [searchValue]); - - const toggleOption = useCallback( - (option: string) => { - if (value.includes(option)) { - onChange(value.filter((v) => v !== option)); - } else { - onChange([...value, option]); - } - }, - [value, onChange] - ); - - const removeItem = useCallback( - (item: string) => { - if (value.includes(item)) { - onChange(value.filter((v) => v !== item)); - } else { - onCustomValuesChange(customValues.filter((v) => v !== item)); - } - }, - [value, onChange, customValues, onCustomValuesChange] - ); - - const addCustomValue = useCallback(() => { - const trimmed = otherInputValue.trim(); - if (!trimmed) return; - - const matchedOption = TECHNOLOGY_OPTIONS.find( - (opt) => opt.toLowerCase() === trimmed.toLowerCase() - ); - - if (matchedOption) { - if (!value.includes(matchedOption)) { - onChange([...value, matchedOption]); - } - } else if (!customValues.includes(trimmed) && !value.includes(trimmed)) { - onCustomValuesChange([...customValues, trimmed]); - } - - setOtherInputValue(""); - }, [otherInputValue, customValues, onCustomValuesChange, value, onChange]); - - const handleOtherKeyDown = useCallback( - (e: React.KeyboardEvent) => { - if (e.key === "Enter") { - e.preventDefault(); - e.stopPropagation(); - addCustomValue(); - } - }, - [addCustomValue] - ); - - return ( -
- {allSelected.length > 0 && ( -
- {allSelected.map((item) => ( - - {item} - - - ))} -
- )} - - { - setSearchValue(val); - }} - > - { - if (Array.isArray(v)) { - onChange(v); - } - }} - virtualFocus - > - -
- - Select your technologies… -
- - - - -
- - -
- - - {filteredOptions.map((option) => ( - { - e.preventDefault(); - toggleOption(option); - }} - > -
- - {option} -
- - ))} - - {filteredOptions.length === 0 && searchValue && ( -
- No matches for “{searchValue}” -
- )} - - -
- {showOtherInput ? ( -
- setOtherInputValue(e.target.value)} - onKeyDown={handleOtherKeyDown} - placeholder="Type and press Enter to add" - className="flex-1 border-none bg-transparent pl-2 text-2sm text-text-bright shadow-none outline-hidden ring-0 placeholder:text-text-dimmed focus:border-none focus:outline-hidden focus:ring-0" - autoFocus - /> - 0 ? "opacity-100" : "opacity-0" - )} - /> - -
- ) : ( - - )} -
- - - -
- ); -} diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx index 397a5eeb953..550cb0d39f4 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug_.projects.new/route.tsx @@ -1,6 +1,6 @@ import { getFormProps, getInputProps, useForm } from "@conform-to/react"; import { parseWithZod } from "@conform-to/zod/v4"; -import { CommandLineIcon, FolderIcon } from "@heroicons/react/20/solid"; +import { FolderIcon } from "@heroicons/react/20/solid"; import { json, redirectDocument, @@ -8,15 +8,12 @@ import { type LoaderFunctionArgs, } from "@remix-run/node"; import { Form, useActionData, useNavigation } from "@remix-run/react"; -import type { Prisma } from "@trigger.dev/database"; -import React, { useEffect, useState } from "react"; import { redirect, typedjson, useTypedLoaderData } from "remix-typedjson"; import invariant from "tiny-invariant"; import { z } from "zod"; import { BackgroundWrapper } from "~/components/BackgroundWrapper"; import { Feedback } from "~/components/Feedback"; import { AppContainer, MainCenteredContainer } from "~/components/layout/AppLayout"; -import { TechnologyPicker } from "~/components/onboarding/TechnologyPicker"; import { Button, LinkButton } from "~/components/primitives/Buttons"; import { Callout } from "~/components/primitives/Callout"; import { Fieldset } from "~/components/primitives/Fieldset"; @@ -26,7 +23,6 @@ import { FormTitle } from "~/components/primitives/FormTitle"; import { Input } from "~/components/primitives/Input"; import { InputGroup } from "~/components/primitives/InputGroup"; import { Label } from "~/components/primitives/Label"; -import { Select, SelectItem } from "~/components/primitives/Select"; import { prisma } from "~/db.server"; import { featuresForRequest } from "~/features.server"; @@ -45,82 +41,6 @@ import { pageMeta } from "~/utils/pageTitle"; export const meta = pageMeta("New project"); -const WORKING_ON_OTHER = "Other/not sure yet"; -const GOALS_OTHER = "Other/not sure yet"; - -const workingOnOptions = [ - "AI agent", - "Media processing pipeline", - "Media generation with AI", - "Event-driven workflow", - "Realtime streaming", - "Internal tool or background job", - WORKING_ON_OTHER, -] as const; - -const goalOptions = [ - "Ship a production workflow", - "Prototype or explore", - "Migrate an existing system", - "Learn how Trigger works", - "Evaluate against alternatives", - GOALS_OTHER, -] as const; - -function shuffleArray(arr: T[]): T[] { - const shuffled = [...arr]; - for (let i = shuffled.length - 1; i > 0; i--) { - const j = Math.floor(Math.random() * (i + 1)); - [shuffled[i], shuffled[j]] = [shuffled[j], shuffled[i]]; - } - return shuffled; -} - -function renderMultiSelectValue(value: string[]) { - if (value.length === 0) return; - - return ( - - {value.slice(0, 2).join(", ")} - {value.length > 2 && +{value.length - 2} more} - - ); -} - -function MultiSelectField({ - value, - setValue, - items, - icon, -}: { - value: string[]; - setValue: (value: string[]) => void; - items: string[]; - icon: React.ReactNode; -}) { - return ( - - value={value} - setValue={setValue} - placeholder="Select some options" - variant="secondary/small" - dropdownIcon - icon={icon} - items={items} - className="h-8 min-w-0 border-0 bg-background-hover pl-2 text-sm text-text-dimmed ring-border-bright transition hover:bg-secondary hover:text-text-dimmed hover:ring-1" - text={renderMultiSelectValue} - > - {(items) => - items.map((item) => ( - - {item} - - )) - } - - ); -} - export async function loader({ params, request }: LoaderFunctionArgs) { const userId = await requireUserId(request); const { organizationSlug } = OrganizationParamsSchema.parse(params); @@ -171,14 +91,6 @@ export async function loader({ params, request }: LoaderFunctionArgs) { const schema = z.object({ projectName: z.string().min(3, "Project name must have at least 3 characters").max(50), projectVersion: z.enum(["v2", "v3"]), - workingOn: z.string().optional(), - workingOnOther: z.string().optional(), - technologies: z.string().optional(), - technologiesOther: z.string().optional(), - goals: z.string().optional(), - goalsOther: z.string().optional(), - workingOnPositions: z.string().optional(), - goalsPositions: z.string().optional(), }); export const action: ActionFunction = async ({ request, params }) => { @@ -198,66 +110,12 @@ export const action: ActionFunction = async ({ request, params }) => { const configurationId = url.searchParams.get("configurationId"); const next = url.searchParams.get("next"); - const stringArraySchema = z.array(z.string()); - - function safeParseStringArray(value: string | undefined): string[] | undefined { - if (!value) return undefined; - try { - const result = stringArraySchema.safeParse(JSON.parse(value)); - return result.success && result.data.length > 0 ? result.data : undefined; - } catch { - return undefined; - } - } - - const numberArraySchema = z.array(z.number()); - function safeParseNumberArray(value: string | undefined): number[] | undefined { - if (!value) return undefined; - try { - const result = numberArraySchema.safeParse(JSON.parse(value)); - return result.success && result.data.length > 0 ? result.data : undefined; - } catch { - return undefined; - } - } - - const onboardingData: Record = {}; - - const workingOn = safeParseStringArray(submission.value.workingOn); - if (workingOn) { - onboardingData.workingOn = workingOn; - const workingOnPositions = safeParseNumberArray(submission.value.workingOnPositions); - if (workingOnPositions) onboardingData.workingOnPositions = workingOnPositions; - } - - if (submission.value.workingOnOther) { - onboardingData.workingOnOther = submission.value.workingOnOther; - } - - const technologies = safeParseStringArray(submission.value.technologies); - if (technologies) onboardingData.technologies = technologies; - - const technologiesOther = safeParseStringArray(submission.value.technologiesOther); - if (technologiesOther) onboardingData.technologiesOther = technologiesOther; - - const goals = safeParseStringArray(submission.value.goals); - if (goals) { - onboardingData.goals = goals; - const goalsPositions = safeParseNumberArray(submission.value.goalsPositions); - if (goalsPositions) onboardingData.goalsPositions = goalsPositions; - } - - if (submission.value.goalsOther) { - onboardingData.goalsOther = submission.value.goalsOther; - } - try { const project = await createProject({ organizationSlug: organizationSlug, name: submission.value.projectName, userId, version: submission.value.projectVersion, - onboardingData: Object.keys(onboardingData).length > 0 ? onboardingData : undefined, }); if (code && configurationId) { @@ -348,28 +206,6 @@ export default function Page() { const navigation = useNavigation(); const isLoading = navigation.state === "submitting" || navigation.state === "loading"; - const [selectedWorkingOn, setSelectedWorkingOn] = useState([]); - const [workingOnOther, setWorkingOnOther] = useState(""); - const [selectedTechnologies, setSelectedTechnologies] = useState([]); - const [customTechnologies, setCustomTechnologies] = useState([]); - const [selectedGoals, setSelectedGoals] = useState([]); - const [goalsOther, setGoalsOther] = useState(""); - - const [shuffledWorkingOn, setShuffledWorkingOn] = useState([...workingOnOptions]); - const [shuffledGoals, setShuffledGoals] = useState([...goalOptions]); - - useEffect(() => { - const nonOther = workingOnOptions.filter((o) => o !== WORKING_ON_OTHER); - // oxlint-disable-next-line react/set-state-in-effect -- This effect intentionally synchronizes route state after an external or lifecycle change. - setShuffledWorkingOn([...shuffleArray(nonOther), WORKING_ON_OTHER]); - - const nonOtherGoals = goalOptions.filter((o) => o !== GOALS_OTHER); - setShuffledGoals([...shuffleArray(nonOtherGoals), GOALS_OTHER]); - }, []); - - const showWorkingOnOther = selectedWorkingOn.includes(WORKING_ON_OTHER); - const showGoalsOther = selectedGoals.includes(GOALS_OTHER); - return ( @@ -414,89 +250,6 @@ export default function Page() { /> )} -
- - - - shuffledWorkingOn.indexOf(v) + 1) - )} - /> - } - /> - {showWorkingOnOther && ( - <> - - setWorkingOnOther(e.target.value)} - placeholder="Tell us what you're working on" - spellCheck={false} - containerClassName="h-8" - /> - - )} - - - - - - - - - - - - - shuffledGoals.indexOf(v) + 1))} - /> - } - /> - {showGoalsOther && ( - <> - - setGoalsOther(e.target.value)} - placeholder="Tell us what you're trying to do with Trigger.dev" - spellCheck={false} - containerClassName="h-8" - /> - - )} - - diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 51bc0bca831..0d73728d457 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -1,5 +1,6 @@ import { Worker } from "node:worker_threads"; import path from "node:path"; +import { performance } from "node:perf_hooks"; import { getMeter, type Counter, @@ -31,15 +32,40 @@ type Task = { // Wall-clock stamp at enqueue; the task-duration histogram measures enqueue -> terminal state // (queue wait + worker compute), so the gap from the worker-reported compute time is queue wait. enqueuedAt: number; + /** Monotonic stamps (performance.now) drive every deadline decision so a wall-clock step can't shed or reap. */ + enqueuedAtMono: number; + dispatchedAtMono?: number; }; type ReapReason = "error" | "exit" | "timeout"; +export type ReapEvent = { + reason: ReapReason; + taskAgeMs?: number; + sinceDispatchMs?: number; + queueDepth: number; + aliveWorkers: number; +}; + +export type OtlpWorkerPoolOptions = { + taskTimeoutMs?: number; + respawnBaseMs?: number; + respawnMaxMs?: number; + /** + * Observer for every reap, carrying the same fields as the warn log. It must not affect pool + * liveness: it runs after the worker is terminated and the respawn is scheduled, and any + * exception it throws or promise it rejects is logged and swallowed. + */ + onReap?: (event: ReapEvent) => void | Promise; +}; + const TASK_TIMEOUT_MS = 30_000; const MAX_QUEUE_DEPTH = 2_000; const RESPAWN_BASE_MS = 500; const RESPAWN_MAX_MS = 30_000; const SHUTDOWN_DRAIN_MS = 5_000; +const STALE_FLOOR_MAX_MS = 1_000; +const SHED_LOG_INTERVAL_MS = 1_000; // Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run // fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers. @@ -48,11 +74,21 @@ export class OtlpWorkerPool { private readonly idle: Worker[] = []; private readonly queue: number[] = []; private readonly tasks = new Map(); - private readonly busyByWorker = new Map(); + private readonly busyByWorker = new Map(); + private readonly computeTimers = new Map(); private nextId = 1; private consecutiveFailures = 0; private isShuttingDown = false; private latestPricingModels: unknown[]; + private readonly taskTimeoutMs: number; + private readonly respawnBaseMs: number; + private readonly respawnMaxMs: number; + private readonly staleFloorMs: number; + private readonly onReap?: (event: ReapEvent) => void | Promise; + private shedInWindow = 0; + private shedWindowStartMono = 0; + private lastShedMono = 0; + private shedFlushTimer?: NodeJS.Timeout; // Pre-allocated per-kind {kind} attribute objects so the per-task record path never allocates. private readonly _kindAttrs: Record = { @@ -69,9 +105,15 @@ export class OtlpWorkerPool { private readonly size: number, private readonly workerPath: string, pricingModels: unknown[], - meter?: Meter + meter?: Meter, + options?: OtlpWorkerPoolOptions ) { this.latestPricingModels = pricingModels; + this.taskTimeoutMs = options?.taskTimeoutMs ?? TASK_TIMEOUT_MS; + this.respawnBaseMs = options?.respawnBaseMs ?? RESPAWN_BASE_MS; + this.respawnMaxMs = options?.respawnMaxMs ?? RESPAWN_MAX_MS; + this.staleFloorMs = Math.min(STALE_FLOOR_MAX_MS, this.taskTimeoutMs / 10); + this.onReap = options?.onReap; this.#setupOtelMetrics(meter); for (let i = 0; i < size; i++) this.spawn(); logger.info("OtlpWorkerPool started", { size, workerPath }); @@ -131,6 +173,15 @@ export class OtlpWorkerPool { } } + /** + * A task that already timed out for its caller recorded its outcome and duration then; its + * compute time only becomes known when the worker finally replies, and skipping it would drop + * exactly the slow samples from the compute histogram. + */ + #recordLateCompute(task: Task, computeMs: number): void { + this._computeDurationHistogram?.record(computeMs, this._kindAttrs[task.message.kind]); + } + private spawn() { const worker = new Worker(this.workerPath, { workerData: { pricingModels: this.latestPricingModels }, @@ -141,7 +192,9 @@ export class OtlpWorkerPool { (msg: { id: number; ok: boolean; result?: any; error?: string; computeMs?: number }) => { if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker this.consecutiveFailures = 0; + const inFlight = this.busyByWorker.get(worker); this.busyByWorker.delete(worker); + this.#clearComputeTimer(worker); const task = this.tasks.get(msg.id); if (task) { clearTimeout(task.timer); @@ -153,6 +206,8 @@ export class OtlpWorkerPool { this.#recordTaskEnd(task, "error", msg.computeMs); task.reject(new Error(msg.error ?? "otlp worker error")); } + } else if (inFlight?.message.id === msg.id && msg.computeMs !== undefined) { + this.#recordLateCompute(inFlight, msg.computeMs); } this.release(worker); } @@ -182,30 +237,92 @@ export class OtlpWorkerPool { const ii = this.idle.indexOf(worker); if (ii !== -1) this.idle.splice(ii, 1); + this.#clearComputeTimer(worker); - const inFlightId = this.busyByWorker.get(worker); + const now = performance.now(); + const inFlight = this.busyByWorker.get(worker); this.busyByWorker.delete(worker); - if (inFlightId !== undefined) { - const task = this.tasks.get(inFlightId); - if (task) { - clearTimeout(task.timer); - this.tasks.delete(inFlightId); - // A timed-out task already recorded its own end + was removed from the map, so this only - // fires for a crash that killed a task mid-flight. - this.#recordTaskEnd(task, "crash"); - task.reject(error); - } + if (inFlight !== undefined && this.tasks.has(inFlight.message.id)) { + clearTimeout(inFlight.timer); + this.tasks.delete(inFlight.message.id); + this.#recordTaskEnd(inFlight, "crash"); + inFlight.reject(error); } this._respawnsCounter?.add(1, { reason }); + const event: ReapEvent = { + reason, + queueDepth: this.queue.length, + aliveWorkers: this.workers.length, + taskAgeMs: inFlight === undefined ? undefined : Math.round(now - inFlight.enqueuedAtMono), + sinceDispatchMs: + inFlight?.dispatchedAtMono === undefined + ? undefined + : Math.round(now - inFlight.dispatchedAtMono), + }; + logger.warn("OtlpWorkerPool reaped worker", { ...event, error: error.message }); void worker.terminate().catch(() => {}); this.scheduleRespawn(); + this.#notifyReap(event); + } + + #notifyReap(event: ReapEvent): void { + if (this.onReap === undefined) return; + try { + const result = this.onReap(event); + if (result !== undefined && typeof result.then === "function") { + result.then(undefined, (thrown) => this.#logObserverError(thrown)); + } + } catch (thrown) { + this.#logObserverError(thrown); + } + } + + #logObserverError(thrown: unknown): void { + logger.error("OtlpWorkerPool onReap observer threw", { + error: thrown instanceof Error ? thrown.message : String(thrown), + }); + } + + #clearComputeTimer(worker: Worker) { + const timer = this.computeTimers.get(worker); + if (timer === undefined) return; + clearTimeout(timer); + this.computeTimers.delete(worker); + } + + /** + * A dispatched task whose caller deadline passed is not evidence the worker is stuck: it may + * have sat in the queue for most of its budget. Give the worker the rest of a full compute + * budget for it, and only reap if it still hasn't replied by then. The worker's reply (or a + * crash reap) clears this timer. + */ + #armComputeTimer(worker: Worker, task: Task, delayMs: number) { + this.#clearComputeTimer(worker); + const timer = setTimeout( + () => { + this.computeTimers.delete(worker); + if (this.busyByWorker.get(worker) !== task) return; + const remainingMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono!); + if (remainingMs > 0) { + this.#armComputeTimer(worker, task, remainingMs); + return; + } + this.reap( + worker, + new Error(`otlp worker stuck for ${this.taskTimeoutMs}ms on a single task`), + "timeout" + ); + }, + Math.max(1, Math.ceil(delayMs)) + ); + this.computeTimers.set(worker, timer); } private scheduleRespawn() { if (this.isShuttingDown) return; if (this.workers.length >= this.size) return; - const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS); + const delay = Math.min(this.respawnBaseMs * 2 ** this.consecutiveFailures, this.respawnMaxMs); this.consecutiveFailures++; setTimeout(() => { if (this.isShuttingDown) return; @@ -219,28 +336,83 @@ export class OtlpWorkerPool { this.drain(); } + /** + * Hand queued tasks to idle workers in FIFO order, shedding any task whose remaining budget is + * below the stale floor so a free worker starts on something it can still finish. The floor is + * a small fraction of the timeout, so only tasks whose deadline is effectively already here are + * dropped; everything else keeps its FIFO turn. + */ private drain() { + const now = performance.now(); while (this.queue.length > 0 && this.idle.length > 0) { - const worker = this.idle.pop()!; const id = this.queue.shift()!; const task = this.tasks.get(id); if (!task) continue; + const remainingMs = task.enqueuedAtMono + this.taskTimeoutMs - now; + if (remainingMs < this.staleFloorMs) { + this.#shed(task, remainingMs, now); + continue; + } + const worker = this.idle.pop()!; task.worker = worker; - this.busyByWorker.set(worker, id); + task.dispatchedAtMono = now; + this.busyByWorker.set(worker, task); worker.postMessage(task.message, task.transfer); } } + /** + * Sheds are aggregated into at most one debug line per second; each line carries the count and + * the span between the first and last shed it covers. + */ + #shed(task: Task, remainingMs: number, now: number) { + clearTimeout(task.timer); + this.tasks.delete(task.message.id); + this.#recordTaskEnd(task, "stale"); + task.reject( + new Error( + `otlp worker task shed after ${Math.round(now - task.enqueuedAtMono)}ms in queue with ${Math.max( + 0, + Math.round(remainingMs) + )}ms of ${this.taskTimeoutMs}ms budget left` + ) + ); + if (this.shedInWindow === 0) this.shedWindowStartMono = now; + this.lastShedMono = now; + this.shedInWindow++; + if (this.shedFlushTimer !== undefined) return; + this.shedFlushTimer = setTimeout(() => { + this.shedFlushTimer = undefined; + logger.debug("OtlpWorkerPool shed stale tasks", { + shed: this.shedInWindow, + windowMs: Math.round(this.lastShedMono - this.shedWindowStartMono), + queueDepth: this.queue.length, + aliveWorkers: this.workers.length, + }); + this.shedInWindow = 0; + }, SHED_LOG_INTERVAL_MS); + this.shedFlushTimer.unref(); + } + + /** + * The caller's budget (queue wait + compute) is spent, so reject it now. Whether the worker is + * at fault depends on how long it has held the task: reap only once it has had a full timeout + * of compute time on this one task, otherwise let it finish via the compute timer and return to + * the idle set on reply. The task is removed here, so neither path can double-reject. + */ private onTimeout(id: number) { const task = this.tasks.get(id); if (!task) return; this.tasks.delete(id); this.#recordTaskEnd(task, "timeout"); - const err = new Error(`otlp worker task timed out after ${TASK_TIMEOUT_MS}ms`); - if (task.worker) { - // Dispatched to a stuck worker: reap it. The task is already removed, so reap won't - // double-reject. - this.reap(task.worker, err, "timeout"); + const err = new Error(`otlp worker task timed out after ${this.taskTimeoutMs}ms`); + if (task.worker !== undefined && task.dispatchedAtMono !== undefined) { + const remainingComputeMs = this.taskTimeoutMs - (performance.now() - task.dispatchedAtMono); + if (remainingComputeMs <= 0) { + this.reap(task.worker, err, "timeout"); + } else { + this.#armComputeTimer(task.worker, task, remainingComputeMs); + } } else { const qi = this.queue.indexOf(id); if (qi !== -1) this.queue.splice(qi, 1); @@ -262,7 +434,7 @@ export class OtlpWorkerPool { } const id = this.nextId++; return new Promise((resolve, reject) => { - const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS); + const timer = setTimeout(() => this.onTimeout(id), this.taskTimeoutMs); this.tasks.set(id, { message: { id, @@ -277,6 +449,7 @@ export class OtlpWorkerPool { reject, timer, enqueuedAt: Date.now(), + enqueuedAtMono: performance.now(), }); this.queue.push(id); this.drain(); @@ -319,6 +492,10 @@ export class OtlpWorkerPool { this.idle.length = 0; this.queue.length = 0; this.busyByWorker.clear(); + for (const timer of this.computeTimers.values()) clearTimeout(timer); + this.computeTimers.clear(); + clearTimeout(this.shedFlushTimer); + this.shedFlushTimer = undefined; // Reject anything that didn't drain within the deadline. for (const [, task] of this.tasks) { clearTimeout(task.timer); diff --git a/apps/webapp/app/v3/runEngineHandlers.server.ts b/apps/webapp/app/v3/runEngineHandlers.server.ts index 4ef52805e1e..2d3b6022940 100644 --- a/apps/webapp/app/v3/runEngineHandlers.server.ts +++ b/apps/webapp/app/v3/runEngineHandlers.server.ts @@ -9,7 +9,6 @@ import { runOpsNewReplica, runOpsLegacyReplica, runOpsNewPrismaClient, - runOpsNewReplicaClient, runOpsLegacyPrismaClient, runOpsShardHandles, } from "~/db.server"; @@ -1060,7 +1059,6 @@ export function setupBatchQueueCallbacks() { engine.setBatchCompletionCallback(async (result: CompleteBatchResult) => { await handleBatchCompletion(result, { splitEnabled: await splitEnabledPromise, - newReplica: runOpsNewReplicaClient, newWriter: runOpsNewPrismaClient, legacyWriter: runOpsLegacyPrismaClient, shards: runOpsShardHandles, diff --git a/apps/webapp/app/v3/runEngineHandlersShared.server.ts b/apps/webapp/app/v3/runEngineHandlersShared.server.ts index a20d531ba7e..a2bfbe19f3d 100644 --- a/apps/webapp/app/v3/runEngineHandlersShared.server.ts +++ b/apps/webapp/app/v3/runEngineHandlersShared.server.ts @@ -76,12 +76,12 @@ export async function readRunForEventOrThrow( * on a single run-ops DB. Length classification is INVALID here: a batch id may * be a run-ops id (cut-over orgs) or a cuid (and cuid-shaped ids can be backfilled * onto NEW), so id-shape does not reliably indicate the row's actual residency. - * The existence probe is the correct signal. + * The existence probe uses the writer because batch finalization is a read-after-write + * path and a lagging replica can incorrectly route a NEW batch to LEGACY. */ export async function resolveBatchRunOpsWriter( batchId: string, deps: { - newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; @@ -101,7 +101,7 @@ export async function resolveBatchRunOpsWriter( return shard.writer; } - const onNew = await deps.newReplica.batchTaskRun.findFirst({ + const onNew = await deps.newWriter.batchTaskRun.findFirst({ where: { id: batchId }, select: { id: true }, }); @@ -119,7 +119,6 @@ export const QUEUE_SIZE_LIMIT_EXCEEDED_ERROR_CODE = "QUEUE_SIZE_LIMIT_EXCEEDED"; export type BatchCompletionDeps = { splitEnabled: boolean; - newReplica: RunOpsPrismaClient; newWriter: RunOpsPrismaClient; legacyWriter: RunOpsPrismaClient; shards?: ReadonlyArray<{ key: string; writer: RunOpsPrismaClient }>; @@ -150,7 +149,6 @@ export async function handleBatchCompletion( // Always probe residency — never special-case on splitEnabled (see commit msg). const runOpsWriter = await resolveBatchRunOpsWriter(batchId, { - newReplica: deps.newReplica, newWriter: deps.newWriter, legacyWriter: deps.legacyWriter, shards: deps.shards, diff --git a/apps/webapp/app/v3/scheduleTimings.server.ts b/apps/webapp/app/v3/scheduleTimings.server.ts index 04c35ec98f9..c22af28fb8b 100644 --- a/apps/webapp/app/v3/scheduleTimings.server.ts +++ b/apps/webapp/app/v3/scheduleTimings.server.ts @@ -88,7 +88,10 @@ export function resolveScheduleTimings( { phaseSecret, includeLastRun, now = new Date() }: ResolveScheduleTimingsOptions ): ScheduleTiming[] { const nominalCache = new Map(); - const previousCache = new Map(); + const previousCache = new Map< + string, + { latestNominal: Date; previousNominal?: Date } | undefined + >(); return inputs.map((input) => { const window: NormalizedScheduleWindow | undefined = resolveScheduleWindow({ @@ -134,51 +137,80 @@ export function resolveScheduleTimings( return { nextRun: nominalAt, nextRunEffectiveAt: effectiveAt, - lastRun: includeLastRun ? resolveLastRun(input, now, previousCache) : undefined, + lastRun: includeLastRun + ? resolveLastRun(input, now, nominalAt, phase, window, previousCache) + : undefined, }; }); } /** - * Approximates "last run" from the cron's previous slot. + * Approximates "last run" from the most recent effective schedule time. * - * Skips inactive schedules — the previous slot reflects what *would* have - * fired. Skips slots that predate `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 this degrades to undefined - * per row rather than failing the whole list. Best-effort by design; the runs - * page is the source of truth. + * Skips inactive schedules and effective times that predate `updatedAt`. Best-effort by design; + * the runs page is the source of truth. */ function resolveLastRun( input: ScheduleTimingInput, now: Date, - cache: Map + nextNominal: Date, + phase: number, + window: NormalizedScheduleWindow | undefined, + cache: Map ): Date | undefined { if (!input.active) { return undefined; } const key = cacheKey(input.cron, input.timezone); + let nominalTimes = cache.get(key); - let previous: Date | undefined; - if (cache.has(key)) { - previous = cache.get(key); - } else { + if (!cache.has(key)) { try { - previous = previousScheduledTimestamp(input.cron, input.timezone, now); + nominalTimes = { + latestNominal: previousScheduledTimestamp( + input.cron, + input.timezone, + new Date(now.getTime() + 1) + ), + }; } catch { - previous = undefined; + nominalTimes = undefined; } - cache.set(key, previous); + cache.set(key, nominalTimes); } - if (!previous) { + if (!nominalTimes) { return undefined; } - return previous.getTime() > input.updatedAt.getTime() ? previous : undefined; + const latestEffective = calculateEffectiveScheduleTime({ + nominalAt: nominalTimes.latestNominal, + nextNominalAt: nextNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + if (latestEffective.getTime() <= now.getTime()) { + return latestEffective.getTime() > input.updatedAt.getTime() ? latestEffective : undefined; + } + + if (!nominalTimes.previousNominal) { + nominalTimes.previousNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + nominalTimes.latestNominal + ); + } + const previousEffective = calculateEffectiveScheduleTime({ + nominalAt: nominalTimes.previousNominal, + nextNominalAt: nominalTimes.latestNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + + return previousEffective.getTime() > input.updatedAt.getTime() ? previousEffective : undefined; } /** diff --git a/apps/webapp/app/v3/services/checkSchedule.server.ts b/apps/webapp/app/v3/services/checkSchedule.server.ts index 946cfad080c..1458aa5f55b 100644 --- a/apps/webapp/app/v3/services/checkSchedule.server.ts +++ b/apps/webapp/app/v3/services/checkSchedule.server.ts @@ -17,7 +17,14 @@ type Schedule = { }; export class CheckScheduleService extends BaseService { - public async call(projectId: string, schedule: Schedule, environmentIds: string[]) { + public async call( + projectId: string, + schedule: Schedule, + environmentIds: string[], + quotaExclusions?: { environmentId: string; scheduleIds: string[] }, + pendingCreations = 0, + quotaLimit?: number + ) { //validate the cron expression try { CronPattern.parse(schedule.cron); @@ -115,15 +122,18 @@ export class CheckScheduleService extends BaseService { //if creating a schedule, check they're under the limits if (!schedule.friendlyId) { - const limit = await getLimit(project.organizationId, "schedules", 100_000_000); + const limit = + quotaLimit ?? (await getLimit(project.organizationId, "schedules", 100_000_000)); const schedulesCount = await CheckScheduleService.getUsedSchedulesCount({ prisma: this._prisma, projectId, + quotaExclusions, }); - if (schedulesCount >= limit) { + const projectedCount = schedulesCount + pendingCreations; + if (projectedCount >= limit) { throw new ServiceValidationError( - `You have created ${schedulesCount}/${limit} schedules so you'll need to increase your limits or delete some schedules.` + `You have created ${projectedCount}/${limit} schedules so you'll need to increase your limits or delete some schedules.` ); } } @@ -132,14 +142,24 @@ export class CheckScheduleService extends BaseService { static async getUsedSchedulesCount({ prisma, projectId, + quotaExclusions, }: { prisma: PrismaClientOrTransaction; projectId: string; + quotaExclusions?: { environmentId: string; scheduleIds: string[] }; }) { return await prisma.taskScheduleInstance.count({ where: { projectId, active: true, + ...(quotaExclusions?.scheduleIds.length + ? { + NOT: { + environmentId: quotaExclusions.environmentId, + taskScheduleId: { in: boundedIn(quotaExclusions.scheduleIds) }, + }, + } + : {}), environment: { projectId, type: { diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts index 675727db80b..11ab03545a9 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts @@ -394,6 +394,7 @@ describe("syncDeclarativeSchedules registration", () => { async ({ prisma, redisOptions }) => { const { project, prodEnv } = await seedProjectWithEnvs(prisma); const schedule = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id]); + const updatedAt = schedule.updatedAt; await seedScheduledTask(prisma, project.id, prodEnv.id); const engine = createTestScheduleEngine(prisma, redisOptions); @@ -412,6 +413,12 @@ describe("syncDeclarativeSchedules registration", () => { expect(before).toBeDefined(); expect(await engine.getJob(jobId)).toEqual(before); + expect( + await prisma.taskSchedule.findUniqueOrThrow({ + where: { id: schedule.id }, + select: { updatedAt: true }, + }) + ).toEqual({ updatedAt }); } finally { await engine.quit(); } @@ -492,6 +499,268 @@ describe("syncDeclarativeSchedules registration", () => { }); }); +describe("syncDeclarativeSchedules quota", () => { + containerTest( + "replaces a schedule at the limit without deleting it before validation", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const old = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const engine = createTestScheduleEngine(prisma, redisOptions); + + try { + await expect( + syncDeclarativeSchedules( + declarativeTasks({ cron: "not-a-cron", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ) + ).rejects.toThrow("Invalid cron expression"); + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + + await syncDeclarativeSchedules( + declarativeTasks({ cron: "0 * * * *", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ); + + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).toBeNull(); + expect( + await prisma.taskSchedule.findFirst({ + where: { projectId: project.id, taskIdentifier: "my-task" }, + }) + ).not.toBeNull(); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 1 + ); + const created = await prisma.taskSchedule.findFirstOrThrow({ + where: { projectId: project.id, taskIdentifier: "my-task" }, + include: { instances: true }, + }); + expect(await engine.getJob(scheduleJobId(created.instances[0].id))).toBeDefined(); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "rejects a later invalid schedule before writing an earlier replacement", + async ({ prisma }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + const old = await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const tasks = [ + { id: "my-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + { id: "other-task", schedule: { cron: "0 * * * *", timezone: "Not/A_Timezone" } }, + ] as TasksArg; + + await expect( + syncDeclarativeSchedules(tasks, noWorker, asEnv(prodEnv), prisma, undefined, undefined, 1) + ).rejects.toThrow("Invalid IANA timezone"); + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + expect(await prisma.taskSchedule.count({ where: { projectId: project.id } })).toBe(1); + + tasks[1].schedule.timezone = "UTC"; + await expect( + syncDeclarativeSchedules(tasks, noWorker, asEnv(prodEnv), prisma, undefined, undefined, 1) + ).rejects.toThrow("You have created 1/1 schedules"); + expect(await prisma.taskSchedule.count({ where: { projectId: project.id } })).toBe(1); + } + ); + + containerTest( + "does not count development schedules toward projected quota", + async ({ prisma, redisOptions }) => { + const { project, devEnv } = await seedProjectWithEnvs(prisma); + await seedScheduledTask(prisma, project.id, devEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${devEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: devEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + await syncDeclarativeSchedules( + [ + { id: "my-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + { id: "other-task", schedule: { cron: "0 * * * *", timezone: "UTC" } }, + ] as TasksArg, + noWorker, + asEnv(devEnv), + prisma, + engine, + undefined, + 1 + ); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 2 + ); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "serializes concurrent replacements at the limit", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + const results = await Promise.allSettled( + ["my-task", "other-task"].map((id) => + syncDeclarativeSchedules( + [{ id, schedule: { cron: "0 * * * *", timezone: "UTC" } }] as TasksArg, + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 1 + ) + ) + ); + expect(results.some((result) => result.status === "fulfilled")).toBe(true); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe( + 1 + ); + expect( + await prisma.taskSchedule.findFirstOrThrow({ where: { projectId: project.id } }) + ).toMatchObject({ + type: "DECLARATIVE", + }); + } finally { + await engine.quit(); + } + } + ); + + containerTest( + "concurrent replacements with spare quota leave only the last declaration", + async ({ prisma, redisOptions }) => { + const { project, prodEnv } = await seedProjectWithEnvs(prisma); + await makeDeclarativeSchedule(prisma, project.id, [prodEnv.id], "old-task"); + await seedScheduledTask(prisma, project.id, prodEnv.id); + const worker = await prisma.backgroundWorker.findFirstOrThrow({ + where: { projectId: project.id }, + }); + await prisma.backgroundWorkerTask.create({ + data: { + friendlyId: `task_other_${prodEnv.id}`, + slug: "other-task", + filePath: "src/trigger/other-task.ts", + workerId: worker.id, + projectId: project.id, + runtimeEnvironmentId: prodEnv.id, + triggerSource: "SCHEDULED", + }, + }); + const engine = createTestScheduleEngine(prisma, redisOptions); + try { + await Promise.all( + ["my-task", "other-task"].map((id) => + syncDeclarativeSchedules( + [{ id, schedule: { cron: "0 * * * *", timezone: "UTC" } }] as TasksArg, + noWorker, + asEnv(prodEnv), + prisma, + engine, + undefined, + 2 + ) + ) + ); + const schedules = await prisma.taskSchedule.findMany({ + where: { projectId: project.id }, + select: { taskIdentifier: true }, + }); + expect(schedules).toHaveLength(1); + expect(["my-task", "other-task"]).toContain(schedules[0].taskIdentifier); + } finally { + await engine.quit(); + } + } + ); + + containerTest("does not borrow quota from another environment's instance", async ({ prisma }) => { + const { project, prodEnv, devEnv } = await seedProjectWithEnvs(prisma); + await prisma.runtimeEnvironment.update({ + where: { id: devEnv.id }, + data: { type: "PRODUCTION" }, + }); + const old = await makeDeclarativeSchedule( + prisma, + project.id, + [prodEnv.id, devEnv.id], + "old-task" + ); + await seedScheduledTask(prisma, project.id, prodEnv.id); + + await expect( + syncDeclarativeSchedules( + declarativeTasks({ cron: "0 * * * *", timezone: "UTC" }), + noWorker, + asEnv(prodEnv), + prisma, + undefined, + undefined, + 1 + ) + ).rejects.toThrow("You have created 1/1 schedules"); + + expect(await prisma.taskSchedule.findUnique({ where: { id: old.id } })).not.toBeNull(); + expect(await prisma.taskScheduleInstance.count({ where: { projectId: project.id } })).toBe(2); + }); +}); + describe("syncDeclarativeSchedules deletion path", () => { containerTest( "does not issue any instance delete when the env owns no instance of the missing schedules", diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts index 13a24576042..f61845dd7f1 100644 --- a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts +++ b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts @@ -1399,6 +1399,7 @@ async function prepareDeclarativeSchedules( minimumWindowDurationSeconds: true, instances: { select: { + id: true, environmentId: true, }, }, @@ -1486,167 +1487,219 @@ export async function syncDeclarativeSchedules( environment: AuthenticatedEnvironment, prisma: PrismaClientOrTransaction, engine: Pick = scheduleEngine, - prepared?: Awaited> + prepared?: Awaited>, + quotaLimit?: number ) { - const { - existingDeclarativeSchedules, - preparedTasks, - defaultWindowDurationSeconds, - warnings = [], - } = prepared ?? (await prepareDeclarativeSchedules(tasks, environment, prisma)); - - //start out by assuming they're all missing - const missingSchedules = new Set( - existingDeclarativeSchedules.map((schedule) => schedule.id) - ); - - const checkSchedule = new CheckScheduleService(prisma); - - // Resource and quota checks must stay after worker resources exist and between schedule writes. - for (const { task, existingSchedule, minimumWindowDurationSeconds } of preparedTasks) { - //this throws errors if the schedule is invalid - await checkSchedule.call( - environment.projectId, - { - cron: task.schedule.cron, - timezone: task.schedule.timezone, - taskIdentifier: task.id, - friendlyId: existingSchedule?.friendlyId, - window: task.schedule.window, - }, - [environment.id] - ); + let { existingDeclarativeSchedules, preparedTasks, defaultWindowDurationSeconds, warnings } = + prepared ?? (await prepareDeclarativeSchedules(tasks, environment, prisma)); + + let missingSchedules = new Set(existingDeclarativeSchedules.map((schedule) => schedule.id)); + const declaredTaskIds = new Set(preparedTasks.map(({ task }) => task.id)); + let quotaExclusions = { + environmentId: environment.id, + scheduleIds: existingDeclarativeSchedules + .filter((schedule) => !declaredTaskIds.has(schedule.taskIdentifier)) + .map((schedule) => schedule.id), + }; - const normalizedWindow = normalizeScheduleWindow(task.schedule.window); - - if (existingSchedule) { - // Compare effective windows so equivalent explicit/default values preserve the pending job. - const previousWindow = resolveScheduleWindow({ - windowDurationSeconds: existingSchedule.windowDurationSeconds, - windowPercentage: existingSchedule.windowPercentage, - defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, - }).window; - const nextWindow = resolveScheduleWindow({ - windowDurationSeconds: normalizedWindow.windowDurationSeconds, - windowPercentage: normalizedWindow.windowPercentage, - defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, - }).window; - const timingChanged = - existingSchedule.generatorExpression !== task.schedule.cron || - existingSchedule.timezone !== task.schedule.timezone || - !scheduleWindowsEqual(previousWindow, nextWindow) || - // Clearing the minimum changes the effective range. - existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; - const schedule = await prisma.taskSchedule.update({ - where: { - id: existingSchedule.id, - }, - data: { - generatorExpression: task.schedule.cron, - generatorDescription: cronstrue.toString(task.schedule.cron), + const validate = async (db: PrismaClientOrTransaction) => { + const checkSchedule = new CheckScheduleService(db); + let pendingCreations = 0; + for (const { task, existingSchedule } of preparedTasks) { + await checkSchedule.call( + environment.projectId, + { + cron: task.schedule.cron, timezone: task.schedule.timezone, - minimumWindowDurationSeconds, - ...normalizedWindow, - }, - include: { - instances: true, - }, - }); - - missingSchedules.delete(existingSchedule.id); - const instances = timingChanged - ? schedule.instances - : schedule.instances.filter((instance) => instance.environmentId === environment.id); - if (instances.length === 0) { - throw new CreateDeclarativeScheduleError( - `Missing instance for declarative schedule ${schedule.id}` - ); - } - for (const instance of instances) { - await engine.registerNextTaskScheduleInstance({ - instanceId: instance.id, - preserveExistingJob: !timingChanged, - }); - } - } else { - const newSchedule = await prisma.taskSchedule.create({ - data: { - friendlyId: generateFriendlyId("sched"), - projectId: environment.projectId, taskIdentifier: task.id, - generatorExpression: task.schedule.cron, - generatorDescription: cronstrue.toString(task.schedule.cron), - timezone: task.schedule.timezone, - type: "DECLARATIVE", - minimumWindowDurationSeconds, - ...normalizedWindow, - defaultWindowDurationSeconds, - instances: { - create: [ - { - environmentId: environment.id, - projectId: environment.projectId, + friendlyId: existingSchedule?.friendlyId, + window: task.schedule.window, + }, + [environment.id], + quotaExclusions, + pendingCreations, + quotaLimit + ); + if (!existingSchedule && environment.type !== "DEVELOPMENT") pendingCreations++; + } + return pendingCreations; + }; + const pendingCreations = await validate(prisma); + + const registrations: Array<{ instanceId: string; preserveExistingJob?: boolean }> = []; + const transactional = + quotaExclusions.scheduleIds.length > 0 && pendingCreations > 0 && "$transaction" in prisma; + const apply = async (db: PrismaClientOrTransaction) => { + for (const { task, existingSchedule, minimumWindowDurationSeconds } of preparedTasks) { + const normalizedWindow = normalizeScheduleWindow(task.schedule.window); + + if (existingSchedule) { + // Compare effective windows so equivalent explicit/default values preserve the pending job. + const previousWindow = resolveScheduleWindow({ + windowDurationSeconds: existingSchedule.windowDurationSeconds, + windowPercentage: existingSchedule.windowPercentage, + defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, + }).window; + const nextWindow = resolveScheduleWindow({ + windowDurationSeconds: normalizedWindow.windowDurationSeconds, + windowPercentage: normalizedWindow.windowPercentage, + defaultWindowDurationSeconds: existingSchedule.defaultWindowDurationSeconds, + }).window; + const timingChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + !scheduleWindowsEqual(previousWindow, nextWindow) || + // Clearing the minimum changes the effective range. + existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; + const persistedValuesChanged = + existingSchedule.generatorExpression !== task.schedule.cron || + existingSchedule.timezone !== task.schedule.timezone || + existingSchedule.windowDurationSeconds !== normalizedWindow.windowDurationSeconds || + existingSchedule.windowPercentage !== normalizedWindow.windowPercentage || + existingSchedule.minimumWindowDurationSeconds !== minimumWindowDurationSeconds; + const schedule = persistedValuesChanged + ? await db.taskSchedule.update({ + where: { + id: existingSchedule.id, + }, + data: { + generatorExpression: task.schedule.cron, + generatorDescription: cronstrue.toString(task.schedule.cron), + timezone: task.schedule.timezone, + minimumWindowDurationSeconds, + ...normalizedWindow, + }, + include: { + instances: true, }, - ], + }) + : existingSchedule; + + missingSchedules.delete(existingSchedule.id); + const instances = timingChanged + ? schedule.instances + : schedule.instances.filter((instance) => instance.environmentId === environment.id); + if (instances.length === 0) { + throw new CreateDeclarativeScheduleError( + `Missing instance for declarative schedule ${schedule.id}` + ); + } + for (const instance of instances) { + const registration = { instanceId: instance.id, preserveExistingJob: !timingChanged }; + if (transactional) registrations.push(registration); + else await engine.registerNextTaskScheduleInstance(registration); + } + } else { + const newSchedule = await db.taskSchedule.create({ + data: { + friendlyId: generateFriendlyId("sched"), + projectId: environment.projectId, + taskIdentifier: task.id, + generatorExpression: task.schedule.cron, + generatorDescription: cronstrue.toString(task.schedule.cron), + timezone: task.schedule.timezone, + type: "DECLARATIVE", + minimumWindowDurationSeconds, + ...normalizedWindow, + defaultWindowDurationSeconds, + instances: { + create: [ + { + environmentId: environment.id, + projectId: environment.projectId, + }, + ], + }, }, - }, - include: { - instances: true, - }, - }); + include: { + instances: true, + }, + }); - const instance = newSchedule.instances.at(0); + const instance = newSchedule.instances.at(0); - if (instance) { - await engine.registerNextTaskScheduleInstance({ instanceId: instance.id }); - } else { - throw new CreateDeclarativeScheduleError( - `Missing instance for declarative schedule ${newSchedule.id}` - ); + if (instance) { + const registration = { instanceId: instance.id }; + if (transactional) registrations.push(registration); + else await engine.registerNextTaskScheduleInstance(registration); + } else { + throw new CreateDeclarativeScheduleError( + `Missing instance for declarative schedule ${newSchedule.id}` + ); + } } } - } - //Delete instances for this environment - //Delete schedules that have no instances left - const potentiallyDeletableSchedules = existingDeclarativeSchedules.filter((schedule) => - missingSchedules.has(schedule.id) - ); + //Delete instances for this environment + //Delete schedules that have no instances left + const potentiallyDeletableSchedules = existingDeclarativeSchedules.filter((schedule) => + missingSchedules.has(schedule.id) + ); - const scheduleIdsToDelete: string[] = []; - const scheduleIdsToDetachFromEnvironment: string[] = []; + const scheduleIdsToDelete: string[] = []; + const scheduleIdsToDetachFromEnvironment: string[] = []; - for (const schedule of potentiallyDeletableSchedules) { - const canDeleteSchedule = - schedule.instances.length === 0 || - schedule.instances.every((instance) => instance.environmentId === environment.id); + for (const schedule of potentiallyDeletableSchedules) { + const canDeleteSchedule = + schedule.instances.length === 0 || + schedule.instances.every((instance) => instance.environmentId === environment.id); - if (canDeleteSchedule) { - scheduleIdsToDelete.push(schedule.id); - } else if (schedule.instances.some((instance) => instance.environmentId === environment.id)) { - scheduleIdsToDetachFromEnvironment.push(schedule.id); + if (canDeleteSchedule) { + scheduleIdsToDelete.push(schedule.id); + } else if (schedule.instances.some((instance) => instance.environmentId === environment.id)) { + scheduleIdsToDetachFromEnvironment.push(schedule.id); + } } - } - if (scheduleIdsToDelete.length > 0) { - await prisma.taskSchedule.deleteMany({ - where: { - id: { - in: boundedIn(scheduleIdsToDelete), + if (scheduleIdsToDelete.length > 0) { + await db.taskSchedule.deleteMany({ + where: { + id: { + in: boundedIn(scheduleIdsToDelete), + }, }, - }, - }); - } + }); + } - if (scheduleIdsToDetachFromEnvironment.length > 0) { - await prisma.taskScheduleInstance.deleteMany({ - where: { - taskScheduleId: { - in: boundedIn(scheduleIdsToDetachFromEnvironment), + if (scheduleIdsToDetachFromEnvironment.length > 0) { + await db.taskScheduleInstance.deleteMany({ + where: { + taskScheduleId: { + in: boundedIn(scheduleIdsToDetachFromEnvironment), + }, + environmentId: environment.id, }, - environmentId: environment.id, + }); + } + }; + + if (transactional) { + await $transaction( + prisma, + "syncDeclarativeSchedules", + async (tx) => { + await tx.$queryRaw`SELECT id FROM "Project" WHERE id = ${environment.projectId} FOR UPDATE`; + ({ existingDeclarativeSchedules, preparedTasks, defaultWindowDurationSeconds, warnings } = + await prepareDeclarativeSchedules(tasks, environment, tx)); + missingSchedules = new Set(existingDeclarativeSchedules.map((schedule) => schedule.id)); + const declaredTaskIds = new Set(preparedTasks.map(({ task }) => task.id)); + quotaExclusions = { + environmentId: environment.id, + scheduleIds: existingDeclarativeSchedules + .filter((schedule) => !declaredTaskIds.has(schedule.taskIdentifier)) + .map((schedule) => schedule.id), + }; + await validate(tx); + await apply(tx); }, - }); + { timeout: 120_000 } + ); + } else { + await apply(prisma); + } + + for (const registration of registrations) { + await engine.registerNextTaskScheduleInstance(registration); } return warnings; diff --git a/apps/webapp/test/fixtures/otlpHangWorker.cjs b/apps/webapp/test/fixtures/otlpHangWorker.cjs new file mode 100644 index 00000000000..d7d90376721 --- /dev/null +++ b/apps/webapp/test/fixtures/otlpHangWorker.cjs @@ -0,0 +1,6 @@ +/** + * Real worker that never replies, to exercise the pool's stuck-worker reap path. + */ +const { parentPort } = require("node:worker_threads"); + +parentPort.on("message", () => {}); diff --git a/apps/webapp/test/fixtures/otlpSlowWorker.cjs b/apps/webapp/test/fixtures/otlpSlowWorker.cjs new file mode 100644 index 00000000000..e5bd1566c9d --- /dev/null +++ b/apps/webapp/test/fixtures/otlpSlowWorker.cjs @@ -0,0 +1,22 @@ +/** + * Real worker that takes a fixed wall-clock time per task, so pool tests can drive queue wait past + * the task deadline while the worker itself stays healthy. Delay comes from + * OTLP_SLOW_WORKER_DELAY_MS (default 100). When OTLP_SLOW_WORKER_HANG_AFTER is set the worker + * answers that many tasks and then never replies again, to model a worker that wedges mid-run. + */ +const { parentPort } = require("node:worker_threads"); + +const delayMs = Number(process.env.OTLP_SLOW_WORKER_DELAY_MS ?? 100); +const hangAfter = process.env.OTLP_SLOW_WORKER_HANG_AFTER + ? Number(process.env.OTLP_SLOW_WORKER_HANG_AFTER) + : Infinity; +let answered = 0; + +parentPort.on("message", (message) => { + if (message && message.type === "pricing") return; + if (answered >= hangAfter) return; + answered++; + setTimeout(() => { + parentPort.postMessage({ id: message.id, ok: true, result: { rows: [] }, computeMs: delayMs }); + }, delayMs); +}); diff --git a/apps/webapp/test/otlpWorkerPoolTimeout.test.ts b/apps/webapp/test/otlpWorkerPoolTimeout.test.ts new file mode 100644 index 00000000000..1e4a9c6d81f --- /dev/null +++ b/apps/webapp/test/otlpWorkerPoolTimeout.test.ts @@ -0,0 +1,292 @@ +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { OtlpWorkerPool, type ReapEvent } from "~/v3/otlpWorkerPool.server"; +import { createInMemoryMetrics } from "./utils/tracing"; +import { gaugeValue, histogramCount, latestMetrics, metricSum } from "./otlpMetrics.helpers"; + +const slowWorker = fileURLToPath(new URL("./fixtures/otlpSlowWorker.cjs", import.meta.url)); +const hangWorker = fileURLToPath(new URL("./fixtures/otlpHangWorker.cjs", import.meta.url)); + +const config = { spanAttributeValueLengthLimit: 8192, defaultEventStore: "clickhouse" }; +const payload = () => new Uint8Array([1, 2, 3, 4]); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe("OtlpWorkerPool task timeouts", () => { + const cleanups: Array<() => Promise> = []; + + afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup(); + } + delete process.env.OTLP_SLOW_WORKER_DELAY_MS; + delete process.env.OTLP_SLOW_WORKER_HANG_AFTER; + }); + + it("keeps a healthy worker alive when queued tasks exceed their deadline under overload", async () => { + const taskTimeoutMs = 500; + const computeMs = 20; + const arrivalIntervalMs = 10; + const runMs = 3000; + process.env.OTLP_SLOW_WORKER_DELAY_MS = String(computeMs); + + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + let resolved = 0; + let rejected = 0; + const pending: Promise[] = []; + const aliveSamples: Array = []; + const okSamples: number[] = []; + + const startedAt = Date.now(); + const enqueue = setInterval(() => { + const p = pool + .runTransform("traces", payload(), config) + .then(() => resolved++) + .catch(() => rejected++); + pending.push(p); + }, arrivalIntervalMs); + + let lastSampleAt = startedAt; + while (Date.now() - startedAt < runMs) { + await sleep(25); + if (Date.now() - startedAt >= 1000 && Date.now() - lastSampleAt >= 200) { + lastSampleAt = Date.now(); + const rm = await latestMetrics(metrics); + aliveSamples.push(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })); + okSamples.push(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "ok" })); + } + } + clearInterval(enqueue); + await Promise.all(pending); + + const rm = await latestMetrics(metrics); + const summary = { + resolved, + rejected, + respawnsTimeout: metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" }), + respawnsError: metricSum(rm, "ingest.worker_pool.respawns", { reason: "error" }), + respawnsExit: metricSum(rm, "ingest.worker_pool.respawns", { reason: "exit" }), + crash: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" }), + ok: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "ok" }), + timeout: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "timeout" }), + stale: metricSum(rm, "ingest.worker_pool.tasks", { outcome: "stale" }), + aliveSamples, + okSamples, + }; + console.log("overload summary", JSON.stringify(summary)); + + expect(summary.timeout + summary.stale).toBeGreaterThan(0); + expect(rejected).toBeGreaterThan(0); + + expect(summary.respawnsTimeout).toBe(0); + expect(summary.respawnsError).toBe(0); + expect(summary.respawnsExit).toBe(0); + expect(summary.crash).toBe(0); + expect(aliveSamples.length).toBeGreaterThan(0); + expect(aliveSamples.every((alive) => alive === 1)).toBe(true); + expect(summary.ok).toBeGreaterThanOrEqual(50); + expect(summary.ok).toBeGreaterThan(okSamples[0]!); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + }, + { timeout: 2000, interval: 50 } + ); + }, 15_000); + + it("lets a healthy worker finish a task whose caller deadline passed and reuses it", async () => { + process.env.OTLP_SLOW_WORKER_DELAY_MS = "200"; + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs: 300, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + await pool.runTransform("traces", payload(), config); + + const results = await Promise.allSettled([ + pool.runTransform("traces", payload(), config), + pool.runTransform("traces", payload(), config), + pool.runTransform("traces", payload(), config), + ]); + expect(results[0]!.status).toBe("fulfilled"); + expect(results[1]!.status).toBe("rejected"); + expect((results[1] as PromiseRejectedResult).reason.message).toMatch(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 2000, interval: 20 } + ); + await expect(pool.runTransform("traces", payload(), config)).resolves.toBeDefined(); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns")).toBe(0); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" })).toBe(0); + expect(histogramCount(rm, "ingest.worker_pool.task.duration")).toBe(5); + expect(histogramCount(rm, "ingest.worker_pool.compute.duration")).toBe(4); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("reaps a worker that hangs only after a full compute budget, not at the caller deadline", async () => { + const taskTimeoutMs = 300; + process.env.OTLP_SLOW_WORKER_DELAY_MS = "100"; + process.env.OTLP_SLOW_WORKER_HANG_AFTER = "2"; + const reaps: ReapEvent[] = []; + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, slowWorker, [], metrics.meter, { + taskTimeoutMs, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: (event) => reaps.push(event), + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + await pool.runTransform("traces", payload(), config); + + const first = pool.runTransform("traces", payload(), config); + const second = pool.runTransform("traces", payload(), config); + await expect(first).resolves.toBeDefined(); + await expect(second).rejects.toThrow(/timed out/); + const rejectedAt = Date.now(); + + await vi.waitFor( + () => { + expect(reaps.some((event) => event.reason === "timeout")).toBe(true); + }, + { timeout: 2000, interval: 10 } + ); + const reapedAt = Date.now(); + const reap = reaps.find((event) => event.reason === "timeout")!; + expect(reap.sinceDispatchMs).toBeGreaterThanOrEqual(taskTimeoutMs); + expect(reap.sinceDispatchMs).toBeLessThan(taskTimeoutMs + 150); + expect(reap.taskAgeMs).toBeGreaterThan(reap.sinceDispatchMs!); + expect(reapedAt - rejectedAt).toBeGreaterThanOrEqual(50); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.respawns")).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("keeps reaping and respawning when the onReap observer throws", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: () => { + throw new Error("observer boom"); + }, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("keeps reaping and respawning when the onReap observer rejects asynchronously", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + onReap: async () => { + throw new Error("async observer boom"); + }, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); + + it("reaps and respawns a worker that never replies", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, hangWorker, [], metrics.meter, { + taskTimeoutMs: 200, + respawnBaseMs: 50, + respawnMaxMs: 200, + }); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/timed out/); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.respawns", { reason: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "timeout" })).toBe(1); + expect(metricSum(rm, "ingest.worker_pool.tasks", { outcome: "crash" })).toBe(0); + }, + { timeout: 3000, interval: 50 } + ); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(1); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "idle" })).toBe(1); + }, + { timeout: 3000, interval: 50 } + ); + }); +}); diff --git a/apps/webapp/test/runEngineHandlers.test.ts b/apps/webapp/test/runEngineHandlers.test.ts index 63f7e88505f..d9a25dd1135 100644 --- a/apps/webapp/test/runEngineHandlers.test.ts +++ b/apps/webapp/test/runEngineHandlers.test.ts @@ -117,7 +117,6 @@ async function seedBatch( function makeBatchDeps( overrides: { splitEnabled?: boolean; - newReplica?: PrismaClient; newWriter?: PrismaClient; legacyWriter?: PrismaClient; legacyReplica?: PrismaClient; @@ -127,7 +126,6 @@ function makeBatchDeps( const tryCompleteBatchCalls: string[] = []; return { splitEnabled: overrides.splitEnabled ?? false, - newReplica: (overrides.newReplica ?? single)!, newWriter: (overrides.newWriter ?? single)!, legacyWriter: (overrides.legacyWriter ?? single)!, tryCompleteBatch: async (batchId: string) => { @@ -508,7 +506,6 @@ describe("runEngineHandlers batch residency routing", () => { const shards = [{ key: "a", writer: prisma14 }] as const; const writer = await resolveBatchRunOpsWriter(gen2BatchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma17, shards: shards as never, @@ -526,7 +523,6 @@ describe("runEngineHandlers batch residency routing", () => { }, { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma17, shards: shards as never, @@ -555,13 +551,6 @@ describe("runEngineHandlers batch residency routing", () => { const shardWriter = {} as never; const writer = await resolveBatchRunOpsWriter(`${"a".repeat(24)}a2`, { - newReplica: { - batchTaskRun: { - findFirst: async () => { - throw new Error("a gen-2 batch id must never probe the NEW store"); - }, - }, - } as never, newWriter: {} as never, legacyWriter: {} as never, shards: [{ key: "a", writer: shardWriter as never }], @@ -573,7 +562,6 @@ describe("runEngineHandlers batch residency routing", () => { it("an unconfigured shard key fails loud rather than writing elsewhere", async () => { await expect( resolveBatchRunOpsWriter(`${"a".repeat(24)}z2`, { - newReplica: {} as never, newWriter: {} as never, legacyWriter: {} as never, shards: [{ key: "a", writer: {} as never }], @@ -581,7 +569,7 @@ describe("runEngineHandlers batch residency routing", () => { ).rejects.toThrow(/shard/i); }); - // True single-DB invariant: the topology's cpFallback makes newReplica and + // True single-DB invariant: the topology's cpFallback makes newWriter and // legacyWriter the SAME control-plane client, so the probe always resolves to // that one client regardless of where length-classification would guess. containerTest("true single-DB resolves to the single client", async ({ prisma }) => { @@ -594,7 +582,6 @@ describe("runEngineHandlers batch residency routing", () => { }); const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma, newWriter: prisma, legacyWriter: prisma, }); @@ -616,7 +603,6 @@ describe("runEngineHandlers batch residency routing", () => { // The probe misses on new (the new DB has no such batch) and resolves the legacy writer. const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, }); @@ -624,7 +610,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, @@ -674,7 +659,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: false, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, @@ -721,7 +705,6 @@ describe("runEngineHandlers batch residency routing", () => { }); const writer = await resolveBatchRunOpsWriter(batchId, { - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, }); @@ -729,7 +712,6 @@ describe("runEngineHandlers batch residency routing", () => { const deps: BatchCompletionDeps = { splitEnabled: true, - newReplica: prisma17, newWriter: prisma17, legacyWriter: prisma14, tryCompleteBatch: async () => {}, diff --git a/apps/webapp/test/scheduleTimings.test.ts b/apps/webapp/test/scheduleTimings.test.ts index 5a3601f9ff9..d98651697ec 100644 --- a/apps/webapp/test/scheduleTimings.test.ts +++ b/apps/webapp/test/scheduleTimings.test.ts @@ -4,9 +4,9 @@ import { SCHEDULE_PHASE_DENOMINATOR, calculateEffectiveScheduleTime, calculateSchedulePhase, + resolveScheduleWindow, } from "@internal/schedule-engine"; import { CronPattern } from "~/v3/schedules"; -import { type NormalizedScheduleWindow } from "@trigger.dev/core/v3"; import { parseExpression } from "cron-parser"; import { describe, expect, it } from "vitest"; import { @@ -77,24 +77,48 @@ function referenceResolve( deduplicationKey: input.deduplicationKey, }); - const window: NormalizedScheduleWindow | undefined = - input.windowPercentage !== null - ? { type: "percentage", percentage: input.windowPercentage } - : input.windowDurationSeconds !== null - ? { type: "duration", durationSeconds: input.windowDurationSeconds } - : undefined; + const window = resolveScheduleWindow({ + windowDurationSeconds: input.windowDurationSeconds, + windowPercentage: input.windowPercentage, + defaultWindowDurationSeconds: input.defaultWindowDurationSeconds, + }).window; const { effectiveAt } = calculateEffectiveScheduleTime({ nominalAt: nominalTimes[0], nextNominalAt: nominalTimes[1], schedulePhase: phase, window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, }); let lastRun: Date | undefined; if (includeLastRun && input.active) { try { - const previous = previousScheduledTimestamp(input.cron, input.timezone, now); + const latestNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + new Date(now.getTime() + 1) + ); + const previousNominal = previousScheduledTimestamp( + input.cron, + input.timezone, + latestNominal + ); + const latestEffective = calculateEffectiveScheduleTime({ + nominalAt: latestNominal, + nextNominalAt: nominalTimes[0], + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + const previousEffective = calculateEffectiveScheduleTime({ + nominalAt: previousNominal, + nextNominalAt: latestNominal, + schedulePhase: phase, + window, + minimumWindowDurationSeconds: input.minimumWindowDurationSeconds, + }).effectiveAt; + const previous = latestEffective <= now ? latestEffective : previousEffective; lastRun = previous.getTime() > input.updatedAt.getTime() ? previous : undefined; } catch { lastRun = undefined; @@ -255,15 +279,24 @@ describe("resolveScheduleTimings", () => { }); it("skips lastRun when the previous slot predates the last config change", () => { - const [stale] = resolveScheduleTimings([input({ cron: "0 0 * * *", updatedAt: now })], { - phaseSecret: PHASE_SECRET, - includeLastRun: true, - now, - }); + const [stale] = resolveScheduleTimings( + [input({ cron: "0 0 * * *", schedulePhase: 0, updatedAt: now })], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now, + } + ); expect(stale.lastRun).toBeUndefined(); const [fresh] = resolveScheduleTimings( - [input({ cron: "0 0 * * *", updatedAt: new Date("2020-01-01T00:00:00.000Z") })], + [ + input({ + cron: "0 0 * * *", + schedulePhase: 0, + updatedAt: new Date("2020-01-01T00:00:00.000Z"), + }), + ], { phaseSecret: PHASE_SECRET, includeLastRun: true, now } ); expect(fresh.lastRun).toEqual(new Date("2024-06-15T00:00:00.000Z")); @@ -282,7 +315,7 @@ describe("resolveScheduleTimings", () => { }); it("resolves lastRun for a valid expression", () => { - const [valid] = resolveScheduleTimings([input({ cron: "0 0 * * *" })], { + const [valid] = resolveScheduleTimings([input({ cron: "0 0 * * *", schedulePhase: 0 })], { phaseSecret: PHASE_SECRET, includeLastRun: true, now, @@ -291,6 +324,56 @@ describe("resolveScheduleTimings", () => { expect(valid.lastRun).toEqual(new Date("2024-06-15T00:00:00.000Z")); }); + it("resolves lastRun from the most recent effective window time", () => { + const [beforeCurrentWindow] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + }), + ], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now, + } + ); + const [afterCurrentWindow] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + }), + ], + { + phaseSecret: PHASE_SECRET, + includeLastRun: true, + now: new Date("2024-06-15T09:40:00.000Z"), + } + ); + + expect(beforeCurrentWindow.lastRun).toEqual(new Date("2024-06-15T08:30:00.000Z")); + expect(afterCurrentWindow.lastRun).toEqual(new Date("2024-06-15T09:30:00.000Z")); + }); + + it("compares updatedAt with the effective window time", () => { + const [timing] = resolveScheduleTimings( + [ + input({ + cron: "0 * * * *", + schedulePhase: SCHEDULE_PHASE_DENOMINATOR / 2, + windowPercentage: 100, + updatedAt: new Date("2024-06-15T08:15:00.000Z"), + }), + ], + { phaseSecret: PHASE_SECRET, includeLastRun: true, now } + ); + + expect(timing.lastRun).toEqual(new Date("2024-06-15T08:30:00.000Z")); + }); + it("honours a caller-supplied schedulePhase over the derived one", () => { const rows = [input({ schedulePhase: 0, windowPercentage: 100, cron: "0 * * * *" })]; diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 9ca98e1d405..ac3cf2f9527 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -144,12 +144,13 @@ services: # routes to it when the OBJECT_STORE_* env vars are set (see .env.example). minio: container_name: ${CONTAINER_PREFIX:-}minio - image: quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e + image: chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b restart: always command: server /data --console-address ":9001" environment: MINIO_ROOT_USER: minioadmin MINIO_ROOT_PASSWORD: minioadmin + MC_HOST_local: http://minioadmin:minioadmin@localhost:9000 volumes: - minio-data:/data ports: @@ -158,25 +159,22 @@ services: networks: - app_network healthcheck: - test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] + test: ["CMD", "mc", "ready", "local"] interval: 5s timeout: 5s retries: 5 start_period: 5s minio-init: - image: quay.io/minio/mc:RELEASE.2025-08-13T08-35-41Z@sha256:a7fe349ef4bd8521fb8497f55c6042871b2ae640607cf99d9bede5e9bdf11727 + image: chainguard/minio-client:latest@sha256:b2bd7824d23d3e3b15bedd7e87fbc3be29d2e213307b4f901e4a1d92356dc20f depends_on: minio: condition: service_healthy networks: - app_network - entrypoint: /bin/sh - command: - - -c - - | - mc alias set local http://minio:9000 minioadmin minioadmin - mc mb -p local/packets || true + environment: + MC_HOST_local: http://minioadmin:minioadmin@minio:9000 + command: ["mb", "--ignore-existing", "local/packets"] restart: "no" electric: diff --git a/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts b/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts index defd36f6147..44964d46bd0 100644 --- a/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts +++ b/internal-packages/run-engine/src/engine/systems/batchSystem.test.ts @@ -113,7 +113,8 @@ class CountingPostgresRunStore extends PostgresRunStore { async function driveBatchToAllChildrenComplete( engine: RunEngine, prisma: PrismaClient, - friendlyPrefix: string + friendlyPrefix: string, + completeChildren = true ) { const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION"); const parentTask = "parent-task"; @@ -183,7 +184,7 @@ async function driveBatchToAllChildrenComplete( queue: `task/${childTask}`, isTest: false, tags: [], - resumeParentOnCompletion: true, + resumeParentOnCompletion: completeChildren, parentTaskRunId: parentRun.id, batch: { id: batch.id, index: 0 }, }, @@ -206,39 +207,41 @@ async function driveBatchToAllChildrenComplete( queue: `task/${childTask}`, isTest: false, tags: [], - resumeParentOnCompletion: true, + resumeParentOnCompletion: completeChildren, parentTaskRunId: parentRun.id, batch: { id: batch.id, index: 1 }, }, prisma ); - for (const child of [child1, child2]) { + if (completeChildren) { + for (const child of [child1, child2]) { + await setTimeout(500); + const dequeued = await engine.dequeueFromWorkerQueue({ + consumerId: "test_consumer", + workerQueue: "main", + }); + const match = dequeued.find((d) => d.run.id === child.id) ?? dequeued[0]; + assertNonNullable(match); + const attempt = await engine.startRunAttempt({ + runId: match.run.id, + snapshotId: match.snapshot.id, + }); + await engine.completeRunAttempt({ + runId: attempt.run.id, + snapshotId: attempt.snapshot.id, + completion: { + id: attempt.run.id, + ok: true, + output: '{"foo":"bar"}', + outputType: "application/json", + }, + }); + } + await setTimeout(500); - const dequeued = await engine.dequeueFromWorkerQueue({ - consumerId: "test_consumer", - workerQueue: "main", - }); - const match = dequeued.find((d) => d.run.id === child.id) ?? dequeued[0]; - assertNonNullable(match); - const attempt = await engine.startRunAttempt({ - runId: match.run.id, - snapshotId: match.snapshot.id, - }); - await engine.completeRunAttempt({ - runId: attempt.run.id, - snapshotId: attempt.snapshot.id, - completion: { - id: attempt.run.id, - ok: true, - output: '{"foo":"bar"}', - outputType: "application/json", - }, - }); } - await setTimeout(500); - return { environment, batch, parentRun, child1, child2 }; } @@ -308,6 +311,57 @@ describe("RunEngine #tryCompleteBatch store routing", () => { } ); + containerTest( + "retries waitpoint completion for a completed batch whose parent was not resumed", + async ({ prisma, redisOptions }) => { + const engine = new RunEngine(createEngineOptions(redisOptions, prisma)); + + try { + const { batch, parentRun } = await driveBatchToAllChildrenComplete( + engine, + prisma, + "run_batch_completion_recovery", + false + ); + + const waitpoint = await prisma.waitpoint.findFirstOrThrow({ + where: { completedByBatchId: batch.id }, + }); + expect(waitpoint.status).toBe("PENDING"); + expect( + await prisma.taskRunWaitpoint.count({ + where: { taskRunId: parentRun.id, waitpointId: waitpoint.id }, + }) + ).toBe(1); + + await prisma.batchTaskRun.update({ + where: { id: batch.id }, + data: { status: "COMPLETED", resumedAt: null }, + }); + + await engine.batchSystem.performCompleteBatch({ batchId: batch.id }); + + const recoveredBatch = await prisma.batchTaskRun.findFirstOrThrow({ + where: { id: batch.id }, + }); + expect(recoveredBatch.resumedAt).not.toBeNull(); + + const recoveredWaitpoint = await prisma.waitpoint.findFirstOrThrow({ + where: { id: waitpoint.id }, + }); + expect(recoveredWaitpoint.status).toBe("COMPLETED"); + + await setTimeout(1_000); + expect(await prisma.taskRunWaitpoint.count({ where: { taskRunId: parentRun.id } })).toBe(0); + const parentExecution = await engine.getRunExecutionData({ runId: parentRun.id }); + assertNonNullable(parentExecution); + expect(parentExecution.snapshot.executionStatus).not.toBe("EXECUTING_WITH_WAITPOINTS"); + } finally { + await engine.quit(); + } + } + ); + // The member-run read is driven by batchId only and does not rely on the // BatchTaskRun.runtimeEnvironmentId FK. A second batch (distinct batchId) must not leak members // into the first batch's batchId-scoped read. diff --git a/internal-packages/run-engine/src/engine/systems/batchSystem.ts b/internal-packages/run-engine/src/engine/systems/batchSystem.ts index 6b100c5f82c..dbd9149e1ff 100644 --- a/internal-packages/run-engine/src/engine/systems/batchSystem.ts +++ b/internal-packages/run-engine/src/engine/systems/batchSystem.ts @@ -170,7 +170,17 @@ export class BatchSystem { } if (batch.status === "COMPLETED") { - this.$.logger.debug("#tryCompleteBatch: Batch already completed", { batchId }); + if (batch.resumedAt) { + this.$.logger.debug("#tryCompleteBatch: Batch already completed and resumed", { + batchId, + }); + return; + } + + this.$.logger.debug("#tryCompleteBatch: Batch completed before its parent resumed", { + batchId, + }); + await this.#resumeBatchParent(batchId); return; } @@ -230,33 +240,40 @@ export class BatchSystem { return; } - //get waitpoint (if there is one) - const waitpoint = await this.$.runStore.findWaitpoint( - { - where: { - completedByBatchId: batchId, - }, - }, - this.$.prisma - ); - - if (!waitpoint) { - this.$.logger.debug( - "RunEngine.unblockRunForBatch(): Waitpoint not found. This is ok, because only batchTriggerAndWait has waitpoints", - { - batchId, - } - ); - return; - } - - await this.waitpointSystem.completeWaitpoint({ - id: waitpoint.id, - output: { value: "Batch waitpoint completed", isError: false }, - }); + await this.#resumeBatchParent(batchId); } else { this.$.logger.debug("#tryCompleteBatch: Not all runs are completed", { batchId }); } }); } + + async #resumeBatchParent(batchId: string): Promise { + const waitpoint = await this.$.runStore.findWaitpoint( + { + where: { completedByBatchId: batchId }, + }, + this.$.prisma + ); + + if (!waitpoint) { + this.$.logger.debug( + "RunEngine.unblockRunForBatch(): Waitpoint not found. This is ok, because only batchTriggerAndWait has waitpoints", + { batchId } + ); + return; + } + + await this.waitpointSystem.completeWaitpoint({ + id: waitpoint.id, + output: { value: "Batch waitpoint completed", isError: false }, + }); + + await this.$.runStore.updateManyBatchTaskRun( + { + where: { id: batchId, resumedAt: null }, + data: { resumedAt: new Date() }, + }, + this.$.prisma + ); + } } diff --git a/internal-packages/testcontainers/src/minio.ts b/internal-packages/testcontainers/src/minio.ts index 6d158d99de2..e96391c7713 100644 --- a/internal-packages/testcontainers/src/minio.ts +++ b/internal-packages/testcontainers/src/minio.ts @@ -17,7 +17,7 @@ export class MinIOContainer extends GenericContainer { private region = "us-east-1"; constructor( - image = "quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e" + image = "chainguard/minio:latest@sha256:6a1d0b45c8669726bba580ced0bfa4cb9fdeed1ed636dfabd81d1577beb6937b" ) { super(image); this.withExposedPorts(MINIO_PORT); @@ -49,13 +49,14 @@ export class MinIOContainer extends GenericContainer { const startedContainer = await super.start(); - // Create the "packets" bucket using MinIO client await x( "docker", [ "exec", startedContainer.getId(), "mc", + "--config-dir", + "/tmp/.mc", "alias", "set", "local", @@ -66,9 +67,11 @@ export class MinIOContainer extends GenericContainer { { throwOnError: true } ); - await x("docker", ["exec", startedContainer.getId(), "mc", "mb", "local/packets"], { - throwOnError: true, - }); + await x( + "docker", + ["exec", startedContainer.getId(), "mc", "--config-dir", "/tmp/.mc", "mb", "local/packets"], + { throwOnError: true } + ); return new StartedMinIOContainer( startedContainer, @@ -123,14 +126,35 @@ export class StartedMinIOContainer extends AbstractStartedContainer { public async resetBucket(bucket = "packets"): Promise { await x( "docker", - ["exec", this.getId(), "mc", "rm", "--recursive", "--force", `local/${bucket}`], + [ + "exec", + this.getId(), + "mc", + "--config-dir", + "/tmp/.mc", + "rm", + "--recursive", + "--force", + `local/${bucket}`, + ], { throwOnError: false, } ); - await x("docker", ["exec", this.getId(), "mc", "mb", "--ignore-existing", `local/${bucket}`], { - throwOnError: true, - }); + await x( + "docker", + [ + "exec", + this.getId(), + "mc", + "--config-dir", + "/tmp/.mc", + "mb", + "--ignore-existing", + `local/${bucket}`, + ], + { throwOnError: true } + ); } /** diff --git a/packages/cli-v3/src/build/buildWorker.test.ts b/packages/cli-v3/src/build/buildWorker.test.ts new file mode 100644 index 00000000000..90f828c014e --- /dev/null +++ b/packages/cli-v3/src/build/buildWorker.test.ts @@ -0,0 +1,31 @@ +import { join, posix, sep, win32 } from "node:path"; +import { describe, expect, it } from "vitest"; +import { rewriteOutputPath } from "./buildWorker.js"; + +describe("rewriteOutputPath", () => { + it("defaults to the host platform's path semantics", () => { + const out = join(sep, "tmp", "out"); + expect(rewriteOutputPath(out, join(out, "src", "a b.mjs"))).toBe("/app/src/a b.mjs"); + }); + + it("maps a bundled file to its /app path on posix", () => { + expect(rewriteOutputPath("/tmp/out", "/tmp/out/src/trigger/a.mjs", posix)).toBe( + "/app/src/trigger/a.mjs" + ); + }); + + it("uses forward slashes and keeps spaces and non-ASCII characters on windows", () => { + const out = "C:\\Users\\me\\AppData\\Local\\Temp\\trigger-build"; + const file = `${out}\\Documents\\DATEN VORBEREITUNG FÜR AUKTIONEN\\trigger.config.mjs`; + + expect(rewriteOutputPath(out, file, win32)).toBe( + "/app/Documents/DATEN VORBEREITUNG FÜR AUKTIONEN/trigger.config.mjs" + ); + }); + + it("tolerates mixed separators and drive letter case on windows", () => { + expect( + rewriteOutputPath("c:\\Users\\me\\out", "C:/Users/me/out/Trigger Demo/index.mjs", win32) + ).toBe("/app/Trigger Demo/index.mjs"); + }); +}); diff --git a/packages/cli-v3/src/build/buildWorker.ts b/packages/cli-v3/src/build/buildWorker.ts index 2aee0192af7..6ad0da8ba07 100644 --- a/packages/cli-v3/src/build/buildWorker.ts +++ b/packages/cli-v3/src/build/buildWorker.ts @@ -22,15 +22,13 @@ import { import { createExternalsBuildExtension } from "./externals.js"; import { tmpdir } from "node:os"; import { mkdtemp, rm } from "node:fs/promises"; -import { join, relative, sep } from "node:path"; +import path, { join } from "node:path"; import { generateContainerfile } from "../deploy/buildImage.js"; import { writeFile } from "node:fs/promises"; import { buildManifestToJSON } from "../utilities/buildManifest.js"; import { logger } from "../utilities/logger.js"; import { readPackageJSON } from "pkg-types"; import { writeJSONFile } from "../utilities/fileSystem.js"; -import { isWindows } from "std-env"; -import { pathToFileURL } from "node:url"; import { logBuildWorkerStart } from "./buildWorkerLogging.js"; import { SdkVersionExtractor } from "./plugins.js"; import { spinner } from "../utilities/windows.js"; @@ -215,17 +213,12 @@ function cleanEntryPath(entry: string): string { return entry.split("?")[0]!; } -function rewriteOutputPath(destinationDir: string, filePath: string) { - if (isWindows) { - return `/app/${relative( - pathToFileURL(destinationDir).pathname, - pathToFileURL(filePath).pathname - ) - .split(sep) - .join("/")}`; - } else { - return `/app/${relative(destinationDir, filePath)}`; - } +export function rewriteOutputPath( + destinationDir: string, + filePath: string, + pathApi: Pick = path +) { + return `/app/${pathApi.relative(destinationDir, filePath).split(pathApi.sep).join("/")}`; } async function writeDeployFiles({ diff --git a/packages/cli-v3/src/deploy/buildImage.test.ts b/packages/cli-v3/src/deploy/buildImage.test.ts index cb7c78242ba..6b25bbf89dc 100644 --- a/packages/cli-v3/src/deploy/buildImage.test.ts +++ b/packages/cli-v3/src/deploy/buildImage.test.ts @@ -62,6 +62,23 @@ describe("generateContainerfile", () => { expect(containerfile).toContain(`FROM ${build} AS build`); }); + it.each(["node", "bun"] as BuildRuntime[])( + "runs the indexer in exec form so entry paths with spaces survive on %s", + async (runtime) => { + const indexScript = "/app/Documents/Trigger Demo/index.mjs"; + const containerfile = await generateContainerfile({ + runtime, + build: {}, + image: undefined, + indexScript, + entrypoint: "entrypoint.js", + }); + + expect(containerfile).toContain(`"${indexScript}" ]`); + expect(containerfile).not.toMatch(/^RUN (node|bun run) \//m); + } + ); + it.each(["node", "bun"] as BuildRuntime[])( "runs no package installation for uncustomized projects on %s", async (runtime) => { diff --git a/packages/cli-v3/src/deploy/buildImage.ts b/packages/cli-v3/src/deploy/buildImage.ts index 0505698e2f3..40b82717cbd 100644 --- a/packages/cli-v3/src/deploy/buildImage.ts +++ b/packages/cli-v3/src/deploy/buildImage.ts @@ -892,7 +892,7 @@ ARG BUILDPLATFORM ENV BUILDPLATFORM=$BUILDPLATFORM TARGETPLATFORM=$TARGETPLATFORM # Run the indexer -RUN bun run ${options.indexScript} +RUN [ "bun", "run", "${options.indexScript}" ] # Development or production stage builds upon the base stage FROM base AS final @@ -1001,7 +1001,7 @@ ARG BUILDPLATFORM ENV BUILDPLATFORM=$BUILDPLATFORM TARGETPLATFORM=$TARGETPLATFORM # Run the indexer -RUN node ${options.indexScript} +RUN [ "node", "${options.indexScript}" ] # Development or production stage builds upon the base stage FROM base AS final diff --git a/packages/core/src/v3/zodIpc.ts b/packages/core/src/v3/zodIpc.ts index ebd5a685df7..e2b5028b376 100644 --- a/packages/core/src/v3/zodIpc.ts +++ b/packages/core/src/v3/zodIpc.ts @@ -124,7 +124,9 @@ const Packet = discriminatedUnion("type", [ }), z.object({ type: z.literal("ACK"), - message: z.any(), + // Optional: a void callback acks with `message: undefined`, and `process.send` + // drops undefined keys when serializing, so the key is absent on arrival. + message: z.any().optional(), id: z.number(), }), z.object({ diff --git a/packages/core/test/fixtures/zodIpcCatalog.ts b/packages/core/test/fixtures/zodIpcCatalog.ts new file mode 100644 index 00000000000..6f24288b049 --- /dev/null +++ b/packages/core/test/fixtures/zodIpcCatalog.ts @@ -0,0 +1,14 @@ +import { z } from "zod/v4"; + +export const ParentToChild = { + FLUSH: { + message: z.object({ timeoutInMs: z.number() }), + callback: z.void(), + }, + PING: { + message: z.object({ value: z.string() }), + callback: z.object({ echoed: z.string() }), + }, +}; + +export const ChildToParent = {}; diff --git a/packages/core/test/fixtures/zodIpcChild.ts b/packages/core/test/fixtures/zodIpcChild.ts new file mode 100644 index 00000000000..721b534ddb9 --- /dev/null +++ b/packages/core/test/fixtures/zodIpcChild.ts @@ -0,0 +1,13 @@ +import { ZodIpcConnection } from "../../src/v3/zodIpc.js"; +import { ChildToParent, ParentToChild } from "./zodIpcCatalog.js"; + +// Forked by zodIpc.test.ts: answers the parent's messages over the real IPC channel. +new ZodIpcConnection({ + listenSchema: ParentToChild, + emitSchema: ChildToParent, + process, + handlers: { + FLUSH: async () => {}, + PING: async ({ value }) => ({ echoed: value }), + }, +}); diff --git a/packages/core/test/zodIpc.test.ts b/packages/core/test/zodIpc.test.ts new file mode 100644 index 00000000000..7b5d2669723 --- /dev/null +++ b/packages/core/test/zodIpc.test.ts @@ -0,0 +1,44 @@ +import { fork, type ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { ZodIpcConnection } from "../src/v3/zodIpc.js"; +import { ChildToParent, ParentToChild } from "./fixtures/zodIpcCatalog.js"; + +const childPath = fileURLToPath(new URL("./fixtures/zodIpcChild.ts", import.meta.url)); + +let child: ChildProcess | undefined; + +afterEach(() => { + child?.kill(); + child = undefined; +}); + +// Node's default IPC serialization is JSON, so an ack with `message: undefined` +// arrives without the `message` key, exactly as it does between a worker and its task run process. +function forkChild() { + child = fork(childPath, { execArgv: ["--import", "tsx"], stdio: "inherit" }); + + return new ZodIpcConnection({ + listenSchema: ChildToParent, + emitSchema: ParentToChild, + process: child, + }); +} + +describe("ZodIpcConnection", () => { + it("resolves sendWithAck for a void callback", async () => { + const connection = forkChild(); + + await expect(connection.sendWithAck("FLUSH", { timeoutInMs: 1000 }, 2000)).resolves.toBe( + undefined + ); + }); + + it("resolves sendWithAck with the callback payload", async () => { + const connection = forkChild(); + + await expect(connection.sendWithAck("PING", { value: "hello" }, 2000)).resolves.toEqual({ + echoed: "hello", + }); + }); +}); diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 8113f89ee3b..0bc2e6bbb7e 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -66,6 +66,7 @@ import type { // ESM-only `ai@7` (see ../imports/ai-runtime.ts). import { type Attributes, trace } from "@opentelemetry/api"; import { traceSessionIdle } from "./sessionTracing.js"; +import { waitForChatRouteAfterIdle } from "./chatRouteWait.js"; import { tool as aiTool, convertToModelMessages, @@ -1780,31 +1781,21 @@ async function waitOnChatRoute( if (options.onSuspend) await options.onSuspend(); span.setAttribute("wait.resolved", "suspended"); - while (true) { - /** - * The floor doubles as the wake cursor: the server completes the - * waitpoint immediately if anything sits after this sequence, so a - * floor that has advanced past an unread record parks a waitpoint - * nothing will complete. Recorded on the span so a run that never woke - * can be diagnosed from its trace alone. - */ - const wakeFrom = router.resumeFloor(); - span.setAttribute("wait.lastSeqNum", wakeFrom ?? -1); - const wake = await session.in.awaitWake({ - timeout: options.timeout, - lastSeqNum: wakeFrom, - }); - if (!wake.ok) { - span.recordException(wake.error); - return { ok: false as const, error: wake.error }; - } - - const record = await router.next(route); - if (!record) continue; + const result = await waitForChatRouteAfterIdle(router, route, { + timeout: options.timeout, + wake: async (timeout, lastSeqNum) => { + span.setAttribute("wait.lastSeqNum", lastSeqNum ?? -1); + return session.in.awaitWake({ timeout, lastSeqNum }); + }, + }); - if (options.onResume) await options.onResume(); - return { ok: true as const, output: record.data as T, record }; + if (!result.ok) { + span.recordException(result.error); + return result; } + + if (options.onResume) await options.onResume(); + return { ok: true as const, output: result.record.data as T, record: result.record }; }, { attributes: { diff --git a/packages/trigger-sdk/src/v3/chatRouteWait.test.ts b/packages/trigger-sdk/src/v3/chatRouteWait.test.ts new file mode 100644 index 00000000000..5dac1ed1968 --- /dev/null +++ b/packages/trigger-sdk/src/v3/chatRouteWait.test.ts @@ -0,0 +1,101 @@ +import { SessionChannelRouter, WaitpointTimeoutError } from "@trigger.dev/core/v3"; +import { describe, expect, it } from "vitest"; +import { waitForChatRouteAfterIdle } from "./chatRouteWait.js"; + +function inputRouter() { + return new SessionChannelRouter({ + kindOf: (data) => (data as { kind?: string }).kind, + routes: [ + { name: "messages", delivery: "queue", replayable: true, kinds: ["message"] }, + { name: "control", delivery: "queue", replayable: false, kinds: ["control"] }, + ], + }); +} + +describe("chat route waits after suspension", () => { + it("re-suspends after a wake that has no record for the requested route", async () => { + const router = inputRouter(); + const timeouts: Array = []; + let now = 0; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "10s", + postWakeTimeoutMs: 0, + now: () => now, + wake: async (timeout) => { + timeouts.push(timeout); + if (timeouts.length === 1) { + router.ingest({ id: "control-1", seqNum: 1, data: { kind: "control" } }); + now = 4_000; + return { ok: true, waitpointId: "waitpoint-1" }; + } + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + }, + }); + + expect(result.ok).toBe(false); + expect(timeouts).toEqual(["10s", "6s"]); + }); + + it("does not re-wake on an unrelated queued message while waiting for another route", async () => { + const router = inputRouter(); + const cursors: Array = []; + + const result = await waitForChatRouteAfterIdle(router, "control", { + postWakeTimeoutMs: 0, + wake: async (_timeout, lastSeqNum) => { + cursors.push(lastSeqNum); + if (cursors.length === 1) { + router.ingest({ id: "message-1", seqNum: 1, data: { kind: "message" } }); + return { ok: true, waitpointId: "waitpoint-1" }; + } + if (lastSeqNum === undefined || lastSeqNum < 1) { + return { ok: true, waitpointId: "waitpoint-repeated" }; + } + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + }, + }); + + expect(result.ok).toBe(false); + expect(cursors).toEqual([undefined, 1]); + expect(router.resumeFloor()).toBe(0); + expect(router.hasPending("messages")).toBe(true); + }); + + it("expires an absolute timeout after an unmatched wake", async () => { + const router = inputRouter(); + const deadline = Date.parse("2026-09-28T10:15:00Z"); + let now = deadline - 1_000; + const timeouts: Array = []; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "2026-09-28T10:15:00Z", + postWakeTimeoutMs: 0, + now: () => now, + wake: async (timeout) => { + timeouts.push(timeout); + now = deadline + 1; + return { ok: true, waitpointId: "waitpoint-1" }; + }, + }); + + expect(result.ok).toBe(false); + expect(timeouts).toEqual(["1s"]); + }); + + it("returns a matching record delivered after the channel wakes", async () => { + const router = inputRouter(); + const message = { id: "message-1", seqNum: 1, data: { kind: "message", text: "hello" } }; + + const result = await waitForChatRouteAfterIdle(router, "messages", { + timeout: "10s", + postWakeTimeoutMs: 0, + wake: async () => { + router.ingest(message); + return { ok: true, waitpointId: "waitpoint-1" }; + }, + }); + + expect(result).toEqual({ ok: true, record: message }); + }); +}); diff --git a/packages/trigger-sdk/src/v3/chatRouteWait.ts b/packages/trigger-sdk/src/v3/chatRouteWait.ts new file mode 100644 index 00000000000..b489c25e500 --- /dev/null +++ b/packages/trigger-sdk/src/v3/chatRouteWait.ts @@ -0,0 +1,64 @@ +import { + type SessionChannelRouter, + type SessionStreamRecord, + WaitpointTimeoutError, +} from "@trigger.dev/core/v3"; +import { parseNaturalLanguageDurationInMs } from "@trigger.dev/core/v3/isomorphic"; + +const POST_WAKE_TIMEOUT_MS = 5_000; + +type ChatRouteWakeResult = { ok: true; waitpointId: string } | { ok: false; error: Error }; + +export async function waitForChatRouteAfterIdle( + router: SessionChannelRouter, + route: string, + options: { + timeout?: string; + postWakeTimeoutMs?: number; + now?: () => number; + wake: ( + timeout: string | undefined, + lastSeqNum: number | undefined + ) => Promise; + } +): Promise<{ ok: true; record: SessionStreamRecord } | { ok: false; error: Error }> { + const now = options.now ?? Date.now; + const timeoutMs = options.timeout ? parseNaturalLanguageDurationInMs(options.timeout) : undefined; + const parsedDeadline = + timeoutMs !== undefined + ? now() + timeoutMs + : options.timeout === undefined + ? undefined + : Date.parse(options.timeout); + const deadline = + parsedDeadline !== undefined && Number.isFinite(parsedDeadline) ? parsedDeadline : undefined; + const postWakeTimeoutMs = options.postWakeTimeoutMs ?? POST_WAKE_TIMEOUT_MS; + + while (true) { + const remainingMs = deadline === undefined ? undefined : deadline - now(); + if (remainingMs !== undefined && remainingMs <= 0) { + return { ok: false, error: new WaitpointTimeoutError("Timed out") }; + } + + const wakeTimeout = + remainingMs === undefined + ? options.timeout + : `${Math.max(1, Math.ceil(remainingMs / 1000))}s`; + // A queued record holds the replay floor back, but should not wake this + // waitpoint again after we have already seen it. Snapshot before checking + // the route so a record arriving during registration still wakes us. + const wakeFrom = router.appliedThrough(); + const buffered = await router.next(route, { timeoutMs: 0 }); + if (buffered) return { ok: true, record: buffered }; + + const wake = await options.wake(wakeTimeout, wakeFrom); + if (!wake.ok) return wake; + + const deliveryBudget = + deadline === undefined + ? postWakeTimeoutMs + : Math.max(0, Math.min(postWakeTimeoutMs, deadline - now())); + const record = await router.next(route, { timeoutMs: deliveryBudget }); + if (record) return { ok: true, record }; + } +}