fix(webapp): keep a throwing bypass from failing the request

The batch item bypass re-authenticates the caller, so a transient failure there
threw out of the middleware and returned a server error instead of falling back
to normal rate limiting. The bypass now catches its own failures, and the
middleware treats a throwing bypass as "no bypass" rather than trusting callers
to honour the contract.

Redis options are now required on the middleware, dropping an unused environment
fallback so the module no longer pulls env into test import graphs.

Author: ericallam

apps/webapp/app/services/apiRateLimit.server.ts

@@ -1,3 +1,4 @@
+import { tryCatch } from "@trigger.dev/core/v3";
 import { env } from "~/env.server";
 import { batchStreamGrants } from "~/runEngine/concerns/batchStreamGrantsInstance.server";
 import { authenticateAuthorizationHeader } from "./apiAuth.server";
@@ -92,12 +93,14 @@ export const apiRateLimiter = authorizationRateLimitMiddleware({
       return false;
     }
 
-    const authenticated = await authenticateAuthorizationHeader(authorizationValue, {
-      allowPublicKey: true,
-      allowJWT: true,
-    });
+    const [authError, authenticated] = await tryCatch(
+      authenticateAuthorizationHeader(authorizationValue, {
+        allowPublicKey: true,
+        allowJWT: true,
+      })
+    );
 
-    if (!authenticated || !authenticated.ok) {
+    if (authError || !authenticated || !authenticated.ok) {
       return false;
     }
 

apps/webapp/app/services/authorizationRateLimitMiddleware.server.ts

@@ -5,7 +5,6 @@ import { Ratelimit } from "@upstash/ratelimit";
 import type { Request as ExpressRequest, Response as ExpressResponse, NextFunction } from "express";
 import { createHash } from "node:crypto";
 import { z } from "zod";
-import { env } from "~/env.server";
 import type { RedisWithClusterOptions } from "~/redis.server";
 import { logger } from "./logger.server";
 import type { Duration, Limiter } from "./rateLimiter.server";
@@ -56,7 +55,7 @@ export type RateLimiterConfig = z.infer<typeof RateLimiterConfig>;
 type LimitConfigOverrideFunction = (authorizationValue: string) => Promise<unknown>;
 
 type Options = {
-  redis?: RedisWithClusterOptions;
+  redis: RedisWithClusterOptions;
   keyPrefix: string;
   pathMatchers: (RegExp | string)[];
   pathWhiteList?: (RegExp | string)[];
@@ -184,16 +183,7 @@ export function authorizationRateLimitMiddleware({
     }),
   });
 
-  const redisClient = createRedisRateLimitClient(
-    redis ?? {
-      port: env.RATE_LIMIT_REDIS_PORT,
-      host: env.RATE_LIMIT_REDIS_HOST,
-      username: env.RATE_LIMIT_REDIS_USERNAME,
-      password: env.RATE_LIMIT_REDIS_PASSWORD,
-      tlsDisabled: env.RATE_LIMIT_REDIS_TLS_DISABLED === "true",
-      clusterMode: env.RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED === "1",
-    }
-  );
+  const redisClient = createRedisRateLimitClient(redis);
 
   return async (req: ExpressRequest, res: ExpressResponse, next: NextFunction) => {
     if (log.requests) {
@@ -255,11 +245,24 @@ export function authorizationRateLimitMiddleware({
       );
     }
 
-    if (bypass && (await bypass(req))) {
-      if (log.requests) {
-        logger.info(`RateLimiter (${keyPrefix}): bypassed ${req.path}`);
+    if (bypass) {
+      let bypassed = false;
+
+      try {
+        bypassed = await bypass(req);
+      } catch (error) {
+        logger.warn(`RateLimiter (${keyPrefix}): bypass threw, applying the limit`, {
+          path: req.path,
+          error: error instanceof Error ? error.message : String(error),
+        });
+      }
+
+      if (bypassed) {
+        if (log.requests) {
+          logger.info(`RateLimiter (${keyPrefix}): bypassed ${req.path}`);
+        }
+        return next();
       }
-      return next();
     }
 
     const hash = createHash("sha256");

internal-packages/run-engine/src/engine/tests/batchTwoPhase.test.ts

@@ -712,12 +712,16 @@ describe("RunEngine 2-Phase Batch API", () => {
         prisma
       );
 
-      await setTimeout(500);
-      const dequeued = await engine.dequeueFromWorkerQueue({
-        consumerId: "test_12345",
-        workerQueue: "main",
-      });
-      expect(dequeued.length).toBe(1);
+      await vi.waitFor(
+        async () => {
+          const dequeued = await engine.dequeueFromWorkerQueue({
+            consumerId: "test_12345",
+            workerQueue: "main",
+          });
+          expect(dequeued.length).toBe(1);
+        },
+        { timeout: 15_000, interval: 100 }
+      );
 
       const initialExecutionData = await engine.getRunExecutionData({ runId: parentRun.id });
       assertNonNullable(initialExecutionData);