fix(batch): rate limiting by token bucket no longer incorrectly goes negative (#2837)

Also improves the BatchTriggerError when a result of getting rate
limited.

Author: ericallam

.changeset/sharp-ravens-doubt.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Improve batch trigger error messages, especially when rate limited

apps/webapp/app/routes/api.v3.batches.ts

@@ -158,7 +158,7 @@ const { action, loader } = createActionApiRoute(
             status: 429,
             headers: {
               "X-RateLimit-Limit": error.limit.toString(),
-              "X-RateLimit-Remaining": error.remaining.toString(),
+              "X-RateLimit-Remaining": Math.max(0, error.remaining).toString(),
               "X-RateLimit-Reset": Math.floor(error.resetAt.getTime() / 1000).toString(),
               "Retry-After": Math.max(
                 1,

apps/webapp/app/runEngine/concerns/batchLimits.server.ts

@@ -115,9 +115,7 @@ export class BatchRateLimitExceededError extends Error {
     public readonly resetAt: Date,
     public readonly itemCount: number
   ) {
-    super(
-      `Batch rate limit exceeded. Attempted to submit ${itemCount} items but only ${remaining} remaining. Limit resets at ${resetAt.toISOString()}`
-    );
+    super(`Batch rate limit exceeded. Limit resets at ${resetAt.toISOString()}`);
     this.name = "BatchRateLimitExceededError";
   }
 }

package.json

@@ -79,7 +79,8 @@
       "[email protected]": "patches/[email protected]",
       "[email protected]": "patches/[email protected]",
       "@kubernetes/[email protected]": "patches/@[email protected]",
-      "@sentry/[email protected]": "patches/@[email protected]"
+      "@sentry/[email protected]": "patches/@[email protected]",
+      "@upstash/[email protected]": "patches/@upstash__ratelimit.patch"
     },
     "overrides": {
       "express@^4>body-parser": "1.20.3",

packages/trigger-sdk/src/v3/index.ts

@@ -1,7 +1,7 @@
 export * from "./cache.js";
 export * from "./config.js";
 export { retry, type RetryOptions } from "./retry.js";
-export { queue } from "./shared.js";
+export { queue, BatchTriggerError } from "./shared.js";
 export * from "./tasks.js";
 export * from "./batch.js";
 export * from "./wait.js";

packages/trigger-sdk/src/v3/shared.ts

@@ -2,6 +2,7 @@ import { SpanKind } from "@opentelemetry/api";
 import { SerializableJson } from "@trigger.dev/core";
 import {
   accessoryAttributes,
+  ApiError,
   apiClientManager,
   ApiRequestOptions,
   conditionallyImportPacket,
@@ -17,6 +18,7 @@ import {
   parsePacket,
   Queue,
   QueueOptions,
+  RateLimitError,
   resourceCatalog,
   runtime,
   SemanticInternalAttributes,
@@ -125,9 +127,6 @@ export { SubtaskUnwrapError, TaskRunPromise };
 
 export type Context = TaskRunContext;
 
-// Re-export for external use (defined later in file)
-export { BatchTriggerError };
-
 export function queue(options: QueueOptions): Queue {
   resourceCatalog.registerQueueMetadata(options);
 
@@ -1592,12 +1591,28 @@ async function executeBatchTwoPhase(
 /**
  * Error thrown when batch trigger operations fail.
  * Includes context about which phase failed and the batch details.
+ *
+ * When the underlying error is a rate limit (429), additional properties are exposed:
+ * - `isRateLimited`: true
+ * - `retryAfterMs`: milliseconds until the rate limit resets
  */
-class BatchTriggerError extends Error {
+export class BatchTriggerError extends Error {
   readonly phase: "create" | "stream";
   readonly batchId?: string;
   readonly itemCount: number;
 
+  /** True if the error was caused by rate limiting (HTTP 429) */
+  readonly isRateLimited: boolean;
+
+  /** Milliseconds until the rate limit resets. Only set when `isRateLimited` is true. */
+  readonly retryAfterMs?: number;
+
+  /** The underlying API error, if the cause was an ApiError */
+  readonly apiError?: ApiError;
+
+  /** The underlying cause of the error */
+  override readonly cause?: unknown;
+
   constructor(
     message: string,
     options: {
@@ -1607,12 +1622,59 @@ class BatchTriggerError extends Error {
       itemCount: number;
     }
   ) {
-    super(message, { cause: options.cause });
+    // Build enhanced message that includes the cause's message
+    const fullMessage = buildBatchErrorMessage(message, options.cause);
+    super(fullMessage, { cause: options.cause });
+
     this.name = "BatchTriggerError";
+    this.cause = options.cause;
     this.phase = options.phase;
     this.batchId = options.batchId;
     this.itemCount = options.itemCount;
+
+    // Extract rate limit info from cause
+    if (options.cause instanceof RateLimitError) {
+      this.isRateLimited = true;
+      this.retryAfterMs = options.cause.millisecondsUntilReset;
+      this.apiError = options.cause;
+    } else if (options.cause instanceof ApiError) {
+      this.isRateLimited = options.cause.status === 429;
+      this.apiError = options.cause;
+    } else {
+      this.isRateLimited = false;
+    }
+  }
+}
+
+/**
+ * Build an enhanced error message that includes context from the cause.
+ */
+function buildBatchErrorMessage(baseMessage: string, cause: unknown): string {
+  if (!cause) {
+    return baseMessage;
+  }
+
+  // Handle RateLimitError specifically for better messaging
+  if (cause instanceof RateLimitError) {
+    const retryMs = cause.millisecondsUntilReset;
+    if (retryMs !== undefined) {
+      const retrySeconds = Math.ceil(retryMs / 1000);
+      return `${baseMessage}: Rate limit exceeded - retry after ${retrySeconds}s`;
+    }
+    return `${baseMessage}: Rate limit exceeded`;
   }
+
+  // Handle other ApiErrors
+  if (cause instanceof ApiError) {
+    return `${baseMessage}: ${cause.message}`;
+  }
+
+  // Handle generic errors
+  if (cause instanceof Error) {
+    return `${baseMessage}: ${cause.message}`;
+  }
+
+  return baseMessage;
 }
 
 /**

patches/@upstash__ratelimit.patch

@@ -0,0 +1,26 @@
+diff --git a/dist/index.js b/dist/index.js
+index 7d1502426320957017988aed0c29974acd70e8da..062769cda055302d737503e5d1ba5e62609c934f 100644
+--- a/dist/index.js
++++ b/dist/index.js
+@@ -841,7 +841,7 @@ var tokenBucketLimitScript = `
+     refilledAt = refilledAt + numRefills * interval
+   end
+ 
+-  if tokens == 0 then
++  if tokens < incrementBy then
+     return {-1, refilledAt + interval}
+   end
+ 
+diff --git a/dist/index.mjs b/dist/index.mjs
+index 25a2c888be27b7c5aff41de63d5df189e0031145..53b4a4b2d2ef55f709f7404cc6a66058b7f3191a 100644
+--- a/dist/index.mjs
++++ b/dist/index.mjs
+@@ -813,7 +813,7 @@ var tokenBucketLimitScript = `
+     refilledAt = refilledAt + numRefills * interval
+   end
+ 
+-  if tokens == 0 then
++  if tokens < incrementBy then
+     return {-1, refilledAt + interval}
+   end
+