Eric Allam
7246f677db
## What
A trigger request carrying a Unicode NUL (`U+0000`) in the **idempotency
key** or **debounce key** reached `prisma.taskRun.create()` and failed
the insert, so the caller got an opaque 500 and the run was never
created.
These two keys are stored in `jsonb` columns (`idempotencyKeyOptions`,
`debounce`), and Postgres rejects a NUL inside a `jsonb` value with
`SQLSTATE 22P05` ("unsupported Unicode escape sequence ... cannot be
converted to text"). This fix strips the NUL from both keys at the
single trigger-input chokepoint (`#buildEngineTriggerInput`), which
every trigger path flows through (single, batch item, mollified, and
drainer replay).
Stripping matches the existing precedent for run errors and task events.
It does not change dedup behaviour: the idempotency **dedup identity**
is the hashed key (a clean 64-char digest), computed independently of
the raw key we clean, so dedup keeps working exactly as before. For
debounce the key is used directly, so the cleaned key also becomes the
grouping key, an acceptable change for input that is already malformed.
## Why not payload / metadata / tags
Those are `text` columns fed by `JSON.stringify`, which escapes a NUL to
a safe escape sequence, so they do not hit this failure on the normal
JSON path. (A raw NUL in a `text` column throws a different code,
`22021`, and is not what triggers this issue.) The observed failures are
the `jsonb` `22P05` variant, which is only reachable via the two key
fields.
## Evidence
Red then green (containerTest, real Postgres): with the fix reverted,
triggering through the real service with a NUL in
`idempotencyKeyOptions.key` / `debounce.key` fails with the exact
`22P05` signature; with the fix, the run is created and the stored key
has the NUL removed.
Full-stack e2e (isolated stack, real HTTP): `POST
/api/v1/tasks/:taskId/trigger` with a NUL inside
`idempotencyKeyOptions.key` (`"acme<NUL>inc"`) and, separately,
`debounce.key` (`"grp<NUL>1"`):
- both returned `HTTP 200` with a created run (previously `500`)
- stored `idempotencyKeyOptions` = `{ "key": "acmeinc", "scope": "run"
}` (7 chars, NUL removed)
- stored `debounce.key` = `"grp1"` (4 chars, NUL removed)
- both runs render in the dashboard
Unit tests cover the helper (strip, no-op fast path, object-reference
reuse, null/undefined pass-through).
## Rollout / rollback
Server-only webapp change, no flag. Zero behaviour change for clean
input; only affects inputs that previously 500'd. Rollback is a straight
revert, no data migration.
## Known limitation
A raw NUL in a plain-string idempotency key (not created via
`idempotencyKeys.create()`) lands in a `text` column and throws `22021`
instead. That variant is not addressed here because stripping it would
change the dedup identity, so it warrants a separate decision. Not
observed in practice.
refs TRI-13030
1026 lines
42 KiB
TypeScript
1026 lines
42 KiB
TypeScript
import {
|
|
type RunEngine,
|
|
RunDuplicateIdempotencyKeyError,
|
|
RunOneTimeUseTokenError,
|
|
} from "@internal/run-engine";
|
|
import type { Tracer } from "@opentelemetry/api";
|
|
import { tryCatch } from "@trigger.dev/core/utils";
|
|
import {
|
|
type TriggerTaskRequestBody,
|
|
formatDurationMilliseconds,
|
|
RunAnnotations,
|
|
TaskRunError,
|
|
taskRunErrorEnhancer,
|
|
taskRunErrorToString,
|
|
TriggerTraceContext,
|
|
} from "@trigger.dev/core/v3";
|
|
import {
|
|
parseNaturalLanguageDurationInMs,
|
|
parseTraceparent,
|
|
RunId,
|
|
serializeTraceparent,
|
|
stringifyDuration,
|
|
} from "@trigger.dev/core/v3/isomorphic";
|
|
import type { PrismaClientOrTransaction } from "@trigger.dev/database";
|
|
import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
|
|
import { logger } from "~/services/logger.server";
|
|
import { parseDelay } from "~/utils/delays";
|
|
import { removeNullBytesFromKey } from "~/utils/nullBytes";
|
|
import { handleMetadataPacket } from "~/utils/packets";
|
|
import { startSpan } from "~/v3/tracing.server";
|
|
import { resolveRunIdMintKind } from "~/v3/engineVersion.server";
|
|
import { resolveInheritedMintKind } from "~/v3/runOpsMigration/resolveInheritedMintKind.server";
|
|
import { mintFriendlyIdForKind } from "~/v3/runOpsMigration/mintAnchoredRunFriendlyId.server";
|
|
import type {
|
|
TriggerTaskServiceOptions,
|
|
TriggerTaskServiceResult,
|
|
} from "../../v3/services/triggerTask.server";
|
|
import { clampMaxDuration } from "../../v3/utils/maxDuration";
|
|
import { clampPriorityMs } from "../../v3/utils/priority";
|
|
import {
|
|
type IdempotencyKeyConcern,
|
|
type ClaimedIdempotency,
|
|
} from "../concerns/idempotencyKeys.server";
|
|
import {
|
|
resolveScheduledQueueSplitEnabled,
|
|
workerQueueForRun,
|
|
} from "../concerns/workerQueueSplit.server";
|
|
import { resolveComputeMigration } from "../concerns/computeMigration.server";
|
|
import { workerRegionRegistry, backingForQueue, regionForQueue } from "~/v3/workerRegions.server";
|
|
import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server";
|
|
import {
|
|
publishClaim as publishMollifierClaim,
|
|
releaseClaim as releaseMollifierClaim,
|
|
} from "~/v3/mollifier/idempotencyClaim.server";
|
|
import type {
|
|
PayloadProcessor,
|
|
QueueManager,
|
|
TraceEventConcern,
|
|
TriggerRacepoints,
|
|
TriggerRacepointSystem,
|
|
TriggerTaskRequest,
|
|
TriggerTaskValidator,
|
|
} from "../types";
|
|
import { env } from "~/env.server";
|
|
import {
|
|
evaluateGate as defaultEvaluateGate,
|
|
type GateOutcome,
|
|
type MollifierEvaluateGate,
|
|
} from "~/v3/mollifier/mollifierGate.server";
|
|
import {
|
|
getMollifierBuffer as defaultGetMollifierBuffer,
|
|
type MollifierGetBuffer,
|
|
} from "~/v3/mollifier/mollifierBuffer.server";
|
|
import { mollifyTrigger } from "~/v3/mollifier/mollifierMollify.server";
|
|
import { QueueSizeLimitExceededError, ServiceValidationError } from "~/v3/services/common.server";
|
|
import { runStore } from "~/v3/runStore.server";
|
|
|
|
class NoopTriggerRacepointSystem implements TriggerRacepointSystem {
|
|
async waitForRacepoint(options: { racepoint: TriggerRacepoints; id: string }): Promise<void> {
|
|
return;
|
|
}
|
|
}
|
|
|
|
export class RunEngineTriggerTaskService {
|
|
private readonly queueConcern: QueueManager;
|
|
private readonly validator: TriggerTaskValidator;
|
|
private readonly payloadProcessor: PayloadProcessor;
|
|
private readonly idempotencyKeyConcern: IdempotencyKeyConcern;
|
|
private readonly prisma: PrismaClientOrTransaction;
|
|
private readonly engine: RunEngine;
|
|
private readonly tracer: Tracer;
|
|
private readonly traceEventConcern: TraceEventConcern;
|
|
private readonly triggerRacepointSystem: TriggerRacepointSystem;
|
|
private readonly metadataMaximumSize: number;
|
|
private readonly maximumDebounceDurationMs: number | undefined;
|
|
// Mollifier hooks are DI'd so tests can drive the call-site's mollify branch
|
|
// deterministically (stub the gate to return mollify, inject a real or fake
|
|
// buffer, force the global-enabled predicate to true so the call site
|
|
// doesn't short-circuit on an unset env). In production all three default
|
|
// to the live module-level singletons + env read.
|
|
private readonly evaluateGate: MollifierEvaluateGate;
|
|
private readonly getMollifierBuffer: MollifierGetBuffer;
|
|
private readonly isMollifierGloballyEnabled: () => boolean;
|
|
|
|
constructor(opts: {
|
|
prisma: PrismaClientOrTransaction;
|
|
engine: RunEngine;
|
|
queueConcern: QueueManager;
|
|
validator: TriggerTaskValidator;
|
|
payloadProcessor: PayloadProcessor;
|
|
idempotencyKeyConcern: IdempotencyKeyConcern;
|
|
traceEventConcern: TraceEventConcern;
|
|
tracer: Tracer;
|
|
metadataMaximumSize: number;
|
|
maximumDebounceDurationMs?: number;
|
|
triggerRacepointSystem?: TriggerRacepointSystem;
|
|
evaluateGate?: MollifierEvaluateGate;
|
|
getMollifierBuffer?: MollifierGetBuffer;
|
|
isMollifierGloballyEnabled?: () => boolean;
|
|
}) {
|
|
this.prisma = opts.prisma;
|
|
this.engine = opts.engine;
|
|
this.queueConcern = opts.queueConcern;
|
|
this.validator = opts.validator;
|
|
this.payloadProcessor = opts.payloadProcessor;
|
|
this.idempotencyKeyConcern = opts.idempotencyKeyConcern;
|
|
this.tracer = opts.tracer;
|
|
this.traceEventConcern = opts.traceEventConcern;
|
|
this.metadataMaximumSize = opts.metadataMaximumSize;
|
|
this.maximumDebounceDurationMs =
|
|
opts.maximumDebounceDurationMs ?? env.RUN_ENGINE_MAXIMUM_DEBOUNCE_DURATION_MS;
|
|
this.triggerRacepointSystem = opts.triggerRacepointSystem ?? new NoopTriggerRacepointSystem();
|
|
this.evaluateGate = opts.evaluateGate ?? defaultEvaluateGate;
|
|
this.getMollifierBuffer = opts.getMollifierBuffer ?? defaultGetMollifierBuffer;
|
|
this.isMollifierGloballyEnabled =
|
|
opts.isMollifierGloballyEnabled ?? (() => env.TRIGGER_MOLLIFIER_ENABLED === "1");
|
|
}
|
|
|
|
/**
|
|
* A debounced run is only pushed back while its new execution time stays inside the effective
|
|
* ceiling, which is the trigger's own `maxDelay` or, failing that, whatever ceiling the server
|
|
* is configured with. The room available to push is that ceiling minus `delay`, so a `delay`
|
|
* at or above it leaves none: the debounce key does nothing and every trigger creates its own
|
|
* run. Rejecting is better than accepting a trigger we know cannot debounce.
|
|
*
|
|
* With no `maxDelay` and no server ceiling there is nothing to conflict with, which is the
|
|
* default.
|
|
*/
|
|
#validateDebounceWindow(
|
|
debounce: NonNullable<NonNullable<TriggerTaskRequestBody["options"]>["debounce"]>
|
|
) {
|
|
const delayMs = parseNaturalLanguageDurationInMs(debounce.delay);
|
|
|
|
if (delayMs === undefined) {
|
|
throw new ServiceValidationError(
|
|
`Invalid debounce delay: ${debounce.delay}. debounce.delay must be a duration, not a ` +
|
|
`date, because it is re-applied every time the run is pushed back. Supported formats: ` +
|
|
`{number}s, {number}m, {number}h or {number}hr, {number}d, {number}w, optionally ` +
|
|
`combined (for example "2h30m").`
|
|
);
|
|
}
|
|
|
|
if (debounce.maxDelay !== undefined) {
|
|
const maxDelayMs = parseNaturalLanguageDurationInMs(debounce.maxDelay);
|
|
|
|
if (maxDelayMs === undefined) {
|
|
throw new ServiceValidationError(
|
|
`Invalid debounce maxDelay: ${debounce.maxDelay}. ` +
|
|
`Supported formats: {number}s, {number}m, {number}h or {number}hr, {number}d, ` +
|
|
`{number}w, optionally combined (for example "2h30m").`
|
|
);
|
|
}
|
|
|
|
if (maxDelayMs <= delayMs) {
|
|
throw new ServiceValidationError(
|
|
`debounce.maxDelay (${debounce.maxDelay}) must be longer than debounce.delay ` +
|
|
`(${debounce.delay}). A debounced run is only pushed back while it stays inside ` +
|
|
`maxDelay, so with these values every trigger would create its own run.`
|
|
);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
const serverCeilingMs = this.maximumDebounceDurationMs;
|
|
|
|
if (serverCeilingMs !== undefined && delayMs >= serverCeilingMs) {
|
|
throw new ServiceValidationError(
|
|
`debounce.delay (${debounce.delay}) is at or above this server's maximum debounce ` +
|
|
`duration of ${formatDurationMilliseconds(serverCeilingMs, { style: "short" })}. A ` +
|
|
`debounced run is only pushed back while it stays inside that window, so with this ` +
|
|
`delay every trigger would create its own run. Either shorten the delay, or set ` +
|
|
`debounce.maxDelay above ${debounce.delay}.`
|
|
);
|
|
}
|
|
}
|
|
|
|
// Mint a new run's friendlyId. The id-kind decides which store the run is born
|
|
// in (cuid → legacy store, run-ops id → new store), so the whole subgraph of a run
|
|
// must agree. Two cases:
|
|
//
|
|
// - ROOT run (no parent): mint by the environment's cutover setting.
|
|
// - CHILD run (has a parent): inherit the parent's residency by id-shape, so a
|
|
// parent and child never split across stores (run-ops parent → run-ops child,
|
|
// cuid parent → cuid child).
|
|
// `region` is the caller-requested region (body.options.region). The id is
|
|
// minted before the worker queue is resolved (the idempotency concern needs
|
|
// the friendlyId first), so the stamped region char reflects the requested
|
|
// region — or the default char when the run targets the default region.
|
|
private async mintRunFriendlyId(
|
|
environment: AuthenticatedEnvironment,
|
|
parentRunFriendlyId?: string,
|
|
region?: string
|
|
): Promise<string> {
|
|
const mintKind = parentRunFriendlyId
|
|
? resolveInheritedMintKind(parentRunFriendlyId)
|
|
: await resolveRunIdMintKind({
|
|
organizationId: environment.organizationId,
|
|
id: environment.id,
|
|
orgFeatureFlags: environment.organization.featureFlags,
|
|
});
|
|
|
|
return mintFriendlyIdForKind(mintKind, region);
|
|
}
|
|
|
|
public async call({
|
|
taskId,
|
|
environment,
|
|
body,
|
|
options = {},
|
|
attempt = 0,
|
|
}: {
|
|
taskId: string;
|
|
environment: AuthenticatedEnvironment;
|
|
body: TriggerTaskRequestBody;
|
|
options?: TriggerTaskServiceOptions;
|
|
attempt?: number;
|
|
}): Promise<TriggerTaskServiceResult | undefined> {
|
|
// Pre-gate idempotency-claim ownership. Set inside the span when
|
|
// `IdempotencyKeyConcern.handleTriggerRequest` returns `claim:
|
|
// {...}`. The try/catch below resolves it once the span finishes.
|
|
let idempotencyClaim: ClaimedIdempotency | undefined;
|
|
try {
|
|
const result = await startSpan(
|
|
this.tracer,
|
|
"RunEngineTriggerTaskService.call()",
|
|
async (span) => {
|
|
span.setAttribute("taskId", taskId);
|
|
span.setAttribute("attempt", attempt);
|
|
|
|
// Mint the run id. A caller-supplied id (idempotent retry) wins;
|
|
// otherwise mint by residency — inheriting the parent's store when a
|
|
// parent is present, else the environment's setting.
|
|
const runFriendlyId =
|
|
options?.runFriendlyId ??
|
|
(await this.mintRunFriendlyId(
|
|
environment,
|
|
body.options?.parentRunId,
|
|
body.options?.region
|
|
));
|
|
const triggerRequest = {
|
|
taskId,
|
|
friendlyId: runFriendlyId,
|
|
environment,
|
|
body,
|
|
options,
|
|
} satisfies TriggerTaskRequest;
|
|
|
|
const maxAttemptsValidation = this.validator.validateMaxAttempts({
|
|
taskId,
|
|
attempt,
|
|
});
|
|
|
|
if (!maxAttemptsValidation.ok) {
|
|
throw maxAttemptsValidation.error;
|
|
}
|
|
|
|
const tagValidation = this.validator.validateTags({
|
|
tags: body.options?.tags,
|
|
});
|
|
|
|
if (!tagValidation.ok) {
|
|
throw tagValidation.error;
|
|
}
|
|
|
|
let planType: string | undefined;
|
|
|
|
if (!options.skipChecks) {
|
|
const entitlementValidation = await this.validator.validateEntitlement({
|
|
environment,
|
|
});
|
|
|
|
if (!entitlementValidation.ok) {
|
|
throw entitlementValidation.error;
|
|
}
|
|
|
|
planType = entitlementValidation.plan?.type;
|
|
} else {
|
|
// When skipChecks is enabled, planType should be passed via options
|
|
planType = options.planType;
|
|
|
|
if (!planType) {
|
|
logger.warn("Plan type not set but skipChecks is enabled", {
|
|
taskId,
|
|
environment: {
|
|
id: environment.id,
|
|
type: environment.type,
|
|
projectId: environment.projectId,
|
|
organizationId: environment.organizationId,
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
// Parse delay from either explicit delay option or debounce.delay
|
|
const delaySource = body.options?.delay ?? body.options?.debounce?.delay;
|
|
const [parseDelayError, delayUntil] = await tryCatch(parseDelay(delaySource));
|
|
|
|
if (parseDelayError) {
|
|
throw new ServiceValidationError(`Invalid delay ${delaySource}`);
|
|
}
|
|
|
|
// Validate debounce options
|
|
if (body.options?.debounce) {
|
|
if (!delayUntil) {
|
|
throw new ServiceValidationError(
|
|
`Debounce requires a valid delay duration. Provided: ${body.options.debounce.delay}`
|
|
);
|
|
}
|
|
|
|
// Always validate debounce.delay separately since it's used for rescheduling
|
|
// This catches the case where options.delay is valid but debounce.delay is invalid
|
|
const [debounceDelayError, debounceDelayUntil] = await tryCatch(
|
|
parseDelay(body.options.debounce.delay)
|
|
);
|
|
|
|
if (debounceDelayError || !debounceDelayUntil) {
|
|
throw new ServiceValidationError(
|
|
`Invalid debounce delay: ${body.options.debounce.delay}. ` +
|
|
`Supported formats: {number}s, {number}m, {number}h or {number}hr, {number}d, ` +
|
|
`{number}w, optionally combined (for example "2h30m").`
|
|
);
|
|
}
|
|
|
|
this.#validateDebounceWindow(body.options.debounce);
|
|
}
|
|
|
|
const parentRun = body.options?.parentRunId
|
|
? await runStore.findRun(
|
|
{
|
|
id: RunId.fromFriendlyId(body.options.parentRunId),
|
|
runtimeEnvironmentId: environment.id,
|
|
},
|
|
this.prisma
|
|
)
|
|
: undefined;
|
|
|
|
const parentRunValidation = this.validator.validateParentRun({
|
|
taskId,
|
|
parentRun: parentRun ?? undefined,
|
|
resumeParentOnCompletion: body.options?.resumeParentOnCompletion,
|
|
});
|
|
|
|
if (!parentRunValidation.ok) {
|
|
throw parentRunValidation.error;
|
|
}
|
|
|
|
const idempotencyKeyConcernResult = await this.idempotencyKeyConcern.handleTriggerRequest(
|
|
triggerRequest,
|
|
parentRun?.taskEventStore
|
|
);
|
|
|
|
if (idempotencyKeyConcernResult.isCached) {
|
|
return idempotencyKeyConcernResult;
|
|
}
|
|
|
|
const {
|
|
idempotencyKey,
|
|
idempotencyKeyExpiresAt,
|
|
claim: claimResult,
|
|
} = idempotencyKeyConcernResult;
|
|
|
|
// If we own an idempotency claim, the trigger pipeline below MUST
|
|
// resolve it — publish on success so waiters see our runId,
|
|
// release on error so the next claimant can retry. Stored in an
|
|
// outer scope so the try/catch at the bottom of `callV2` can act
|
|
// on whichever return path or throw the pipeline takes.
|
|
idempotencyClaim = claimResult;
|
|
|
|
if (idempotencyKey) {
|
|
await this.triggerRacepointSystem.waitForRacepoint({
|
|
racepoint: "idempotencyKey",
|
|
id: idempotencyKey,
|
|
});
|
|
}
|
|
|
|
const lockedToBackgroundWorker = body.options?.lockToVersion
|
|
? await this.prisma.backgroundWorker.findFirst({
|
|
where: {
|
|
projectId: environment.projectId,
|
|
runtimeEnvironmentId: environment.id,
|
|
version: body.options?.lockToVersion,
|
|
},
|
|
select: {
|
|
id: true,
|
|
version: true,
|
|
sdkVersion: true,
|
|
cliVersion: true,
|
|
},
|
|
})
|
|
: undefined;
|
|
|
|
const { queueName, lockedQueueId, taskTtl, taskKind } =
|
|
await this.queueConcern.resolveQueueProperties(
|
|
triggerRequest,
|
|
lockedToBackgroundWorker ?? undefined
|
|
);
|
|
|
|
// Resolve TTL with precedence: per-trigger > task-level > dev default
|
|
let ttl: string | undefined;
|
|
|
|
if (body.options?.ttl !== undefined) {
|
|
ttl =
|
|
typeof body.options.ttl === "number"
|
|
? stringifyDuration(body.options.ttl)
|
|
: body.options.ttl;
|
|
} else {
|
|
ttl = taskTtl ?? (environment.type === "DEVELOPMENT" ? "10m" : undefined);
|
|
}
|
|
|
|
if (!options.skipChecks) {
|
|
const queueSizeGuard = await this.queueConcern.validateQueueLimits(
|
|
environment,
|
|
queueName
|
|
);
|
|
|
|
if (!queueSizeGuard.ok) {
|
|
throw new QueueSizeLimitExceededError(
|
|
`Cannot trigger ${taskId} as the queue size limit for this environment has been reached. The maximum size is ${queueSizeGuard.maximumSize}`,
|
|
queueSizeGuard.maximumSize ?? 0,
|
|
undefined,
|
|
"warn"
|
|
);
|
|
}
|
|
}
|
|
|
|
const metadataPacket = body.options?.metadata
|
|
? handleMetadataPacket(
|
|
body.options?.metadata,
|
|
body.options?.metadataType ?? "application/json",
|
|
this.metadataMaximumSize
|
|
)
|
|
: undefined;
|
|
|
|
const tags = (
|
|
body.options?.tags
|
|
? typeof body.options.tags === "string"
|
|
? [body.options.tags]
|
|
: body.options.tags
|
|
: []
|
|
).filter((tag) => tag.trim().length > 0);
|
|
|
|
const depth = parentRun ? parentRun.depth + 1 : 0;
|
|
|
|
const workerQueueResult = await this.queueConcern.getWorkerQueue(
|
|
environment,
|
|
body.options?.region
|
|
);
|
|
const baseWorkerQueue = workerQueueResult?.masterQueue;
|
|
const enableFastPath = workerQueueResult?.enableFastPath ?? false;
|
|
|
|
// Rewrite the region to its compute backing for migration-enrolled orgs,
|
|
// from the in-memory snapshots (no DB query). A cold read (registry not yet
|
|
// loaded) returns undefined/[] and the resolver falls back to not-migrated.
|
|
const workerGroups = workerRegionRegistry.current() ?? [];
|
|
const region = baseWorkerQueue
|
|
? regionForQueue(baseWorkerQueue, workerGroups)
|
|
: undefined;
|
|
const backing = baseWorkerQueue
|
|
? backingForQueue(baseWorkerQueue, workerGroups)
|
|
: undefined;
|
|
const migrated = resolveComputeMigration({
|
|
baseWorkerQueue,
|
|
baseEnableFastPath: enableFastPath,
|
|
region,
|
|
backing,
|
|
planType,
|
|
orgId: environment.organization.id,
|
|
orgFeatureFlags: environment.organization.featureFlags as Record<
|
|
string,
|
|
unknown
|
|
> | null,
|
|
flags: globalFlagsRegistry.current(),
|
|
envType: environment.type,
|
|
});
|
|
|
|
const triggerSource = options.triggerSource ?? "api";
|
|
const triggerAction = options.triggerAction ?? "trigger";
|
|
const parentAnnotations = RunAnnotations.safeParse(parentRun?.annotations).data;
|
|
const annotations = {
|
|
triggerSource,
|
|
triggerAction,
|
|
rootTriggerSource: parentAnnotations?.rootTriggerSource ?? triggerSource,
|
|
rootScheduleId: parentAnnotations?.rootScheduleId || options.scheduleId || undefined,
|
|
taskKind: taskKind ?? "STANDARD",
|
|
};
|
|
|
|
// Route runs in a scheduled lineage (the scheduled run itself and every
|
|
// descendant, via the propagated rootTriggerSource) to a dedicated
|
|
// `<region>:scheduled` worker queue so a separate consumer fleet can
|
|
// dequeue them independently of standard/agent runs. Gated per-org with
|
|
// a global default, never applied to dev. Reads only the in-memory org
|
|
// flags already on the environment — no DB query on the hot path.
|
|
const scheduledQueueSplitEnabled =
|
|
environment.type !== "DEVELOPMENT" &&
|
|
resolveScheduledQueueSplitEnabled({
|
|
orgFeatureFlags: environment.organization.featureFlags as Record<
|
|
string,
|
|
unknown
|
|
> | null,
|
|
globalDefault: env.TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED === "1",
|
|
});
|
|
const workerQueue =
|
|
migrated.workerQueue !== undefined
|
|
? workerQueueForRun({
|
|
workerQueue: migrated.workerQueue,
|
|
rootTriggerSource: annotations.rootTriggerSource,
|
|
splitEnabled: scheduledQueueSplitEnabled,
|
|
})
|
|
: migrated.workerQueue;
|
|
|
|
try {
|
|
return await this.traceEventConcern.traceRun(
|
|
triggerRequest,
|
|
parentRun?.taskEventStore,
|
|
async (event, store) => {
|
|
event.setAttribute("queueName", queueName);
|
|
span.setAttribute("queueName", queueName);
|
|
event.setAttribute("runId", runFriendlyId);
|
|
span.setAttribute("runId", runFriendlyId);
|
|
|
|
// Short-circuit when mollifier is globally off (the default
|
|
// for every deployment that hasn't opted in). Avoids the
|
|
// GateInputs allocation, the deps spread inside `evaluateGate`,
|
|
// and the `mollifier.decisions{outcome=pass_through}` OTel
|
|
// increment on every trigger — `triggerTask` is the
|
|
// highest-throughput code path in the system. The check goes
|
|
// through a DI'd predicate so unit tests that inject a custom
|
|
// `evaluateGate` can also override the gate-on check (the
|
|
// default reads `env.TRIGGER_MOLLIFIER_ENABLED`, which is "0"
|
|
// in CI where no .env file is present).
|
|
//
|
|
// Batch items bypass the mollifier gate entirely.
|
|
//
|
|
// The mollify path returns a stripped run-shape `{ id,
|
|
// friendlyId, spanId }` with no PG row written. Batch
|
|
// tracking relies on `BatchTaskRunItem`, a join row whose
|
|
// `taskRunId` column has a NOT NULL FK to `TaskRun.id` —
|
|
// creating that join at trigger-time (in
|
|
// `batchTriggerV3.server.ts:871`) fails with FK violation
|
|
// for any mollified item, and skipping it at trigger-time
|
|
// would silently drop the batch↔run link forever because
|
|
// the drainer's materialise path doesn't (yet) create
|
|
// `BatchTaskRunItem`. Either side alone is wrong:
|
|
// - skip at trigger-time only → batch progress
|
|
// under-reports forever, `batchTriggerAndWait` parent
|
|
// stays parked
|
|
// - mollify at trigger-time only → FK violation, 500
|
|
//
|
|
// The proper end state is a drainer-side
|
|
// `BatchTaskRunItem` create-on-materialise (the snapshot
|
|
// already carries `batch: { id, index }` so the drainer
|
|
// has the info). That belongs in the drainer / replay PR,
|
|
// not here. Until that lands, batch triggers pass-through
|
|
// — they lose the burst-protection benefit, but the path
|
|
// works end-to-end.
|
|
const skipMollifierForBatch = !!options.batchId;
|
|
const mollifierOutcome: GateOutcome | null =
|
|
this.isMollifierGloballyEnabled() && !skipMollifierForBatch
|
|
? await this.evaluateGate({
|
|
envId: environment.id,
|
|
orgId: environment.organizationId,
|
|
taskId,
|
|
orgFeatureFlags:
|
|
(environment.organization.featureFlags as Record<
|
|
string,
|
|
unknown
|
|
> | null) ?? null,
|
|
options: {
|
|
debounce: body.options?.debounce,
|
|
oneTimeUseToken: options.oneTimeUseToken,
|
|
parentTaskRunId: body.options?.parentRunId,
|
|
resumeParentOnCompletion: body.options?.resumeParentOnCompletion,
|
|
},
|
|
})
|
|
: null;
|
|
|
|
// When the gate says mollify, write the engine.trigger input
|
|
// snapshot into the Redis buffer and return a synthesised
|
|
// TriggerTaskServiceResult. The customer never waits on
|
|
// Postgres; the drainer materialises the run later by replaying
|
|
// engine.trigger against the snapshot. The run span has already
|
|
// been opened by traceRun above (PARTIAL event in ClickHouse),
|
|
// so its traceId/spanId live in the snapshot and the drainer's
|
|
// `mollifier.drained` span parents on the same trace — buffered
|
|
// runs become visible in the dashboard's trace view immediately,
|
|
// not only after the drainer fires.
|
|
if (mollifierOutcome?.action === "mollify") {
|
|
const mollifierBuffer = this.getMollifierBuffer();
|
|
if (mollifierBuffer && !body.options?.debounce) {
|
|
event.setAttribute("mollifier.reason", mollifierOutcome.decision.reason);
|
|
event.setAttribute("mollifier.count", String(mollifierOutcome.decision.count));
|
|
event.setAttribute(
|
|
"mollifier.threshold",
|
|
String(mollifierOutcome.decision.threshold)
|
|
);
|
|
event.setAttribute("taskRunId", runFriendlyId);
|
|
|
|
const payloadPacket = await this.payloadProcessor.process(triggerRequest);
|
|
|
|
const engineTriggerInput = this.#buildEngineTriggerInput({
|
|
runFriendlyId,
|
|
environment,
|
|
idempotencyKey,
|
|
idempotencyKeyExpiresAt,
|
|
body,
|
|
options,
|
|
queueName,
|
|
lockedQueueId,
|
|
workerQueue,
|
|
region: migrated.region,
|
|
enableFastPath: migrated.enableFastPath,
|
|
lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
|
|
delayUntil,
|
|
ttl,
|
|
metadataPacket,
|
|
tags,
|
|
depth,
|
|
parentRun: parentRun ?? undefined,
|
|
annotations,
|
|
planType,
|
|
taskId,
|
|
payloadPacket,
|
|
traceContext: this.#propagateExternalTraceContext(
|
|
event.traceContext,
|
|
parentRun?.traceContext,
|
|
event.traceparent?.spanId
|
|
),
|
|
traceId: event.traceId,
|
|
spanId: event.spanId,
|
|
parentSpanId:
|
|
options.parentAsLinkType === "replay"
|
|
? undefined
|
|
: event.traceparent?.spanId,
|
|
taskEventStore: store,
|
|
});
|
|
|
|
const result = await mollifyTrigger({
|
|
runFriendlyId,
|
|
environmentId: environment.id,
|
|
organizationId: environment.organizationId,
|
|
engineTriggerInput,
|
|
decision: mollifierOutcome.decision,
|
|
buffer: mollifierBuffer,
|
|
// Idempotency-key triple wires the buffer's SETNX into
|
|
// the trigger-time dedup symmetric with PG.
|
|
idempotencyKey,
|
|
taskIdentifier: taskId,
|
|
});
|
|
|
|
logger.debug("mollifier.buffered", {
|
|
runId: runFriendlyId,
|
|
envId: environment.id,
|
|
orgId: environment.organizationId,
|
|
taskId,
|
|
reason: mollifierOutcome.decision.reason,
|
|
});
|
|
|
|
// Synthetic result is structurally narrower than the full
|
|
// TaskRun; the route handler only reads
|
|
// `result.run.friendlyId`. traceRun flushes the PARTIAL
|
|
// run-span event to ClickHouse on callback return.
|
|
// `isMollified` flags the route to skip the request-
|
|
// idempotency cache write — see the field's contract on
|
|
// `TriggerTaskServiceResult`.
|
|
return {
|
|
...(result as unknown as TriggerTaskServiceResult),
|
|
isMollified: true,
|
|
};
|
|
}
|
|
if (!mollifierBuffer) {
|
|
logger.warn(
|
|
"mollifier gate said mollify but buffer is null — falling through to pass-through"
|
|
);
|
|
}
|
|
}
|
|
|
|
const payloadPacket = await this.payloadProcessor.process(triggerRequest);
|
|
|
|
const baseEngineInput = this.#buildEngineTriggerInput({
|
|
runFriendlyId,
|
|
environment,
|
|
idempotencyKey,
|
|
idempotencyKeyExpiresAt,
|
|
body,
|
|
options,
|
|
queueName,
|
|
lockedQueueId,
|
|
workerQueue,
|
|
region: migrated.region,
|
|
enableFastPath: migrated.enableFastPath,
|
|
lockedToBackgroundWorker: lockedToBackgroundWorker ?? undefined,
|
|
delayUntil,
|
|
ttl,
|
|
metadataPacket,
|
|
tags,
|
|
depth,
|
|
parentRun: parentRun ?? undefined,
|
|
annotations,
|
|
planType,
|
|
taskId,
|
|
payloadPacket,
|
|
traceContext: this.#propagateExternalTraceContext(
|
|
event.traceContext,
|
|
parentRun?.traceContext,
|
|
event.traceparent?.spanId
|
|
),
|
|
traceId: event.traceId,
|
|
spanId: event.spanId,
|
|
parentSpanId:
|
|
options.parentAsLinkType === "replay" ? undefined : event.traceparent?.spanId,
|
|
taskEventStore: store,
|
|
});
|
|
|
|
const taskRun = await this.engine.trigger(
|
|
{
|
|
...baseEngineInput,
|
|
// onDebounced is a closure over webapp state (triggerRequest +
|
|
// traceEventConcern) and can't be serialised into the mollifier
|
|
// snapshot. The pass-through path attaches it here; the drainer
|
|
// path replays without it. The debounce and triggerAndWait gate
|
|
// bypasses ensure neither reaches the mollify branch.
|
|
onDebounced:
|
|
body.options?.debounce && body.options?.resumeParentOnCompletion
|
|
? async ({ existingRun, waitpoint, debounceKey }) => {
|
|
return await this.traceEventConcern.traceDebouncedRun(
|
|
triggerRequest,
|
|
parentRun?.taskEventStore,
|
|
{
|
|
existingRun,
|
|
debounceKey,
|
|
incomplete: waitpoint.status === "PENDING",
|
|
isError: waitpoint.outputIsError,
|
|
},
|
|
async (spanEvent) => {
|
|
const spanId =
|
|
options?.parentAsLinkType === "replay"
|
|
? spanEvent.spanId
|
|
: spanEvent.traceparent?.spanId
|
|
? `${spanEvent.traceparent.spanId}:${spanEvent.spanId}`
|
|
: spanEvent.spanId;
|
|
return spanId;
|
|
}
|
|
);
|
|
}
|
|
: undefined,
|
|
},
|
|
this.prisma
|
|
);
|
|
|
|
// If the returned run has a different friendlyId, it was debounced.
|
|
// For triggerAndWait: stop the outer span since a replacement debounced span was created via onDebounced.
|
|
// For regular trigger: let the span complete normally - no replacement span needed since the
|
|
// original run already has its span from when it was first created.
|
|
if (
|
|
taskRun.friendlyId !== runFriendlyId &&
|
|
body.options?.debounce &&
|
|
body.options?.resumeParentOnCompletion
|
|
) {
|
|
event.stop();
|
|
}
|
|
|
|
const error = taskRun.error ? TaskRunError.parse(taskRun.error) : undefined;
|
|
|
|
if (error) {
|
|
event.failWithError(error);
|
|
}
|
|
|
|
const result = { run: taskRun, error, isCached: false };
|
|
|
|
if (result?.error) {
|
|
throw new ServiceValidationError(
|
|
taskRunErrorToString(taskRunErrorEnhancer(result.error))
|
|
);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
);
|
|
} catch (error) {
|
|
if (error instanceof RunDuplicateIdempotencyKeyError) {
|
|
//retry calling this function, because this time it will return the idempotent run
|
|
return await this.call({
|
|
taskId,
|
|
environment,
|
|
body,
|
|
options: { ...options, runFriendlyId },
|
|
attempt: attempt + 1,
|
|
});
|
|
}
|
|
|
|
if (error instanceof RunOneTimeUseTokenError) {
|
|
throw new ServiceValidationError(
|
|
`Cannot trigger ${taskId} with a one-time use token as it has already been used.`
|
|
);
|
|
}
|
|
|
|
throw error;
|
|
}
|
|
}
|
|
);
|
|
// Pipeline returned successfully — publish the claim if we held
|
|
// one. Waiters polling for our key resolve to this runId.
|
|
if (idempotencyClaim && result?.run?.friendlyId) {
|
|
const published = await publishMollifierClaim({
|
|
envId: idempotencyClaim.envId,
|
|
taskIdentifier: idempotencyClaim.taskIdentifier,
|
|
idempotencyKey: idempotencyClaim.idempotencyKey,
|
|
token: idempotencyClaim.token,
|
|
runId: result.run.friendlyId,
|
|
ttlSeconds: env.TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS,
|
|
});
|
|
if (!published) {
|
|
// Our claim expired mid-pipeline and another claimant took it, so this publish no-op'd: a
|
|
// different run is now canonical for this key while we return ours (a cross-DB dup under the
|
|
// split). Rare now the claim TTL is floored (C1); surfaced for monitoring pending auto-
|
|
// convergence (re-resolve the current winner + cancel this orphan).
|
|
logger.warn("mollifier claim publish no-op'd; winner lost the claim mid-pipeline", {
|
|
envId: idempotencyClaim.envId,
|
|
taskIdentifier: idempotencyClaim.taskIdentifier,
|
|
runId: result.run.friendlyId,
|
|
});
|
|
}
|
|
}
|
|
return result;
|
|
} catch (err) {
|
|
// Pipeline threw — release the claim so the next claimant can
|
|
// retry. Re-throw so the caller sees the original error.
|
|
if (idempotencyClaim) {
|
|
await releaseMollifierClaim(idempotencyClaim);
|
|
}
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
// Build the engine.trigger() input object from the values gathered during
|
|
// this.call(). Extracted so the mollify path can construct the
|
|
// same input shape without re-entering the trace-run span. The pass-through
|
|
// path spreads this result and attaches `onDebounced` inline; the mollify
|
|
// path serialises it into the buffer for drainer replay.
|
|
#buildEngineTriggerInput(args: {
|
|
runFriendlyId: string;
|
|
environment: AuthenticatedEnvironment;
|
|
idempotencyKey?: string;
|
|
idempotencyKeyExpiresAt?: Date;
|
|
body: TriggerTaskRequest["body"];
|
|
options: TriggerTaskServiceOptions;
|
|
queueName: string;
|
|
lockedQueueId?: string;
|
|
workerQueue?: string;
|
|
region?: string;
|
|
enableFastPath: boolean;
|
|
lockedToBackgroundWorker?: {
|
|
id: string;
|
|
version: string;
|
|
sdkVersion: string;
|
|
cliVersion: string;
|
|
};
|
|
delayUntil?: Date;
|
|
ttl?: string;
|
|
metadataPacket?: { data?: string; dataType: string };
|
|
tags: string[];
|
|
depth: number;
|
|
parentRun?: {
|
|
id: string;
|
|
rootTaskRunId?: string | null;
|
|
queueTimestamp?: Date | null;
|
|
taskEventStore?: string;
|
|
};
|
|
annotations: {
|
|
triggerSource: string;
|
|
triggerAction: string;
|
|
rootTriggerSource: string;
|
|
rootScheduleId?: string | undefined;
|
|
};
|
|
planType?: string;
|
|
taskId: string;
|
|
payloadPacket: { data?: string; dataType: string };
|
|
traceContext: TriggerTraceContext;
|
|
traceId: string;
|
|
spanId: string;
|
|
parentSpanId: string | undefined;
|
|
taskEventStore: string;
|
|
}) {
|
|
return {
|
|
friendlyId: args.runFriendlyId,
|
|
environment: args.environment,
|
|
idempotencyKey: args.idempotencyKey,
|
|
idempotencyKeyExpiresAt: args.idempotencyKey ? args.idempotencyKeyExpiresAt : undefined,
|
|
idempotencyKeyOptions: removeNullBytesFromKey(args.body.options?.idempotencyKeyOptions),
|
|
taskIdentifier: args.taskId,
|
|
payload: args.payloadPacket.data ?? "",
|
|
payloadType: args.payloadPacket.dataType,
|
|
context: args.body.context,
|
|
traceContext: args.traceContext,
|
|
traceId: args.traceId,
|
|
spanId: args.spanId,
|
|
parentSpanId: args.parentSpanId,
|
|
replayedFromTaskRunFriendlyId: args.options.replayedFromTaskRunFriendlyId,
|
|
lockedToVersionId: args.lockedToBackgroundWorker?.id,
|
|
taskVersion: args.lockedToBackgroundWorker?.version,
|
|
sdkVersion: args.lockedToBackgroundWorker?.sdkVersion,
|
|
cliVersion: args.lockedToBackgroundWorker?.cliVersion,
|
|
// Schema-level coercion now lands `body.options.concurrencyKey` as
|
|
// `string` on the API path, but the BatchQueue worker rebuilds
|
|
// body.options from Redis-stored items (Record<string, unknown>),
|
|
// which can still carry the pre-fix shape from in-flight batches.
|
|
concurrencyKey:
|
|
typeof args.body.options?.concurrencyKey === "number"
|
|
? String(args.body.options.concurrencyKey)
|
|
: args.body.options?.concurrencyKey,
|
|
queue: args.queueName,
|
|
lockedQueueId: args.lockedQueueId,
|
|
workerQueue: args.workerQueue,
|
|
region: args.region,
|
|
enableFastPath: args.enableFastPath,
|
|
isTest: args.body.options?.test ?? false,
|
|
delayUntil: args.delayUntil,
|
|
queuedAt: args.delayUntil ? undefined : new Date(),
|
|
maxAttempts: args.body.options?.maxAttempts,
|
|
taskEventStore: args.taskEventStore,
|
|
ttl: args.ttl,
|
|
tags: args.tags,
|
|
oneTimeUseToken: args.options.oneTimeUseToken,
|
|
parentTaskRunId: args.parentRun?.id,
|
|
rootTaskRunId: args.parentRun?.rootTaskRunId ?? args.parentRun?.id,
|
|
batch: args.options?.batchId
|
|
? { id: args.options.batchId, index: args.options.batchIndex ?? 0 }
|
|
: undefined,
|
|
resumeParentOnCompletion: args.body.options?.resumeParentOnCompletion,
|
|
depth: args.depth,
|
|
metadata: args.metadataPacket?.data,
|
|
metadataType: args.metadataPacket?.dataType,
|
|
seedMetadata: args.metadataPacket?.data,
|
|
seedMetadataType: args.metadataPacket?.dataType,
|
|
maxDurationInSeconds: args.body.options?.maxDuration
|
|
? clampMaxDuration(args.body.options.maxDuration)
|
|
: undefined,
|
|
machine: args.body.options?.machine,
|
|
priorityMs: args.body.options?.priority
|
|
? clampPriorityMs(args.body.options.priority)
|
|
: undefined,
|
|
queueTimestamp:
|
|
args.options.queueTimestamp ??
|
|
(args.parentRun && args.body.options?.resumeParentOnCompletion
|
|
? (args.parentRun.queueTimestamp ?? undefined)
|
|
: undefined),
|
|
scheduleId: args.options.scheduleId,
|
|
scheduleInstanceId: args.options.scheduleInstanceId,
|
|
createdAt: args.options.overrideCreatedAt,
|
|
bulkActionId: args.body.options?.bulkActionId,
|
|
planType: args.planType,
|
|
realtimeStreamsVersion: args.options.realtimeStreamsVersion,
|
|
streamBasinName: args.environment.organization.streamBasinName,
|
|
debounce: removeNullBytesFromKey(args.body.options?.debounce),
|
|
annotations: args.annotations,
|
|
};
|
|
}
|
|
|
|
#propagateExternalTraceContext(
|
|
traceContext: Record<string, unknown>,
|
|
parentRunTraceContext: unknown,
|
|
parentSpanId: string | undefined
|
|
): TriggerTraceContext {
|
|
if (!parentRunTraceContext) {
|
|
return traceContext;
|
|
}
|
|
|
|
const parsedParentRunTraceContext = TriggerTraceContext.safeParse(parentRunTraceContext);
|
|
|
|
if (!parsedParentRunTraceContext.success) {
|
|
return traceContext;
|
|
}
|
|
|
|
const { external } = parsedParentRunTraceContext.data;
|
|
|
|
if (!external) {
|
|
return traceContext;
|
|
}
|
|
|
|
if (!external.traceparent) {
|
|
return traceContext;
|
|
}
|
|
|
|
const parsedTraceparent = parseTraceparent(external.traceparent);
|
|
|
|
if (!parsedTraceparent) {
|
|
return traceContext;
|
|
}
|
|
|
|
const newExternalTraceparent = serializeTraceparent(
|
|
parsedTraceparent.traceId,
|
|
parentSpanId ?? parsedTraceparent.spanId,
|
|
parsedTraceparent.traceFlags
|
|
);
|
|
|
|
return {
|
|
...traceContext,
|
|
external: {
|
|
...external,
|
|
traceparent: newExternalTraceparent,
|
|
},
|
|
};
|
|
}
|
|
}
|