feat: use GenerationID to gate stale checkpointing

Author: Linell

.changeset/tender-areas-type.md

@@ -0,0 +1,5 @@
+---
+"inngest": minor
+---
+
+More gracefully handle 'stale' checkpoints to prevent duplicated runs.

packages/inngest/src/api/api.ts

@@ -23,6 +23,18 @@ import {
 
 type FetchT = typeof fetch;
 
+/**
+ * Thrown when the executor has already requeued the current run. Returning
+ * buffered ops after this would let the executor memoize them as canonical and
+ * chain the next dispatch off this dead invocation, producing duplicates.
+ */
+export class StaleDispatchError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "StaleDispatchError";
+  }
+}
+
 const realtimeSubscriptionTokenSchema = z.object({
   jwt: z.string(),
 });
@@ -538,12 +550,14 @@ export class InngestApi {
     runId: string;
     fnId: string;
     queueItemId: string;
+    generationId?: number;
     steps: OutgoingOp[];
   }): Promise<void> {
     const body = JSON.stringify({
       run_id: args.runId,
       fn_id: args.fnId,
       qi_id: args.queueItemId,
+      ...(args.generationId ? { generation_id: args.generationId } : {}),
       steps: args.steps,
       ts: new Date().valueOf(),
     });
@@ -563,6 +577,14 @@ export class InngestApi {
     }
 
     const res = result.value;
+    // 409 means the executor has already requeued. Halt rather than returning
+    // buffered ops, which would let the executor memoize them as canonical and
+    // chain the next dispatch off this dead invocation. See EXE-1552.
+    if (res.status === 409) {
+      throw new StaleDispatchError(
+        `Stale dispatch: checkpoint returned 409 (run ${args.runId})`,
+      );
+    }
     if (!res.ok) {
       throw new Error(
         `Failed to checkpoint async: ${res.status} ${

packages/inngest/src/components/InngestCommHandler.ts

@@ -2192,6 +2192,7 @@ export class InngestCommHandler<
           },
           internalFnId: ctx?.fn_id,
           queueItemId: ctx?.qi_id,
+          generationId: ctx?.generation_id,
           stepState,
           requestedRunStep,
           timer,

packages/inngest/src/components/execution/InngestExecution.ts

@@ -127,6 +127,12 @@ export interface InngestExecutionOptions {
    */
   queueItemId?: string;
 
+  /**
+   * Echoed back on every async checkpoint POST so the executor can reject
+   * stale dispatches with 409.
+   */
+  generationId?: number;
+
   /**
    * Headers to be sent with any request to Inngest during this execution.
    */

packages/inngest/src/components/execution/engine.ts

@@ -3,6 +3,7 @@ import hashjs from "hash.js";
 import ms, { type StringValue } from "ms";
 import { z } from "zod/v3";
 
+import { StaleDispatchError } from "../../api/api.ts";
 import {
   defaultMaxRetries,
   ExecutionVersion,
@@ -91,6 +92,12 @@ import {
 
 const { sha1 } = hashjs;
 
+// Defends against `instanceof` failing across bundle boundaries.
+const isStaleDispatchError = (err: unknown): boolean =>
+  err instanceof StaleDispatchError ||
+  // biome-ignore lint/suspicious/noExplicitAny: name check across boundaries
+  (err as any)?.name === "StaleDispatchError";
+
 /**
  * Retry configuration for checkpoint operations.
  *
@@ -466,9 +473,13 @@ class InngestExecutionEngine
             runId: this.fnArg.runId,
             fnId: internalFnId,
             queueItemId,
+            generationId: this.options.generationId,
             steps,
           }),
-        CHECKPOINT_RETRY_OPTIONS,
+        {
+          ...CHECKPOINT_RETRY_OPTIONS,
+          shouldRetry: (err) => !isStaleDispatchError(err),
+        },
       );
     } else {
       throw new Error(
@@ -864,6 +875,21 @@ class InngestExecutionEngine
             this.state.checkpointingStepBuffer,
           ));
         } catch (err) {
+          // Stale dispatch: the executor has already requeued. Returning
+          // buffered ops would let it memoize them as canonical and chain the
+          // next dispatch off this dead invocation, producing duplicate step
+          // executions.
+          if (isStaleDispatchError(err)) {
+            this.devDebug("stale dispatch detected; halting execution");
+            return {
+              type: "function-rejected" as const,
+              ctx: this.fnArg,
+              ops: {},
+              error: serializeError(err),
+              retriable: false,
+            };
+          }
+
           // If checkpointing fails for any reason, fall back to returning
           // ALL buffered steps to the executor via the normal async flow.
           // The executor persists completed steps and rediscovers any

packages/inngest/src/helpers/functions.ts

@@ -160,6 +160,7 @@ export const parseFnData = (
               disable_immediate_execution: z.boolean().default(false),
               use_api: z.boolean().default(false),
               qi_id: z.string().optional(),
+              generation_id: z.number().optional(),
               stack: z
                 .object({
                   stack: z

packages/inngest/src/helpers/promises.ts

@@ -236,6 +236,7 @@ export const retryWithBackoff = async <T>(
   opts?: {
     maxAttempts?: number;
     baseDelay?: number;
+    shouldRetry?: (err: unknown) => boolean;
   },
 ): Promise<T> => {
   const maxAttempts = opts?.maxAttempts || 5;
@@ -245,7 +246,7 @@ export const retryWithBackoff = async <T>(
     try {
       return await fn();
     } catch (err) {
-      if (attempt >= maxAttempts) {
+      if (attempt >= maxAttempts || opts?.shouldRetry?.(err) === false) {
         throw err;
       }