V2 backcompatibility for V3

packages/inngest/etc/inngest.api.md

@@ -193,10 +193,12 @@ export class InngestCommHandler<Input extends any[] = any[], Output = any, Strea
     protected registerBody(url: URL): RegisterRequest;
     protected reqUrl(url: URL): URL;
     // Warning: (ae-forgotten-export) The symbol "ServerTiming" needs to be exported by the entry point index.d.ts
-    // Warning: (ae-forgotten-export) The symbol "ExecutionResult" needs to be exported by the entry point index.d.ts
     //
     // (undocumented)
-    protected runStep(functionId: string, stepId: string | null, data: unknown, timer: ServerTiming): Promise<ExecutionResult>;
+    protected runStep(functionId: string, stepId: string | null, data: unknown, timer: ServerTiming): {
+        version: ExecutionVersion;
+        result: Promise<ExecutionResult>;
+    };
     protected readonly serveHost: string | undefined;
     protected readonly servePath: string | undefined;
     protected signingKey: string | undefined;
@@ -390,6 +392,8 @@ export type ZodEventSchemas = Record<string, {
 
 // Warnings were encountered during analysis:
 //
+// src/components/InngestCommHandler.ts:803:8 - (ae-forgotten-export) The symbol "ExecutionVersion" needs to be exported by the entry point index.d.ts
+// src/components/InngestCommHandler.ts:803:35 - (ae-forgotten-export) The symbol "ExecutionResult" needs to be exported by the entry point index.d.ts
 // src/components/InngestMiddleware.ts:264:3 - (ae-forgotten-export) The symbol "InitialRunInfo" needs to be exported by the entry point index.d.ts
 // src/components/InngestMiddleware.ts:277:5 - (ae-forgotten-export) The symbol "MiddlewareRunInput" needs to be exported by the entry point index.d.ts
 // src/components/InngestMiddleware.ts:283:5 - (ae-forgotten-export) The symbol "BlankHook" needs to be exported by the entry point index.d.ts

packages/inngest/src/components/InngestCommHandler.ts

@@ -23,13 +23,14 @@ import {
   type Env,
 } from "../helpers/env";
 import { rethrowError, serializeError } from "../helpers/errors";
-import { parseFnData } from "../helpers/functions";
+import { fetchAllFnData, parseFnData } from "../helpers/functions";
 import { runAsPromise } from "../helpers/promises";
 import { createStream } from "../helpers/stream";
 import { hashSigningKey, stringify } from "../helpers/strings";
 import { type MaybePromise } from "../helpers/types";
 import {
   logLevels,
+  type EventPayload,
   type FunctionConfig,
   type IntrospectRequest,
   type LogLevel,
@@ -40,15 +41,17 @@ import {
 import { version } from "../version";
 import { type AnyInngest } from "./Inngest";
 import {
+  type AnyInngestFunction,
+  type InngestFunction,
+} from "./InngestFunction";
+import {
+  PREFERRED_EXECUTION_VERSION,
   type ExecutionResult,
   type ExecutionResultHandler,
   type ExecutionResultHandlers,
+  type ExecutionVersion,
   type InngestExecutionOptions,
-} from "./InngestExecution";
-import {
-  type AnyInngestFunction,
-  type InngestFunction,
-} from "./InngestFunction";
+} from "./execution/InngestExecution";
 
 /**
  * A set of options that can be passed to a serve handler, intended to be used
@@ -536,6 +539,13 @@ export class InngestCommHandler<
         headers: {
           ...getInngestHeaders(),
           ...res.headers,
+          ...(res.version === null
+            ? {}
+            : {
+                [headerKeys.RequestVersion]: (
+                  res.version ?? PREFERRED_EXECUTION_VERSION
+                ).toString(),
+              }),
         },
       });
 
@@ -568,6 +578,7 @@ export class InngestCommHandler<
                 status: 201,
                 headers: getInngestHeaders(),
                 body: stream,
+                version: null,
               }
             );
           });
@@ -639,6 +650,22 @@ export class InngestCommHandler<
         const body = await actions.body("processing run request");
         this.validateSignature(signature ?? undefined, body);
 
+        const fnId = await getQuerystring(
+          "processing run request",
+          queryKeys.FnId
+        );
+        if (!fnId) {
+          // TODO PrettyError
+          throw new Error("No function ID found in request");
+        }
+
+        const stepId =
+          (await getQuerystring("processing run request", queryKeys.StepId)) ||
+          null;
+
+        const { version, result } = this.runStep(fnId, stepId, body, timer);
+        const stepOutput = await result;
+
         const resultHandlers: ExecutionResultHandlers<ActionResponse> = {
           "function-rejected": (result) => {
             return {
@@ -651,6 +678,7 @@ export class InngestCommHandler<
                   : {}),
               },
               body: stringify(result.error),
+              version,
             };
           },
           "function-resolved": (result) => {
@@ -660,6 +688,7 @@ export class InngestCommHandler<
                 "Content-Type": "application/json",
               },
               body: stringify(result.data),
+              version,
             };
           },
           "step-not-found": (_result) => {
@@ -671,44 +700,32 @@ export class InngestCommHandler<
               status: 999,
               headers: { "Content-Type": "application/json" },
               body: "",
+              version,
             };
           },
           "step-ran": (result) => {
             return {
               status: 206,
               headers: { "Content-Type": "application/json" },
               body: stringify([result.step]),
+              version,
             };
           },
           "steps-found": (result) => {
             return {
               status: 206,
               headers: { "Content-Type": "application/json" },
               body: stringify(result.steps),
+              version,
             };
           },
         };
 
-        const fnId = await getQuerystring(
-          "processing run request",
-          queryKeys.FnId
-        );
-        if (!fnId) {
-          // TODO PrettyError
-          throw new Error("No function ID found in request");
-        }
-
-        const stepId =
-          (await getQuerystring("processing run request", queryKeys.StepId)) ||
-          null;
-
-        const result = await this.runStep(fnId, stepId, body, timer);
-
         const handler = resultHandlers[
-          result.type
+          stepOutput.type
         ] as ExecutionResultHandler<ActionResponse>;
 
-        return await handler(result);
+        return await handler(stepOutput);
       }
 
       if (method === "GET") {
@@ -727,6 +744,7 @@ export class InngestCommHandler<
           headers: {
             "Content-Type": "application/json",
           },
+          version: undefined,
         };
       }
 
@@ -748,6 +766,7 @@ export class InngestCommHandler<
           headers: {
             "Content-Type": "application/json",
           },
+          version: undefined,
         };
       }
     } catch (err) {
@@ -760,6 +779,7 @@ export class InngestCommHandler<
         headers: {
           "Content-Type": "application/json",
         },
+        version: undefined,
       };
     }
 
@@ -771,48 +791,68 @@ export class InngestCommHandler<
         skipDevServer: this._skipDevServer,
       }),
       headers: {},
+      version: undefined,
     };
   }
 
-  protected async runStep(
+  protected runStep(
     functionId: string,
     stepId: string | null,
     data: unknown,
     timer: ServerTiming
-  ): Promise<ExecutionResult> {
+  ): { version: ExecutionVersion; result: Promise<ExecutionResult> } {
     const fn = this.fns[functionId];
     if (!fn) {
       // TODO PrettyError
       throw new Error(`Could not find function with ID "${functionId}"`);
     }
 
-    const fndata = await parseFnData(data, this.client["inngestApi"]);
-    if (!fndata.ok) {
-      throw new Error(fndata.error);
-    }
-    const { event, events, steps, ctx } = fndata.value;
+    const immediateFnData = parseFnData(data);
+    const { version } = immediateFnData;
 
-    const stepState = Object.entries(steps ?? {}).reduce<
-      InngestExecutionOptions["stepState"]
-    >((acc, [id, data]) => {
-      return {
-        ...acc,
-        // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
-        [id]: { id, data },
-      };
-    }, {});
+    const result = runAsPromise(async () => {
+      const fnData = await fetchAllFnData(
+        immediateFnData,
+        this.client["inngestApi"]
+      );
+      if (!fnData.ok) {
+        throw new Error(fnData.error);
+      }
+      const { event, events, steps, ctx, version } = fnData.value;
+
+      const stepState = Object.entries(steps ?? {}).reduce<
+        InngestExecutionOptions["stepState"]
+      >((acc, [id, data]) => {
+        return {
+          ...acc,
+          // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment
+          [id]: { id, data },
+        };
+      }, {});
+
+      const execution = fn.fn["createExecution"](version, {
+        runId: ctx?.run_id || "",
+        data: {
+          event: event as EventPayload,
+          events: events as [EventPayload, ...EventPayload[]],
+          runId: ctx?.run_id || "",
+          attempt: ctx?.attempt ?? 0,
+        },
+        stepState,
+        requestedRunStep: stepId === "step" ? undefined : stepId || undefined,
+        timer,
+        isFailureHandler: fn.onFailure,
+        disableImmediateExecution:
+          fnData.value.version === 1
+            ? fnData.value.ctx?.disable_immediate_execution
+            : undefined,
+        stepCompletionOrder: ctx?.stack?.stack ?? [],
+      });
 
-    const execution = fn.fn["createExecution"]({
-      data: { event, events, runId: ctx?.run_id, attempt: ctx?.attempt },
-      stepState,
-      requestedRunStep: stepId === "step" ? undefined : stepId || undefined,
-      timer,
-      isFailureHandler: fn.onFailure,
-      disableImmediateExecution: fndata.value.ctx?.disable_immediate_execution,
-      stepCompletionOrder: ctx?.stack?.stack ?? [],
+      return execution.start();
     });
 
-    return execution.start();
+    return { version, result };
   }
 
   protected configs(url: URL): FunctionConfig[] {
@@ -1203,6 +1243,17 @@ export interface ActionResponse<
    * A stringified body to return.
    */
   body: TBody;
+
+  /**
+   * The version of the execution engine that was used to run this action.
+   *
+   * If the action didn't use the execution engine (for example, a GET request
+   * as a health check), this will be `undefined`.
+   *
+   * If the version should be entirely omitted from the response (for example,
+   * when sending preliminary headers when streaming), this will be `null`.
+   */
+  version: ExecutionVersion | null | undefined;
 }
 
 /**