feat(webapp,webhook-engine,redis-worker): queue-first webhook ingest

Webhook deliveries are acknowledged only once they are durably queued,
and the queued job records the delivery, so an acknowledged delivery can
no longer be lost or left pending, and a provider retry of a delivery
that was cut off mid-way is processed instead of dropped. The webhooks
Redis is now one cluster connection configured with
`WEBHOOK_REDIS_HOST`, `WEBHOOK_REDIS_PORT`, `WEBHOOK_REDIS_USERNAME`,
`WEBHOOK_REDIS_PASSWORD` and `WEBHOOK_REDIS_TLS_DISABLED`, replacing the
separate worker and waiter Redis settings. When an environment has too
many deliveries waiting, ingest answers 503 with a `Retry-After` header
so the provider retries later. Ingress refusals now close the
connection, so a provider's next request on a kept-alive connection no
longer hangs after a refused one.

Mono-RevId: 0b649edbe0287a4b6ca6126bdc256df884297e0d

Author: ericallam

.env.schema

@@ -112,6 +112,8 @@ REALTIME_STREAMS_DEFAULT_VERSION=v2
 # OTEL_PROMETHEUS_HOST_PORT=18889    # default 8889
 # PROMETHEUS_HOST_PORT=19090         # default 9090
 # GRAFANA_HOST_PORT=13001            # default 3001
+# --- valkey cluster (only needed with COMPOSE_PROFILES=valkey-cluster) ---
+# VALKEY_CLUSTER_1_HOST_PORT=17101   # default 7101, and 7102..7106 for _2.._6
 # (and update DATABASE_URL / CLICKHOUSE_URL / REDIS_PORT / APP_ORIGIN /
 #  LOGIN_ORIGIN / ELECTRIC_ORIGIN / REALTIME_STREAMS_S2_ENDPOINT to match)
 

apps/webapp/app/env.server.ts

@@ -1940,67 +1940,17 @@ const EnvironmentSchema = z
     WEBHOOK_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().positive().default(20),
     WEBHOOK_WORKER_TENANT_CONCURRENCY: z.coerce.number().int().positive().default(100),
     WEBHOOK_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
-    WEBHOOK_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(30_000),
     WEBHOOK_ENABLED: z.string().default("0"),
 
-    WEBHOOK_WORKER_REDIS_HOST: z
-      .string()
-      .optional()
-      .transform((v) => v ?? process.env.REDIS_HOST),
-    WEBHOOK_WORKER_REDIS_PORT: z.coerce
-      .number()
-      .optional()
-      .transform(
-        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
-      ),
-    WEBHOOK_WORKER_REDIS_USERNAME: z
-      .string()
-      .optional()
-      .transform((v) => v ?? process.env.REDIS_USERNAME),
-    WEBHOOK_WORKER_REDIS_PASSWORD: z
-      .string()
-      .optional()
-      .transform((v) => v ?? process.env.REDIS_PASSWORD),
-    WEBHOOK_WORKER_REDIS_TLS_DISABLED: z
-      .string()
-      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
-
-    // Webhook waiter store (also the ingest front gate). Falls back to the webhook worker Redis.
-    // Cluster mode connects to the host/port as a cluster seed node.
-    WEBHOOK_WAITER_REDIS_HOST: z
-      .string()
-      .optional()
-      .transform((v) => v ?? process.env.WEBHOOK_WORKER_REDIS_HOST ?? process.env.REDIS_HOST),
-    WEBHOOK_WAITER_REDIS_PORT: z.coerce
-      .number()
-      .optional()
-      .transform(
-        (v) =>
-          v ??
-          (process.env.WEBHOOK_WORKER_REDIS_PORT
-            ? parseInt(process.env.WEBHOOK_WORKER_REDIS_PORT)
-            : process.env.REDIS_PORT
-              ? parseInt(process.env.REDIS_PORT)
-              : undefined)
-      ),
-    WEBHOOK_WAITER_REDIS_USERNAME: z
-      .string()
-      .optional()
-      .transform(
-        (v) => v ?? process.env.WEBHOOK_WORKER_REDIS_USERNAME ?? process.env.REDIS_USERNAME
-      ),
-    WEBHOOK_WAITER_REDIS_PASSWORD: z
-      .string()
-      .optional()
-      .transform(
-        (v) => v ?? process.env.WEBHOOK_WORKER_REDIS_PASSWORD ?? process.env.REDIS_PASSWORD
-      ),
-    WEBHOOK_WAITER_REDIS_TLS_DISABLED: z
-      .string()
-      .default(
-        process.env.WEBHOOK_WORKER_REDIS_TLS_DISABLED ?? process.env.REDIS_TLS_DISABLED ?? "false"
-      ),
-    WEBHOOK_WAITER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+    // The webhooks Redis: a cluster (durable in production), reached through any node as a seed.
+    // It holds the job queue, the ingest front gate, waiters and session append claims.
+    WEBHOOK_REDIS_HOST: z.string().optional(),
+    WEBHOOK_REDIS_PORT: z.coerce.number().optional(),
+    WEBHOOK_REDIS_USERNAME: z.string().optional(),
+    WEBHOOK_REDIS_PASSWORD: z.string().optional(),
+    WEBHOOK_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+    // The share of a primary's maxmemory past which ingest refuses deliveries whose jobs live on it.
+    WEBHOOK_QUEUE_HIGH_WATER: z.coerce.number().positive().max(1).default(0.75),
     WEBHOOK_WAITER_MAX_PER_ENVIRONMENT: z.coerce.number().int().positive().default(1_000_000),
     WEBHOOK_WAITER_MAX_PER_ENDPOINT: z.coerce.number().int().positive().default(10_000),
     WEBHOOK_WAITER_MAX_SHAPES: z.coerce.number().int().positive().default(25),

