fix(webapp): close metadata-PUT parent/root race with drainer terminal-failure delete

When the metadata-PUT route's primary buffered mutation succeeded, the
route then did a second `findRunByIdWithMollifierFallback` read to
obtain `parentTaskRunFriendlyId` / `rootTaskRunFriendlyId` for fanning
out `body.parentOperations` and `body.rootOperations` to their
respective runs. If the drainer's terminal-failure path
(`buffer.fail`) ran between the primary mutation landing on the
snapshot and the second read — atomically marking FAILED and DELing
the entry hash — the second read returned null and the route
silently skipped the entire `routeOperationsToRun` fan-out. The
caller's parent/root operations were permanently lost even though
their primary metadata write had succeeded.

Move the parent/root id capture INSIDE
`applyMetadataMutationToBufferedRun` — its CAS read already loads
the snapshot, so it can extract the ids cheaply on any iteration of
its retry loop and surface them on the `applied` outcome. The route
then uses the outcome's captured ids directly without a second
buffer round trip, closing the race window.

FriendlyIds (not internal cuids) are surfaced because the consuming
`routeOperationsToRun` helper gates on the `run_…` prefix to decide
whether to attempt the buffer fallback; cuids would skip that path.
The snapshot's `parentTaskRunId` / `rootTaskRunId` are
engine-side cuids, so `RunId.toFriendlyId` converts them — identical
to what `readFallback.server.ts` does when assembling its
SyntheticRun.

Self-fallback to `runId` matches PG semantics: the PG service routes
to `taskRun.parentTaskRun?.id ?? taskRun.id` and equivalently for
root, so a top-level run's parent/root ops land on itself rather
than being silently dropped.

Devin follow-up on PR #3756.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -204,35 +204,34 @@ const { action } = createActionApiRoute(
     // materialised by the time the child is buffered, so the existing
     // service handles them; if they're also buffered, the helper
     // recurses through the buffered mutation path).
-    const bufferedEntry = await findRunByIdWithMollifierFallback({
-      runId,
-      environmentId: env.id,
-      organizationId: env.organizationId,
-    });
-    if (bufferedEntry) {
-      // Both parent and root use the friendlyIds derived in
-      // `readFallback.server.ts` via `internalRunIdToFriendlyId` from the
-      // internal IDs the engine snapshot carries (`parentTaskRunId` /
-      // `rootTaskRunId`). The PG-side `UpdateMetadataService` would
-      // route to `taskRun.parentTaskRun?.id ?? taskRun.id` and
-      // `taskRun.rootTaskRun?.id ?? taskRun.id` respectively — i.e. fall
-      // back to the run itself when there's no parent / root. Mirror
-      // that self-fallback with `?? runId` so a top-level run's
-      // parent/root ops land on itself (matching PG semantics) instead
-      // of being silently dropped.
-      await Promise.all([
-        routeOperationsToRun(
-          bufferedEntry.parentTaskRunFriendlyId ?? runId,
-          body.parentOperations,
-          env,
-        ),
-        routeOperationsToRun(
-          bufferedEntry.rootTaskRunFriendlyId ?? runId,
-          body.rootOperations,
-          env,
-        ),
-      ]);
-    }
+    //
+    // Use the parent/root friendlyIds the buffered mutation captured
+    // during its internal read — NOT a second `findRunByIdWithMollifierFallback`
+    // call here. The drainer's terminal-failure path DELetes the entry
+    // hash atomically, so if it fires between the primary mutation
+    // landing and our route's second read, `bufferedEntry` would come
+    // back null and the route would silently drop `parentOperations` /
+    // `rootOperations` after the customer's primary mutation already
+    // landed on the snapshot. Capturing the ids in the helper's first
+    // CAS read closes that race.
+    //
+    // Self-fallback to `runId` matches PG semantics: the PG service
+    // routes to `taskRun.parentTaskRun?.id ?? taskRun.id` and
+    // `taskRun.rootTaskRun?.id ?? taskRun.id`, so a top-level run's
+    // parent/root ops land on itself rather than being silently
+    // dropped.
+    await Promise.all([
+      routeOperationsToRun(
+        bufferOutcome.parentTaskRunFriendlyId ?? runId,
+        body.parentOperations,
+        env,
+      ),
+      routeOperationsToRun(
+        bufferOutcome.rootTaskRunFriendlyId ?? runId,
+        body.rootOperations,
+        env,
+      ),
+    ]);
 
     return json({ metadata: bufferOutcome.newMetadata }, { status: 200 });
   }

