feat: add workflowRunCreatedAt to context and pass it in invoke requests

examples/ci/app/ci/constants.ts

@@ -109,6 +109,9 @@ export const TEST_ROUTES: Pick<TestConfig, RouteConfigs>[] = [
   },
   {
     route: "qstash-trigger-fetch/workflows/mainWorkflow",
+  },
+  {
+    route: "invoke-parent-failure/workflow",
   }
 
   /**

examples/ci/app/ci/utils.ts

@@ -54,7 +54,7 @@ export const getTestConfig = async (route: string) => {
   )
 
   if (response.status !== 200) {
-    throw new Error(`Failed to get the error config: ${response.statusText}`)
+    throw new Error(`Failed to get the test config: ${response.statusText}`)
   }
 
   const testConfig = await response.json() as Parameters<typeof testServe>[1]

examples/ci/app/test-routes/invoke-parent-failure/[...]/route.ts

@@ -0,0 +1,98 @@
+
+import { Client, WorkflowContext, WorkflowNonRetryableError } from "@upstash/workflow";
+import { createWorkflow, serveMany } from "@upstash/workflow/nextjs";
+import { BASE_URL, CI_RANDOM_ID_HEADER, CI_ROUTE_HEADER } from "app/ci/constants";
+import { saveResult } from "app/ci/upstash/redis";
+import { testServe } from "app/ci/utils";
+
+const client = new Client({
+  baseUrl: process.env.QSTASH_URL!,
+  token: process.env.QSTASH_TOKEN!,
+})
+
+const FAILING_STEP_NAME = "failing step"
+const INVOKE_CHILD_STEP_NAME = "invoke child"
+
+const workflow = createWorkflow(async (context: WorkflowContext) => {
+  const workflowRunId = await context.run("step 1", async () => {
+    console.log("workflow says hi")
+    return `workflow-run-id-${(Math.random() * 1000).toFixed(0)}`
+  })
+
+  await context.invoke("invoke child", {
+    workflow: testWorkflow,
+    body: undefined,
+    headers: {
+      [CI_ROUTE_HEADER]: context.headers.get(CI_ROUTE_HEADER) as string,
+      [CI_RANDOM_ID_HEADER]: context.headers.get(CI_RANDOM_ID_HEADER) as string,
+    },
+    workflowRunId,
+  })
+
+  await context.run("verify step", async () => {
+
+    for (let i = 0; i < 30; i++) {
+      const workflowLogs = await client.logs({ workflowRunId: `wfr_${workflowRunId}` })
+      const workflowRun = workflowLogs.runs[0]
+
+      if (workflowRun && workflowRun.steps[1] && workflowRun.steps[1].type === "parallel") {
+        const invokeChild = workflowRun.steps[1].steps?.find(s => s.stepName === INVOKE_CHILD_STEP_NAME)
+        const failingStep = workflowRun.steps[1].steps?.find(s => s.stepName === FAILING_STEP_NAME)
+
+        if (invokeChild && invokeChild.state === "STEP_SUCCESS" && failingStep && failingStep.state === "STEP_FAILED") {
+          return { success: true }
+        }
+      }
+      // sleep for 1 sec
+      await new Promise(r => setTimeout(r, 1000));
+    }
+
+    console.warn("child workflow did not fail within expected time");
+    throw new WorkflowNonRetryableError("child workflow did not fail within expected time")
+  })
+
+  await saveResult(
+    context,
+    "done invoke"
+  )
+})
+
+const testWorkflow = createWorkflow(async (context: WorkflowContext) => {
+  await Promise.all([
+    context.invoke(INVOKE_CHILD_STEP_NAME, {
+      workflow: childWorkflow,
+      body: undefined,
+      headers: {
+        [CI_ROUTE_HEADER]: context.headers.get(CI_ROUTE_HEADER) as string,
+        [CI_RANDOM_ID_HEADER]: context.headers.get(CI_RANDOM_ID_HEADER) as string,
+      },
+    }),
+    context.run(FAILING_STEP_NAME, async () => {
+      // sleep for 1 sec
+      await new Promise(r => setTimeout(r, 1000));
+      throw new WorkflowNonRetryableError("step failed")
+    })
+  ])
+
+})
+
+const childWorkflow = createWorkflow(async (context: WorkflowContext) => {
+  await context.sleep("sleep 3s", 3)
+  await context.run("child step", async () => {
+    console.log("child workflow step")
+  })
+})
+
+export const { POST, GET } = testServe(serveMany({
+  workflow,
+  testWorkflow,
+  childWorkflow,
+}, {
+  baseUrl: BASE_URL
+}),
+  {
+    expectedCallCount: 11,
+    expectedResult: "done invoke",
+    payload: undefined,
+  }
+)

src/client/index.test.ts

@@ -512,6 +512,7 @@ describe("workflow client", () => {
             workflowRunId,
             steps: [],
             url: "https://httpstat.us/200",
+            workflowRunCreatedAt: 0,
           }),
         });
 
@@ -656,6 +657,7 @@ describe("workflow client", () => {
             workflowRunId,
             steps: [],
             url: "https://httpstat.us/400",
+            workflowRunCreatedAt: 0,
           }),
           failureUrl: "https://400check.requestcatcher.com/",
           retries: 0,

src/client/index.ts

@@ -245,6 +245,7 @@ export class Client {
         workflowRunId: finalWorkflowRunId,
         telemetry: option.disableTelemetry ? undefined : { sdk: SDK_TELEMETRY },
         label: option.label,
+        workflowRunCreatedAt: Date.now(), // pass a timestamp (server will override it)
       });
 
       return {

src/constants.ts

@@ -3,6 +3,7 @@ import { Telemetry } from "./types";
 export const WORKFLOW_ID_HEADER = "Upstash-Workflow-RunId";
 export const WORKFLOW_INIT_HEADER = "Upstash-Workflow-Init";
 export const WORKFLOW_URL_HEADER = "Upstash-Workflow-Url";
+export const WORKFLOW_CREATED_AT_HEADER = "Upstash-Workflow-CreatedAt";
 export const WORKFLOW_FAILURE_HEADER = "Upstash-Workflow-Is-Failure";
 export const WORKFLOW_FAILURE_CALLBACK_HEADER = "Upstash-Workflow-Failure-Callback";
 export const WORKFLOW_FEATURE_HEADER = "Upstash-Feature-Set";

src/context/auto-executor.test.ts

@@ -92,6 +92,7 @@ describe("auto-executor", () => {
       steps,
       url: WORKFLOW_ENDPOINT,
       invokeCount: 7,
+      workflowRunCreatedAt: 0,
     });
   };
 

src/context/context.test.ts

@@ -24,6 +24,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers({ "upstash-label": label }) as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
       label,
     });
     expect(context.label).toBe(label);
@@ -43,6 +44,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
     });
 
     const throws = async () => {
@@ -67,6 +69,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
     });
 
     const throws = async () => {
@@ -89,6 +92,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
     });
 
     const throws = async () => {
@@ -111,6 +115,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
     });
 
     const throws = async () => {
@@ -134,6 +139,7 @@ describe("context tests", () => {
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
       invokeCount: 5,
+      workflowRunCreatedAt: 0,
     });
 
     await mockQStashServer({
@@ -183,6 +189,7 @@ describe("context tests", () => {
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
         invokeCount: 5,
+        workflowRunCreatedAt: 0,
       });
 
       const eventId = "my-event-id";
@@ -235,6 +242,7 @@ describe("context tests", () => {
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
         invokeCount: 1,
+        workflowRunCreatedAt: 0,
       });
 
       const eventId = "my-event-id";
@@ -306,6 +314,7 @@ describe("context tests", () => {
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
         invokeCount: 7,
+        workflowRunCreatedAt: 0,
       });
       await mockQStashServer({
         execute: () => {
@@ -384,6 +393,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
       await mockQStashServer({
         execute: () => {
@@ -445,6 +455,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
       await mockQStashServer({
         execute: () => {
@@ -509,6 +520,7 @@ describe("context tests", () => {
       url: WORKFLOW_ENDPOINT,
       headers: new Headers() as Headers,
       workflowRunId: "wfr-id",
+      workflowRunCreatedAt: 0,
     });
     try {
       await context.cancel();
@@ -530,6 +542,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       await mockQStashServer({
@@ -559,6 +572,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       await mockQStashServer({
@@ -589,6 +603,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       const openAIToken = `hello-there`;
@@ -670,6 +685,7 @@ describe("context tests", () => {
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
         invokeCount: 5,
+        workflowRunCreatedAt: 0,
       });
 
       const openAIToken = `hello-there`;
@@ -753,6 +769,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       const resendToken = `hello-there`;
@@ -832,6 +849,7 @@ describe("context tests", () => {
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
         invokeCount: 3,
+        workflowRunCreatedAt: 0,
       });
 
       const anthropicToken = `hello-there`;
@@ -925,6 +943,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       await mockQStashServer({
@@ -972,6 +991,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       const webhook = {
@@ -1028,6 +1048,7 @@ describe("context tests", () => {
         url: WORKFLOW_ENDPOINT,
         headers: new Headers() as Headers,
         workflowRunId: "wfr-id",
+        workflowRunCreatedAt: 0,
       });
 
       // First create webhook

src/context/context.ts

@@ -60,6 +60,10 @@ export class WorkflowContext<TInitialPayload = unknown> {
    * Run id of the workflow
    */
   public readonly workflowRunId: string;
