fix(webapp): bound metadata flush retries, add optional Redis buffer

The metadata flusher now gives up on a run after a bounded number of
version conflicts and comes back to it on the next flush, so one parent
being updated by many children at once no longer stalls metadata updates
for other runs. Operations for a run that no longer exists are dropped.

Buffered parent and root operations can optionally live in Redis instead
of process memory, so they survive a webapp restart and only one webapp
instance writes a given run at a time:

```bash
BATCH_METADATA_OPERATIONS_BUFFER_BACKEND=redis
METADATA_BUFFER_REDIS_HOST=your-redis-host  # defaults to REDIS_HOST
```

The default is still the in-process buffer.

Mono-RevId: 685c35d1d5780cb0a2a19b69da8b7308e3f911a0

Author: ericallam

.server-changes/metadata-flush-bounded-retries.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Parent and root metadata updates from child tasks no longer stall behind a parent run that is being updated by many children at once, and can optionally be buffered in Redis so they survive a webapp restart.

apps/webapp/app/env.server.ts

@@ -1144,6 +1144,32 @@ const EnvironmentSchema = z
     BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     BATCH_METADATA_OPERATIONS_FLUSH_ENABLED: z.string().default("1"),
     BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED: z.string().default("1"),
+    BATCH_METADATA_OPERATIONS_BUFFER_BACKEND: z.enum(["memory", "redis"]).default("memory"),
+    BATCH_METADATA_OPERATIONS_DRAIN_BATCH_SIZE: z.coerce.number().int().default(10),
+    BATCH_METADATA_OPERATIONS_MAX_PENDING_PER_RUN: z.coerce.number().int().default(1_000),
+    METADATA_BUFFER_REDIS_LOCK_TTL_MS: z.coerce.number().int().default(30_000),
+    METADATA_BUFFER_REDIS_HOST: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_HOST),
+    METADATA_BUFFER_REDIS_PORT: z.coerce
+      .number()
+      .optional()
+      .transform(
+        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
+      ),
+    METADATA_BUFFER_REDIS_USERNAME: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_USERNAME),
+    METADATA_BUFFER_REDIS_PASSWORD: z
+      .string()
+      .optional()
+      .transform((v) => v ?? process.env.REDIS_PASSWORD),
+    METADATA_BUFFER_REDIS_TLS_DISABLED: z
+      .string()
+      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
+    METADATA_BUFFER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
     // Run Engine 2.0
     RUN_ENGINE_WORKER_COUNT: z.coerce.number().int().default(4),

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

@@ -80,19 +80,24 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 const { action } = createActionApiRoute(
   {
     params: ParamsSchema,
+    headers: z.object({ "x-trigger-request-idempotency-key": z.string().nullish() }),
     body: UpdateMetadataRequestBody,
     maxContentLength: 1024 * 1024 * 2, // 2MB
     method: "PUT",
   },
-  async ({ authentication, body, params }) => {
+  async ({ authentication, body, params, headers }) => {
     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));
+    const [pgError, pgResult] = await tryCatch(
+      updateMetadataService.call(runId, body, env, {
+        requestIdempotencyKey: headers["x-trigger-request-idempotency-key"] ?? undefined,
+      })
+    );
     if (pgError) {
       if (pgError instanceof ServiceValidationError) {
         return json({ error: pgError.message }, { status: pgError.status ?? 422 });