apps/webapp/app/v3/mollifier/applyMetadataMutation.server.ts

@@ -1,11 +1,33 @@
 import { applyMetadataOperations } from "@trigger.dev/core/v3";
 import type { FlushedRunMetadata } from "@trigger.dev/core/v3/schemas";
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
 import type { MollifierBuffer } from "@trigger.dev/redis-worker";
 import { logger } from "~/services/logger.server";
 import { getMollifierBuffer } from "./mollifierBuffer.server";
 
+// On `applied` we surface the parent/root friendlyIds captured during
+// the snapshot read. Callers that fan parent/root metadata operations
+// out to their respective runs can use these without a second
+// `findRunByIdWithMollifierFallback` round trip — and, more importantly,
+// without racing the drainer's terminal-failure path (which atomically
+// DELetes the entry hash). Without these on the outcome the second
+// read can come back null mid-route, silently dropping the caller's
+// parentOperations / rootOperations after the primary mutation already
+// landed on the snapshot.
+//
+// FriendlyIds (not internal cuids) because the consuming
+// `routeOperationsToRun` helper gates on the `run_…` prefix to decide
+// whether to attempt the buffer fallback; cuids would skip that path.
+// The snapshot's `parentTaskRunId` / `rootTaskRunId` are engine-side
+// cuids, so we convert via `RunId.toFriendlyId` here — identical to
+// what `readFallback.server.ts` does when assembling its SyntheticRun.
 export type ApplyMetadataMutationOutcome =
-  | { kind: "applied"; newMetadata: Record<string, unknown> }
+  | {
+      kind: "applied";
+      newMetadata: Record<string, unknown>;
+      parentTaskRunFriendlyId: string | undefined;
+      rootTaskRunFriendlyId: string | undefined;
+    }
   | { kind: "not_found" }
   | { kind: "busy" }
   | { kind: "version_exhausted" }
