feat(webapp,redis-worker): metadata PUT handles buffered runs (Phase C3)

Closes the last API-parity gap in the master plan.

redis-worker side:
- New casSetMetadata Lua command with optimistic lock on a
  metadataVersion entry-hash field. Returns applied / version_conflict /
  not_found / busy. Mirrors the PG-side UpdateMetadataService's CAS
  loop so concurrent metadata.increment / metadata.set / metadata.append
  calls against a buffered run never lose deltas.
- accept Lua initialises metadataVersion=0; BufferEntrySchema gains
  the field.

webapp side:
- applyMetadataMutationToBufferedRun helper does the read-apply-CAS-
  retry loop in JS, reusing the existing @trigger.dev/core
  applyMetadataOperations function (no Lua re-implementation of the 6
  operation types).
- metadata PUT route does PG-first via the existing service (which
  owns the full request shape: parent/root ops, batching, validation),
  then falls through to the buffer helper on PG miss. busy and
  version_exhausted return 503 with retry hint; not_found returns 404.
- Parent/root operations on a buffered target are fanned out to the
  snapshot's parentTaskRunId via the existing service. If the parent
  is also buffered the helper recurses. Best-effort — parent/root
  ingestion failures do not surface to the caller.

Tests: 3 new redis-worker tests covering CAS apply / version conflict /
not_found-busy paths. All 71 redis-worker mollifier + 68 webapp
mollifier tests green.

.changeset/mollifier-buffer-metadata-cas.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Add `MollifierBuffer.casSetMetadata` — optimistic-lock metadata write for buffered runs. Adds a `metadataVersion` field to the entry hash; the Lua refuses the write if the expected version has moved, returning `{ kind: "version_conflict", currentVersion }` so the caller can retry. Mirrors the PG-side `UpdateMetadataService` retry-on-conflict pattern, so concurrent `metadata.increment` / `metadata.append` / `metadata.set` calls against a buffered run never lose deltas.

.server-changes/mollifier-metadata-put-buffered.md

@@ -0,0 +1,14 @@
+---
+area: webapp
+type: feature
+---
+
+`PUT /api/v1/runs/{id}/metadata` now handles buffered runs (Phase C3). Closes the last endpoint in the mollifier API-parity master plan.
+
+PG remains canonical when the row exists — `UpdateMetadataService.call` owns the full request shape including parent/root operations, the metadataVersion CAS loop, batching, and validation. The route falls through to the buffer only when the existing service returns `undefined` (no PG row).
+
+Buffer path uses a new `applyMetadataMutationToBufferedRun` helper that mirrors the PG service's optimistic-lock pattern: read the snapshot, apply the body's `metadata` replace + `operations` deltas in JS via the existing `applyMetadataOperations` from `@trigger.dev/core`, CAS-write back via `buffer.casSetMetadata`, retry on `version_conflict` up to 3 times. Concurrent `metadata.increment` / `metadata.set` / `metadata.append` calls against the same buffered run never lose deltas.
+
+`busy` (entry is DRAINING or already materialised) and `version_exhausted` (pathological contention) return 503 with a retry hint. `not_found` returns 404.
+
+`parentOperations` and `rootOperations` on a buffered target run are fanned out to the snapshot's `parentTaskRunId` via the existing service (parent is typically PG-materialised by the time the child enters the buffer). If the parent is also buffered, the helper recurses through the same CAS path. Best-effort — parent/root ingestion failures do not surface to the caller.

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

