Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fix-console-interceptor-2900.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/core": patch
---

Fix: ConsoleInterceptor now delegates to original console methods to preserve log chain when other interceptors (like Sentry) are present. (#2900)
5 changes: 5 additions & 0 deletions .changeset/fix-docker-hub-rate-limit-2911.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/cli-v3": patch
---

Fix: Native build server failed with Docker Hub rate limits. Added support for checking checking `DOCKER_USERNAME` and `DOCKER_PASSWORD` in environment variables and logging into Docker Hub before building. (#2911)
5 changes: 5 additions & 0 deletions .changeset/fix-github-install-node-version-2913.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/cli-v3": patch
---

Fix: Ignore engine checks during deployment install phase to prevent failure on build server when Node version mismatch exists. (#2913)
5 changes: 5 additions & 0 deletions .changeset/fix-orphaned-workers-2909.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/cli-v3": patch
---

Fix: `trigger.dev dev` command left orphaned worker processes when exited via Ctrl+C (SIGINT). Added signal handlers to ensure proper cleanup of child processes and lockfiles. (#2909)
5 changes: 5 additions & 0 deletions .changeset/fix-sentry-oom-2920.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@trigger.dev/cli-v3": patch
---

Fix Sentry OOM: Allow disabling `source-map-support` via `TRIGGER_SOURCE_MAPS=false`. Also supports `node` for native source maps. (#2920)
6 changes: 6 additions & 0 deletions .server-changes/mollifier-trigger.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
area: webapp
type: feature
---

Mollifier trigger-time decisions: gate `engine.trigger`, mollify bursts into the buffer, claim idempotency keys, and read-fallback for buffered runs.
218 changes: 217 additions & 1 deletion apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,50 @@ import { RunId } from "@trigger.dev/core/v3/isomorphic";
import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database";
import { logger } from "~/services/logger.server";
import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
import { ServiceValidationError } from "~/v3/services/common.server";
import type { RunEngine } from "~/v3/runEngine.server";
import { shouldIdempotencyKeyBeCleared } from "~/v3/taskStatus";
import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
import { claimOrAwait } from "~/v3/mollifier/idempotencyClaim.server";
import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server";
import type { TraceEventConcern, TriggerTaskRequest } from "../types";

// In-memory per-org mollifier-enabled check, shared with `evaluateGate`
// (same `Organization.featureFlags` JSON, no DB read). Used to gate the
// pre-gate claim's Redis round-trip so non-mollifier orgs don't pay it
// during staged rollout — see the comment above the claim block in
// handleTriggerRequest.
const resolveOrgMollifierFlag = makeResolveMollifierFlag();

// Claim ownership context returned to the caller when the
// IdempotencyKeyConcern won a pre-gate claim. Caller MUST publish the
// winning runId on pipeline success (`publishClaim`) or release the
// claim on failure (`releaseClaim`).
export type ClaimedIdempotency = {
envId: string;
taskIdentifier: string;
idempotencyKey: string;
// Ownership token from `claimOrAwait`. The caller's trigger pipeline
// MUST thread this into publishClaim/releaseClaim so the buffer's
// compare-and-act protects the slot against a stale predecessor.
token: string;
};

export type IdempotencyKeyConcernResult =
| { isCached: true; run: TaskRun }
| { isCached: false; idempotencyKey?: string; idempotencyKeyExpiresAt?: Date };
| {
isCached: false;
idempotencyKey?: string;
idempotencyKeyExpiresAt?: Date;
// Set when this trigger holds a pre-gate claim. The caller's
// trigger pipeline MUST resolve the claim by either publishing
// the runId on success or releasing on failure. Undefined when
// the request has no idempotency key, when the buffer is
// unavailable, or when the request is a triggerAndWait (claim
// path skipped per plan doc).
claim?: ClaimedIdempotency;
};

export class IdempotencyKeyConcern {
constructor(
Expand All @@ -17,6 +54,47 @@ export class IdempotencyKeyConcern {
private readonly traceEventConcern: TraceEventConcern
) {}

// Buffer-side idempotency dedup. Resolves an idempotency key against the
// mollifier buffer when PG missed. Returns a SyntheticRun cast to
// TaskRun so the route handler (which only reads run.id / run.friendlyId)
// can echo the buffered run's friendlyId as a cached hit. Returns null
// for any failure or miss — buffer outages must not 500 the trigger
// hot path; we fail open to "no cache hit" and let the request through.
private async findBufferedRunWithIdempotency(
environmentId: string,
organizationId: string,
taskIdentifier: string,
idempotencyKey: string,
): Promise<TaskRun | null> {
const buffer = getMollifierBuffer();
if (!buffer) return null;

let bufferedRunId: string | null;
try {
bufferedRunId = await buffer.lookupIdempotency({
envId: environmentId,
taskIdentifier,
idempotencyKey,
});
} catch (err) {
logger.error("IdempotencyKeyConcern: buffer lookupIdempotency failed", {
environmentId,
taskIdentifier,
err: err instanceof Error ? err.message : String(err),
});
return null;
}
if (!bufferedRunId) return null;

const synthetic = await findRunByIdWithMollifierFallback({
runId: bufferedRunId,
environmentId,
organizationId,
});
if (!synthetic) return null;
return synthetic as unknown as TaskRun;
}

async handleTriggerRequest(
request: TriggerTaskRequest,
parentStore: string | undefined
Expand Down Expand Up @@ -44,6 +122,25 @@ export class IdempotencyKeyConcern {
})
: undefined;

// Buffer fallback per the mollifier-idempotency design. PG missed —
// the same key may belong to a buffered run that hasn't materialised
// yet. Skipped when `resumeParentOnCompletion` is set: blocking a
// parent on a buffered child via waitpoint requires a PG row that
// doesn't exist yet. The follow-up accept's SETNX in mollifyTrigger
// still dedupes the trigger itself; the waitpoint just doesn't fire
// for this rare race window.
if (!existingRun && idempotencyKey && !request.body.options?.resumeParentOnCompletion) {
const buffered = await this.findBufferedRunWithIdempotency(
request.environment.id,
request.environment.organizationId,
request.taskId,
idempotencyKey,
);
if (buffered) {
return { isCached: true, run: buffered };
}
}
Comment on lines +132 to +142

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Every idempotency-keyed trigger that misses the database now pays an extra Redis round-trip, even for organisations not enrolled in the new buffering

The buffered-run lookup runs for all organisations (findBufferedRunWithIdempotency at apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts:133) without the per-organisation enrolment check that the neighbouring claim step uses, adding a Redis call to the hottest request path for tenants that never buffer anything.
Impact: Trigger latency increases for all idempotency-keyed calls once the feature is switched on globally, even for customers who are not enrolled.

Inconsistent gating between the two buffer touches

The pre-gate claim is explicitly gated by resolveOrgMollifierFlag precisely to keep the Redis round-trip off the hot path for non-enrolled orgs (apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts:251-262, with a long comment explaining the rationale). The buffer fallback added just above it (lines 132-142) performs buffer.lookupIdempotency unconditionally whenever getMollifierBuffer() is non-null, i.e. whenever TRIGGER_MOLLIFIER_ENABLED=1 globally (apps/webapp/app/v3/mollifier/mollifierBuffer.server.ts:29-32). Since a non-enrolled org can never have a buffered run (the gate always returns pass_through for it), this lookup can only ever miss, so the same org-flag guard should apply here.

Prompt for agents
In IdempotencyKeyConcern.handleTriggerRequest, the buffer idempotency fallback (findBufferedRunWithIdempotency) runs for every idempotency-keyed trigger whenever the global mollifier buffer exists, while the pre-gate claim below it is deliberately gated on the per-org mollifier feature flag to avoid a Redis round-trip on the trigger hot path. Non-enrolled orgs can never have buffered runs, so the lookup can only miss. Apply the same per-org flag check (resolveOrgMollifierFlag) to the buffer fallback, ideally resolving the flag once and reusing it for both the fallback and the claim.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


if (existingRun) {
// The idempotency key has expired
if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
Expand Down Expand Up @@ -133,6 +230,125 @@ export class IdempotencyKeyConcern {
return { isCached: true, run: existingRun };
}

// Pre-gate claim — closes the PG+buffer race during gate transition.
// All same-key triggers serialise here before evaluateGate decides
// PG-pass-through vs mollify. Skipped for triggerAndWait
// (resumeParentOnCompletion) — that path bypasses the gate entirely
// and its existing PG-side dedup is sufficient.
//
// Also gated on the same per-org mollifier flag the gate uses: when
// `TRIGGER_MOLLIFIER_ENABLED=1` globally for staged rollout, the buffer
// singleton is constructed and `claimOrAwait` would otherwise issue a
// Redis SETNX for EVERY idempotency-keyed trigger — including orgs
// that haven't opted in. Those orgs never enter the mollify branch
// (the gate always returns pass_through for them), so there's no
// buffer activity to serialise against; PG's unique constraint
// already deduplicates concurrent same-key races. Resolving the org
// flag is a pure in-memory read of `Organization.featureFlags` — no
// DB query, same predicate the gate uses — keeping the claim's Redis
// RTT off the hot path for non-opted-in orgs during incremental
// rollout.
const claimEligible =
!request.body.options?.resumeParentOnCompletion &&
(await resolveOrgMollifierFlag({
envId: request.environment.id,
orgId: request.environment.organizationId,
taskId: request.taskId,
orgFeatureFlags:
((request.environment.organization?.featureFlags as
| Record<string, unknown>
| null
| undefined) ?? null),
}));
if (claimEligible) {
const ttlSeconds = Math.max(
1,
Math.min(
30,
Math.ceil((idempotencyKeyExpiresAt.getTime() - Date.now()) / 1000),
),
);
const outcome = await claimOrAwait({
envId: request.environment.id,
taskIdentifier: request.taskId,
idempotencyKey,
ttlSeconds,
});
if (outcome.kind === "resolved") {
// Another concurrent trigger committed first. Re-resolve via the
// existing checks: writer-side PG findFirst first (defeats
// replica lag), then buffer fallback for the buffered case.
const writerRun = await this.prisma.taskRun.findFirst({
where: {
runtimeEnvironmentId: request.environment.id,
idempotencyKey,
taskIdentifier: request.taskId,
},
include: { associatedWaitpoint: true },
});
if (writerRun) {
return { isCached: true, run: writerRun };
}
const buffered = await this.findBufferedRunWithIdempotency(
request.environment.id,
request.environment.organizationId,
request.taskId,
idempotencyKey,
);
if (buffered) {
return { isCached: true, run: buffered };
}
// Claim resolved to a runId nothing can find — the run was
// genuinely lost (claimant errored after publish, drain failed,
// or both the PG row and buffer entry TTL'd out). This is
// terminal, not transient: `lookupIdempotency` self-heals a
// dangling pointer, and `ack` keeps the entry hash as a
// read-fallback past the PG write, so re-polling cannot conjure
// a run that is gone. Falling through to a fresh trigger is the
// correct recovery.
//
// Why falling through claimless is safe (no duplicate runs):
// concurrent triggers that also fall through here converge on a
// single run via the same dedup backstops the claim layer relies
// on — the PG unique constraint on the idempotency key
// (RunDuplicateIdempotencyKeyError → retry resolves to the
// winner) for the pass-through path, and `accept`'s idempotency
// SETNX (`duplicate_idempotency`) for the mollify path. Once the
// first fall-through commits a run, later callers find it via the
// writer-PG / buffer lookups above despite the stale `resolved:`
// slot, which the slot's TTL clears within ~30s. The residual
// cost is a few redundant (deduped) trigger attempts in that
// window, not duplicate runs.
logger.warn("idempotency claim resolved but runId not findable", {
envId: request.environment.id,
taskIdentifier: request.taskId,
claimedRunId: outcome.runId,
});
}
if (outcome.kind === "timed_out") {
throw new ServiceValidationError(
"Idempotency claim resolution timed out",
503,
);
}
Comment on lines +328 to +333

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Two simultaneous triggers with the same idempotency key can make the second one fail with a 503 instead of returning the first run

When two triggers share an idempotency key, the second one now blocks waiting for the first (claimOrAwait at apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts:271) and is rejected with a 503 error if the first hasn't finished within five seconds, instead of resolving to the existing run as before.
Impact: Callers that fire duplicate idempotent triggers concurrently can receive a hard error for a request that previously succeeded and returned the already-created run.

Wait/timeout path converts a previously benign race into an error response

claimOrAwait polls for up to DEFAULT_CLAIM_WAIT_MS (5s, apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts:17) and returns { kind: "timed_out" } if the winner hasn't published. The concern then throws new ServiceValidationError("Idempotency claim resolution timed out", 503) (apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts:328-333).

Before this change the loser proceeded through the pipeline and the Postgres unique constraint produced RunDuplicateIdempotencyKeyError, which RunEngineTriggerTaskService.call retries and resolves into the cached run (apps/webapp/app/runEngine/services/triggerTask.server.ts:600-610). Any winner slower than 5s (slow payload processing, queue-limit checks, Postgres contention), or any winner that fails to publish (publish is best-effort and swallows Redis errors, apps/webapp/app/v3/mollifier/idempotencyClaim.server.ts:179-185), now turns the loser's request into a 503. Falling through to the normal pipeline on timeout would preserve the previous, safe behaviour.

Prompt for agents
In IdempotencyKeyConcern.handleTriggerRequest, a claimOrAwait timeout currently throws ServiceValidationError(503). Previously the same race was resolved harmlessly by the Postgres unique constraint on the idempotency key (RunDuplicateIdempotencyKeyError, retried by RunEngineTriggerTaskService.call, which then returns the cached run). Consider falling through to a normal (claimless) trigger on timeout — the same recovery the code already documents for the 'resolved but unfindable' case — so a slow or non-publishing winner cannot turn a duplicate trigger into a customer-visible 503.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

if (outcome.kind === "claimed") {
// Caller MUST publish/release. Signalled via the result's
// `claim` field, including the ownership token so the buffer
// can compare-and-act on the slot we now own.
return {
isCached: false,
idempotencyKey,
idempotencyKeyExpiresAt,
claim: {
envId: request.environment.id,
taskIdentifier: request.taskId,
idempotencyKey,
token: outcome.token,
},
};
}
}

return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
}
}
Loading