@@ -68,6 +90,25 @@ export async function applyMetadataMutationToBufferedRun(input: {
     const currentMetadataType =
       typeof snapshot.metadataType === "string" ? snapshot.metadataType : "application/json";
 
+    // Capture parent/root ids during this read so the caller can fan
+    // parent/root operations out without a second buffer.getEntry. If
+    // the drainer's terminal-failure path runs between our CAS-write
+    // below and the route's follow-up, the entry hash would be DELd
+    // and a second read would return null — silently dropping the
+    // caller's `body.parentOperations` / `body.rootOperations`. The ids
+    // themselves are immutable for a run, so capturing them on any
+    // loop iteration is fine.
+    const snapshotParentTaskRunInternalId =
+      typeof snapshot.parentTaskRunId === "string" ? snapshot.parentTaskRunId : undefined;
+    const snapshotParentTaskRunFriendlyId = snapshotParentTaskRunInternalId
+      ? RunId.toFriendlyId(snapshotParentTaskRunInternalId)
+      : undefined;
+    const snapshotRootTaskRunInternalId =
+      typeof snapshot.rootTaskRunId === "string" ? snapshot.rootTaskRunId : undefined;
+    const snapshotRootTaskRunFriendlyId = snapshotRootTaskRunInternalId
+      ? RunId.toFriendlyId(snapshotRootTaskRunInternalId)
+      : undefined;
+
     // Match PG semantics: `body.operations` and `body.metadata` are
     // mutually exclusive on a single request. The PG service
     // (`UpdateMetadataService.#updateRunMetadata`) branches on
@@ -126,7 +167,12 @@ export async function applyMetadataMutationToBufferedRun(input: {
     });
 
     if (cas.kind === "applied") {
-      return { kind: "applied", newMetadata: metadataObject };
+      return {
+        kind: "applied",
+        newMetadata: metadataObject,
+        parentTaskRunFriendlyId: snapshotParentTaskRunFriendlyId,
+        rootTaskRunFriendlyId: snapshotRootTaskRunFriendlyId,
+      };
     }
     if (cas.kind === "not_found") return { kind: "not_found" };
     if (cas.kind === "busy") return { kind: "busy" };

apps/webapp/test/mollifierApplyMetadataMutation.test.ts

@@ -4,6 +4,7 @@ vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
 
 import { applyMetadataMutationToBufferedRun } from "~/v3/mollifier/applyMetadataMutation.server";
 import type { BufferEntry, MollifierBuffer, CasSetMetadataResult } from "@trigger.dev/redis-worker";
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
 
 // Regression for a CAS retry-exhaustion bug: the default `maxRetries`
 // was 3, matching the PG-side service, but that exhausts fast when N
@@ -261,6 +262,67 @@ describe("applyMetadataMutationToBufferedRun — retry behaviour", () => {
     expect(stub.buffer.casSetMetadata).not.toHaveBeenCalled();
   });
 
+  it("surfaces parent/root friendlyIds on `applied` so the route can fan parent/root ops without a second buffer read", async () => {
+    // Regression: the metadata route used to do a SECOND
+    // `findRunByIdWithMollifierFallback` after the primary CAS to
+    // obtain parent/root friendlyIds for `routeOperationsToRun`.
+    // If the drainer's terminal-failure path ran between the CAS and
+    // the second read, the entry hash was DELd and the second read
+    // came back null — the route silently skipped the entire
+    // parent/root fan-out, dropping `body.parentOperations` /
+    // `body.rootOperations` after the primary mutation already
+    // landed. The helper now captures the ids inside its own read
+    // loop and surfaces them on the `applied` outcome so the route
+    // never needs a second round trip.
+    //
+    // Engine-side snapshot stores internal cuids; we expect the
+    // helper to convert via `RunId.toFriendlyId` so the outcome
+    // matches what `readFallback.server.ts` would have produced.
+    const parentFriendly = RunId.generate().friendlyId;
+    const rootFriendly = RunId.generate().friendlyId;
+    const parentInternal = RunId.fromFriendlyId(parentFriendly);
+    const rootInternal = RunId.fromFriendlyId(rootFriendly);
+    const stub = makeBufferStub({
+      parentTaskRunId: parentInternal,
+      rootTaskRunId: rootInternal,
+    });
+    const result = await applyMetadataMutationToBufferedRun({
+      runId: "run_1",
+      environmentId: "env_a",
+      organizationId: "org_1",
+      maximumSize: 1024 * 1024,
+      body: { metadata: { counter: 1 } },
+      buffer: stub.buffer,
+    });
+    expect(result.kind).toBe("applied");
+    if (result.kind === "applied") {
+      expect(result.parentTaskRunFriendlyId).toBe(parentFriendly);
+      expect(result.rootTaskRunFriendlyId).toBe(rootFriendly);
+    }
+  });
+
+  it("`applied` parent/root ids are undefined when the snapshot carries neither (top-level run)", async () => {
+    // Top-level runs (parentTaskRunId/rootTaskRunId both undefined in
+    // the engine-trigger snapshot) must surface as undefined on the
+    // outcome so the route's `?? runId` self-fallback fires —
+    // matching the PG service's `taskRun.parentTaskRun?.id ??
+    // taskRun.id` semantics.
+    const stub = makeBufferStub({});
+    const result = await applyMetadataMutationToBufferedRun({
+      runId: "run_1",
+      environmentId: "env_a",
+      organizationId: "org_1",
+      maximumSize: 1024 * 1024,
+      body: { metadata: { counter: 1 } },
+      buffer: stub.buffer,
+    });
+    expect(result.kind).toBe("applied");
+    if (result.kind === "applied") {
+      expect(result.parentTaskRunFriendlyId).toBeUndefined();
+      expect(result.rootTaskRunFriendlyId).toBeUndefined();
+    }
+  });
+
   it("N-way concurrent applies all converge under default budget", async () => {
     // Simulate N parallel writers against a shared state. Each writer
     // reads, applies a delta, CAS-writes. The Lua CAS forces them to