fix(webapp): mollifier trigger-time decisions (PR #3753)

Phase 2 of the mollifier rollout — move the divert decision from the
drainer (dequeue time) to the call site (trigger time). When the gate
returns mollify, the trigger pipeline writes a canonical engine.trigger
snapshot into the Redis buffer and returns a synthesised HTTP 200 to the
customer immediately. The materialised TaskRun row is created later when
the drainer replays the snapshot through engine.trigger.

Call-site changes
------------------
- Evaluate the mollifier gate inside the traceRun span instead of before
  it, so the mollified run gets a ClickHouse PARTIAL event immediately.
- Extract #buildEngineTriggerInput() so the same engine.trigger shape is
  shared between the mollify (buffer snapshot) and pass-through paths.
- Replace the Phase-1 dual-write (buffer.accept + engine.trigger) with
  mollifyTrigger(), which writes the buffer and returns a synthetic
  result. engine.trigger is no longer called on the mollify path.
- Thread idempotency claim ownership through the call pipeline: publish
  the claim with the synthesised runId on success, release on error.

Test changes
-------------
- MockTraceEventConcern now produces realistic W3C traceparent values.
- mollify dual-write test replaced with synthetic-result + no-PG test.
- Removed orphan-entry and debounce-match orphan tests.

Drive-by: MollifierBuffer no longer accepts entryTtlSeconds (the
sliding-window TTL is now managed internally by the buffer class).

Author: deepshekhardas

.server-changes/mollifier-trigger.md

@@ -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.

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -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(
@@ -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
@@ -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 };
+      }
+    }
+
     if (existingRun) {
       // The idempotency key has expired
       if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
@@ -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,
+        );
+      }
+      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 };
   }
 }