@@ -1,6 +1,7 @@
 import type { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { tryCatch } from "@trigger.dev/core/utils";
+import type { RunMetadataChangeOperation } from "@trigger.dev/core/v3/schemas";
 import { UpdateMetadataRequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
 import { $replica } from "~/db.server";
@@ -8,6 +9,7 @@ import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { updateMetadataService } from "~/services/metadata/updateMetadataInstance.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
 import { ServiceValidationError } from "~/v3/services/common.server";
+import { applyMetadataMutationToBufferedRun } from "~/v3/mollifier/applyMetadataMutation.server";
 import { findRunByIdWithMollifierFallback } from "~/v3/mollifier/readFallback.server";
 
 const ParamsSchema = z.object({
@@ -45,10 +47,6 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
     organizationId: env.organizationId,
   });
   if (buffered) {
-    // Buffered snapshot stores metadata as the original packet shape
-    // (could be a string for application/json payloads). Pass through
-    // without re-encoding — the consumer expects the same shape PG would
-    // return.
     return json(
       {
         metadata: buffered.metadata ?? null,
@@ -61,6 +59,43 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   return json({ error: "Run not found" }, { status: 404 });
 }
 
+// Route parent/root operations to the existing PG service by directly
+// invoking it against the parent/root runId. The service ingests via
+// its batching worker, which targets PG by id. If the parent/root is
+// itself buffered we recurse through our buffered-mutation helper.
+// `_ingestion_only` flag: a synthetic body that has the operations
+// promoted to top-level `operations` so the service applies them to
+// `targetRunId` directly.
+async function routeOperationsToRun(
+  targetRunId: string | undefined,
+  operations: RunMetadataChangeOperation[] | undefined,
+  env: { id: string; organizationId: string }
+): Promise<void> {
+  if (!targetRunId || !operations || operations.length === 0) return;
+
+  // Try PG first via the existing service (this is how parent/root
+  // operations have always landed; preserve that).
+  const [error] = await tryCatch(
+    updateMetadataService.call(
+      targetRunId,
+      { operations },
+      { id: env.id, organizationId: env.organizationId } as unknown as Parameters<
+        typeof updateMetadataService.call
+      >[2]
+    )
+  );
+  if (!error) return;
+
+  // PG service threw — could be "Cannot update metadata for a completed
+  // run" or similar. If the target is buffered, route operations to its
+  // snapshot too. Best-effort; do not surface this failure to the
+  // caller — the parent/root ops are auxiliary.
+  await applyMetadataMutationToBufferedRun({
+    runId: targetRunId,
+    body: { operations },
+  });
+}
+
 const { action } = createActionApiRoute(
   {
     params: ParamsSchema,
@@ -69,23 +104,72 @@ const { action } = createActionApiRoute(
     method: "PUT",
   },
   async ({ authentication, body, params }) => {
-    const [error, result] = await tryCatch(
-      updateMetadataService.call(params.runId, body, authentication.environment)
+    const env = authentication.environment;
+    const runId = params.runId;
+
+    // PG-canonical path. If the run is in PG, the existing service
+    // owns the full request shape including parent/root operations,
+    // metadataVersion CAS, batching, validation — none of which the
+    // buffer side needs to reimplement.
+    const [pgError, pgResult] = await tryCatch(
+      updateMetadataService.call(runId, body, env)
     );
-
-    if (error) {
-      if (error instanceof ServiceValidationError) {
-        return json({ error: error.message }, { status: error.status ?? 422 });
+    if (pgError) {
+      if (pgError instanceof ServiceValidationError) {
+        return json({ error: pgError.message }, { status: pgError.status ?? 422 });
       }
-
       return json({ error: "Internal Server Error" }, { status: 500 });
     }
-
-    if (!result) {
-      return json({ error: "Task Run not found" }, { status: 404 });
+    if (pgResult) {
+      return json(pgResult, { status: 200 });
     }
 
-    return json(result, { status: 200 });
+    // PG miss. Target run is either buffered or genuinely absent.
+    const bufferOutcome = await applyMetadataMutationToBufferedRun({
+      runId,
+      body: { metadata: body.metadata, operations: body.operations },
+    });
+
+    if (bufferOutcome.kind === "not_found") {
+      return json({ error: "Task Run not found" }, { status: 404 });
+    }
+    if (bufferOutcome.kind === "busy") {
+      // Entry is materialising. Best path is to retry the PG call —
+      // the row may be visible now. We don't waste a roundtrip in
+      // the happy path, but a 503 here would be customer-visible
+      // breakage for legitimately-burst workloads. Hand back 503 with
+      // a retry hint; SDK retry policy converges.
+      return json({ error: "Run materialising, retry shortly" }, { status: 503 });
+    }
+    if (bufferOutcome.kind === "version_exhausted") {
+      // Pathological contention — many concurrent metadata writers on
+      // the same buffered runId. Surface as 503 rather than silently
+      // dropping the request.
+      return json({ error: "Metadata write contention; retry shortly" }, { status: 503 });
+    }
+
+    // Buffered metadata mutation succeeded. Fan parent/root operations
+    // out to their respective runs (parent/root are typically PG-
+    // 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) {
+      await Promise.all([
+        routeOperationsToRun(bufferedEntry.parentTaskRunId, body.parentOperations, env),
+        // The snapshot doesn't carry rootTaskRunId; fall back to parent
+        // as a rough proxy (matches the existing service's nil-coalesce
+        // behaviour where rootTaskRun defaults to the parent). Phase D
+        // / future work could thread rootTaskRunId through the snapshot.
+        routeOperationsToRun(bufferedEntry.parentTaskRunId, body.rootOperations, env),
+      ]);
+    }
+
+    return json({ metadata: bufferOutcome.newMetadata }, { status: 200 });
   }
 );
 

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

@@ -0,0 +1,90 @@
+import { applyMetadataOperations } from "@trigger.dev/core/v3";
+import type { FlushedRunMetadata } from "@trigger.dev/core/v3/schemas";
+import type { MollifierBuffer } from "@trigger.dev/redis-worker";
+import { logger } from "~/services/logger.server";
+import { getMollifierBuffer } from "./mollifierBuffer.server";
+
+export type ApplyMetadataMutationOutcome =
+  | { kind: "applied"; newMetadata: Record<string, unknown> }
+  | { kind: "not_found" }
+  | { kind: "busy" }
+  | { kind: "version_exhausted" };
+
+// Apply a metadata PUT (body.metadata replace AND/OR body.operations
+// deltas) to a buffered run's snapshot. Mirrors the PG-side
+// `UpdateMetadataService.#updateRunMetadataWithOperations` retry loop:
+// read snapshot → apply operations in JS → CAS-write back with the
+// observed `metadataVersion`. Retries on conflict; bounded by
+// `maxRetries`. The Lua CAS is the atomicity primitive — concurrent
+// callers never lose an increment / append / set.
+export async function applyMetadataMutationToBufferedRun(input: {
+  runId: string;
+  body: Pick<FlushedRunMetadata, "metadata" | "operations">;
+  buffer?: MollifierBuffer | null;
+  maxRetries?: number;
+}): Promise<ApplyMetadataMutationOutcome> {
+  const buffer = input.buffer ?? getMollifierBuffer();
+  if (!buffer) return { kind: "not_found" };
+
+  const maxRetries = input.maxRetries ?? 3;
+  for (let attempt = 0; attempt <= maxRetries; attempt++) {
+    const entry = await buffer.getEntry(input.runId);
+    if (!entry) return { kind: "not_found" };
+    if (entry.status !== "QUEUED" || entry.materialised) {
+      return { kind: "busy" };
+    }
+
+    const snapshot = JSON.parse(entry.payload) as Record<string, unknown>;
+    const currentMetadataType =
+      typeof snapshot.metadataType === "string" ? snapshot.metadataType : "application/json";
+
+    // Starting point: either the body's replace metadata, or whatever's
+    // already on the snapshot. PG-side service uses the same precedence
+    // (replace overrides existing, operations apply on top).
+    let metadataObject: Record<string, unknown>;
+    if (input.body.metadata !== undefined) {
+      metadataObject = input.body.metadata as Record<string, unknown>;
+    } else if (typeof snapshot.metadata === "string") {
+      try {
+        metadataObject = JSON.parse(snapshot.metadata) as Record<string, unknown>;
+      } catch {
+        metadataObject = {};
+      }
+    } else {
+      metadataObject = {};
+    }
+
+    if (input.body.operations?.length) {
+      const result = applyMetadataOperations(metadataObject, input.body.operations);
+      metadataObject = result.newMetadata;
+    }
+
+    const newMetadataStr = JSON.stringify(metadataObject);
+    const cas = await buffer.casSetMetadata({
+      runId: input.runId,
+      expectedVersion: entry.metadataVersion,
+      newMetadata: newMetadataStr,
+      newMetadataType: currentMetadataType,
+    });
+
+    if (cas.kind === "applied") {
+      return { kind: "applied", newMetadata: metadataObject };
+    }
+    if (cas.kind === "not_found") return { kind: "not_found" };
+    if (cas.kind === "busy") return { kind: "busy" };
+    // version_conflict — another caller wrote between our read + CAS.
+    // Loop to re-read and retry.
+    logger.debug("applyMetadataMutationToBufferedRun: version_conflict, retrying", {
+      runId: input.runId,
+      attempt,
+      observedVersion: entry.metadataVersion,
+      currentVersion: cas.currentVersion,
+    });
+  }
+
+  logger.warn("applyMetadataMutationToBufferedRun: retries exhausted", {
+    runId: input.runId,
+    maxRetries,
+  });
+  return { kind: "version_exhausted" };
+}

packages/redis-worker/src/mollifier/buffer.test.ts

@@ -1374,6 +1374,129 @@ describe("MollifierBuffer idempotency lookup", () => {
   );
 });
 
+describe("MollifierBuffer.casSetMetadata", () => {
+  redisTest(
+    "applies when expectedVersion matches; increments version; updates payload",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        entryTtlSeconds: 600,
+        logger: new Logger("test", "log"),
+      });
+      try {
+        await buffer.accept({
+          runId: "cas1",
+          envId: "env_c",
+          orgId: "org_1",
+          payload: serialiseSnapshot({ metadata: '{"v":1}', metadataType: "application/json" }),
+        });
+        const result = await buffer.casSetMetadata({
+          runId: "cas1",
+          expectedVersion: 0,
+          newMetadata: '{"v":2}',
+          newMetadataType: "application/json",
+        });
+        expect(result).toEqual({ kind: "applied", newVersion: 1 });
+
+        const entry = await buffer.getEntry("cas1");
+        expect(entry!.metadataVersion).toBe(1);
+        const payload = JSON.parse(entry!.payload) as { metadata: string };
+        expect(payload.metadata).toBe('{"v":2}');
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "returns version_conflict when expectedVersion is stale",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        entryTtlSeconds: 600,
+        logger: new Logger("test", "log"),
+      });
+      try {
+        await buffer.accept({
+          runId: "cas2",
+          envId: "env_c",
+          orgId: "org_1",
+          payload: serialiseSnapshot({}),
+        });
+        await buffer.casSetMetadata({
+          runId: "cas2",
+          expectedVersion: 0,
+          newMetadata: '{"a":1}',
+          newMetadataType: "application/json",
+        });
+
+        // Second write with stale expectedVersion = 0 must conflict.
+        const result = await buffer.casSetMetadata({
+          runId: "cas2",
+          expectedVersion: 0,
+          newMetadata: '{"a":2}',
+          newMetadataType: "application/json",
+        });
+        expect(result).toEqual({ kind: "version_conflict", currentVersion: 1 });
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "returns not_found / busy on missing or terminal entries",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        entryTtlSeconds: 600,
+        logger: new Logger("test", "log"),
+      });
+      try {
+        const nf = await buffer.casSetMetadata({
+          runId: "absent",
+          expectedVersion: 0,
+          newMetadata: "{}",
+          newMetadataType: "application/json",
+        });
+        expect(nf).toEqual({ kind: "not_found" });
+
+        await buffer.accept({
+          runId: "cas3",
+          envId: "env_c",
+          orgId: "org_1",
+          payload: serialiseSnapshot({}),
+        });
+        await buffer.pop("env_c");
+        const busy = await buffer.casSetMetadata({
+          runId: "cas3",
+          expectedVersion: 0,
+          newMetadata: "{}",
+          newMetadataType: "application/json",
+        });
+        expect(busy).toEqual({ kind: "busy" });
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+});
+
 describe("MollifierBuffer.mutateSnapshot", () => {
   redisTest(
     "returns not_found when no entry exists for the runId",

packages/redis-worker/src/mollifier/buffer.ts

@@ -27,6 +27,12 @@ export type SnapshotPatch =
 
 export type MutateSnapshotResult = "applied_to_snapshot" | "not_found" | "busy";
 
+export type CasSetMetadataResult =
+  | { kind: "applied"; newVersion: number }
+  | { kind: "version_conflict"; currentVersion: number }
+  | { kind: "not_found" }
+  | { kind: "busy" };
+
 export type AcceptResult =
   | { kind: "accepted" }
   | { kind: "duplicate_run_id" }
@@ -236,6 +242,36 @@ export class MollifierBuffer {
     throw new Error(`MollifierBuffer.mutateSnapshot: unexpected Lua return value: ${result}`);
   }
 
+  // Optimistic compare-and-swap on the snapshot's metadata. Caller reads
+  // the current metadataVersion via getEntry, applies operations in JS via
+  // `applyMetadataOperations`, then calls this with the new metadata + the
+  // expected version. Lua refuses if the version has moved (caller retries
+  // up to N times). Mirrors the PG-side `UpdateMetadataService` retry
+  // loop so concurrent increment/append operations don't lose deltas.
+  async casSetMetadata(input: {
+    runId: string;
+    expectedVersion: number;
+    newMetadata: string;
+    newMetadataType: string;
+  }): Promise<CasSetMetadataResult> {
+    const entryKey = `mollifier:entries:${input.runId}`;
+    const raw = (await this.redis.casSetMollifierMetadata(
+      entryKey,
+      String(input.expectedVersion),
+      input.newMetadata,
+      input.newMetadataType,
+    )) as string;
+    if (raw === "not_found") return { kind: "not_found" };
+    if (raw === "busy") return { kind: "busy" };
+    if (raw.startsWith("conflict:")) {
+      return { kind: "version_conflict", currentVersion: Number(raw.slice("conflict:".length)) };
+    }
+    if (raw.startsWith("applied:")) {
+      return { kind: "applied", newVersion: Number(raw.slice("applied:".length)) };
+    }
+    throw new Error(`MollifierBuffer.casSetMetadata: unexpected Lua return: ${raw}`);
+  }
+
   // Resolve a buffered run by (env, task, idempotencyKey) tuple. Used by
   // `IdempotencyKeyConcern.handleTriggerRequest` after the PG check
   // misses — same key may belong to a buffered run waiting to drain. The
@@ -370,7 +406,8 @@ export class MollifierBuffer {
           'attempts', '0',
           'createdAt', createdAt,
           'createdAtMicros', createdAtMicros,
-          'idempotencyLookupKey', idempotencyLookupKey)
+          'idempotencyLookupKey', idempotencyLookupKey,
+          'metadataVersion', '0')
         redis.call('EXPIRE', entryKey, ttlSeconds)
         -- ZSET keyed by createdAtMicros: ZPOPMIN drains oldest-first
         -- (FIFO); listing pagination uses ZREVRANGEBYSCORE with a
@@ -484,6 +521,49 @@ export class MollifierBuffer {
       `,
     });
 
+    this.redis.defineCommand("casSetMollifierMetadata", {
+      numberOfKeys: 1,
+      lua: `
+        local entryKey = KEYS[1]
+        local expectedVersion = tonumber(ARGV[1])
+        local newMetadata = ARGV[2]
+        local newMetadataType = ARGV[3]
+
+        if redis.call('EXISTS', entryKey) == 0 then
+          return 'not_found'
+        end
+
+        local status = redis.call('HGET', entryKey, 'status')
+        local materialised = redis.call('HGET', entryKey, 'materialised')
+        if status ~= 'QUEUED' or materialised == 'true' then
+          return 'busy'
+        end
+
+        local currentVersionStr = redis.call('HGET', entryKey, 'metadataVersion') or '0'
+        local currentVersion = tonumber(currentVersionStr) or 0
+        if currentVersion ~= expectedVersion then
+          return 'conflict:' .. tostring(currentVersion)
+        end
+
+        -- Write the new metadata onto the snapshot's payload JSON. We
+        -- keep the rest of the payload intact — only metadata/metadataType
+        -- change. metadataVersion is denormalised on the hash for cheap
+        -- CAS reads; it's intentionally NOT stored inside the payload
+        -- itself (PG-side metadataVersion is a column, not a JSON field).
+        local payloadJson = redis.call('HGET', entryKey, 'payload')
+        local ok, payload = pcall(cjson.decode, payloadJson)
+        if not ok then return 'busy' end
+        payload.metadata = newMetadata
+        payload.metadataType = newMetadataType
+
+        local newVersion = currentVersion + 1
+        redis.call('HSET', entryKey,
+          'payload', cjson.encode(payload),
+          'metadataVersion', tostring(newVersion))
+        return 'applied:' .. tostring(newVersion)
+      `,
+    });
+
     this.redis.defineCommand("resetMollifierIdempotency", {
       numberOfKeys: 1,
       lua: `
@@ -687,6 +767,13 @@ declare module "@internal/redis" {
       patchJson: string,
       callback?: Callback<string>,
     ): Result<string, Context>;
+    casSetMollifierMetadata(
+      entryKey: string,
+      expectedVersion: string,
+      newMetadata: string,
+      newMetadataType: string,
+      callback?: Callback<string>,
+    ): Result<string, Context>;
     resetMollifierIdempotency(
       lookupKey: string,
       entryPrefix: string,

packages/redis-worker/src/mollifier/index.ts

@@ -3,6 +3,7 @@ export {
   type MollifierBufferOptions,
   type SnapshotPatch,
   type MutateSnapshotResult,
+  type CasSetMetadataResult,
 } from "./buffer.js";
 export {
   MollifierDrainer,

packages/redis-worker/src/mollifier/schemas.ts

@@ -61,6 +61,11 @@ export const BufferEntrySchema = z.object({
   // ack Lua reads this to DEL the lookup atomically with marking the
   // entry materialised (Q5).
   idempotencyLookupKey: z.string().optional().default(""),
+  // Optimistic-lock counter for the snapshot's `metadata` field.
+  // Incremented atomically by the CAS metadata Lua. Matches the
+  // semantic of `TaskRun.metadataVersion` on the PG side (which the
+  // UpdateMetadataService uses for the same retry-on-conflict pattern).
+  metadataVersion: stringToInt.default("0"),
   lastError: stringToError.optional(),
 });