apps/webapp/app/routes/api.v1.webhooks.deliveries.$deliveryId.replay.ts

@@ -81,6 +81,11 @@ const { action, loader } = createActionApiRoute(
           { error: `The endpoint has no subscriber "${searchParams.targetId}".` },
           { status: 404 }
         );
+      case "queue_full":
+        return json(
+          { error: "This environment has too many webhook deliveries waiting. Try again later." },
+          { status: 503, headers: { "Retry-After": "60" } }
+        );
     }
   }
 );

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.webhooks.endpoints.$endpointParam.send.ts

@@ -215,7 +215,10 @@ export async function action({ request, params }: ActionFunctionArgs): Promise<W
   switch (result.outcome) {
     case "accepted": {
       const friendlyId = result.deliveryFriendlyId;
-      if (shouldRedirect) throw redirect(deliveryPathFor(friendlyId));
+      if (shouldRedirect) {
+        await webhookEngine.waitForDeliveryRow(result.deliveryId, 3_000);
+        throw redirect(deliveryPathFor(friendlyId));
+      }
       return {
         success: true,
         httpStatus: answer.status,
@@ -260,6 +263,14 @@ export async function action({ request, params }: ActionFunctionArgs): Promise<W
       return { success: false, error: "This endpoint is not active." };
     case "enqueue_failed":
       return { success: false, error: result.error ?? "Failed to record the delivery." };
+    case "queue_full":
+    case "over_capacity":
+      return {
+        success: false,
+        error: "Too many webhook deliveries are waiting in this environment. Try again shortly.",
+        httpStatus: answer.status,
+        responseBody: answer.body ?? "",
+      };
     default:
       return { success: false, error: "The delivery could not be sent." };
   }

apps/webapp/app/routes/webhooks.v1.ingest.$opaqueId.ts

@@ -61,6 +61,13 @@ export async function action({ request, params }: ActionFunctionArgs) {
     case "enqueue_failed":
       logger.error("webhook ingress enqueue failed", { opaqueId, error: result.error });
       break;
+    case "queue_full":
+    case "over_capacity":
+      logger.warn("webhook ingress refused: queue at capacity", {
+        opaqueId,
+        reason: result.outcome,
+      });
+      break;
     default:
       break;
   }

apps/webapp/app/services/realtime/s2realtimeStreams.server.ts

@@ -644,7 +644,10 @@ export class S2RealtimeStreams implements StreamResponder, StreamIngestor {
           lastError = new Error(`S2 append failed: 409 ${res.statusText} ${text}`.trim());
         } else {
           await cancelResponseBody(res);
-          const httpError = new Error(`S2 append failed: ${res.status} ${res.statusText}`);
+          const httpError = Object.assign(
+            new Error(`S2 append failed: ${res.status} ${res.statusText}`),
+            { status: res.status }
+          );
           if (res.status >= 400 && res.status < 500) {
             throw httpError;
           }

apps/webapp/app/v3/commonWorker.server.ts

@@ -31,6 +31,7 @@ import { BulkActionService } from "./services/bulk/BulkActionV2.server";
 import { prisma } from "~/db.server";
 import { sweepPreviewAutoArchives } from "~/services/previewAutoArchive.server";
 import { controlPlaneResolver } from "./runOpsMigration/controlPlaneResolver.server";
+import { webhookEngine } from "./webhookEngine.server";
 
 function initializeWorker() {
   const redisOptions = {
@@ -54,6 +55,17 @@ function initializeWorker() {
     name: "common-worker",
     redisOptions,
     catalog: {
+      "webhooks.ensurePartitions": {
+        schema: CronSchema,
+        visibilityTimeoutMs: 60_000 * 5,
+        ...(env.WEBHOOK_ENABLED === "1"
+          ? {
+              cron: env.WEBHOOK_PARTITION_ENSURE_SCHEDULE ?? "17 3 * * *",
+              jitterInMs: env.WEBHOOK_PARTITION_ENSURE_JITTER_MS ?? 5 * 60_000,
+            }
+          : {}),
+        retry: { maxAttempts: 1 },
+      },
       "previewBranches.autoArchiveSweep": {
         schema: CronSchema,
         cron: "* * * * *",
@@ -204,6 +216,9 @@ function initializeWorker() {
     shutdownTimeoutMs: env.COMMON_WORKER_SHUTDOWN_TIMEOUT_MS,
     logger: new Logger("CommonWorker", env.COMMON_WORKER_LOG_LEVEL),
     jobs: {
+      "webhooks.ensurePartitions": async () => {
+        await webhookEngine.ensurePartitions();
+      },
       "previewBranches.autoArchiveSweep": async () => {
         await sweepPreviewAutoArchives(prisma, (ids) => {
           for (const id of ids) controlPlaneResolver.invalidateEnvironment(id);