+  /**
+   * Creation time of the workflow run
+   */
+  public readonly workflowRunCreatedAt: number;
   /**
    * URL of the workflow
    *
@@ -153,6 +157,7 @@ export class WorkflowContext<TInitialPayload = unknown> {
   constructor({
     qstashClient,
     workflowRunId,
+    workflowRunCreatedAt,
     headers,
     steps,
     url,
@@ -165,6 +170,7 @@ export class WorkflowContext<TInitialPayload = unknown> {
   }: {
     qstashClient: WorkflowClient;
     workflowRunId: string;
+    workflowRunCreatedAt: number;
     headers: Headers;
     steps: Step[];
     url: string;
@@ -177,6 +183,7 @@ export class WorkflowContext<TInitialPayload = unknown> {
   }) {
     this.qstashClient = qstashClient;
     this.workflowRunId = workflowRunId;
+    this.workflowRunCreatedAt = workflowRunCreatedAt;
     this.steps = steps;
     this.url = url;
     this.headers = headers;

src/context/steps.test.ts

@@ -30,6 +30,7 @@ const createMockContext = () => {
     url: WORKFLOW_ENDPOINT,
     headers: new Headers() as Headers,
     workflowRunId: "test-wfr-id",
+    workflowRunCreatedAt: 0,
   });
 };
 

src/context/steps.ts

@@ -714,6 +714,7 @@ export class LazyInvokeStep<TResult = unknown, TBody = unknown> extends BaseLazy
         Object.entries(invokerHeaders).map((pairs) => [pairs[0], [pairs[1]]])
       ),
       workflowRunId: context.workflowRunId,
+      workflowRunCreatedAt: context.workflowRunCreatedAt,
       workflowUrl: context.url,
       step,
     };