context.cancel & rename errors & lazy fetch

examples/ci/app/ci/constants.ts

@@ -76,4 +76,31 @@ export const TEST_ROUTES: Pick<TestConfig, "route" | "waitForSeconds">[] = [
     route: "call/workflow-with-failureUrl",
     waitForSeconds: 5
   },
+
+  /**
+   * TEST LARGE PAYLOAD CASES
+   * 
+   * disabled because they are unpredictable in CI.
+   * they are checked locally instead.
+   */
+  // {
+  //   route: "large-payload/call-result/workflow",
+  //   waitForSeconds: 9
+  // },
+  // {
+  //   route: "large-payload/error",
+  //   waitForSeconds: 9
+  // },
+  // {
+  //   route: "large-payload/initial",
+  //   waitForSeconds: 9
+  // },
+  // {
+  //   route: "large-payload/step-result",
+  //   waitForSeconds: 6
+  // },
+  // {
+  //   route: "large-payload/step-result-parallel",
+  //   waitForSeconds: 12
+  // },
 ]

examples/ci/app/test-routes/call/constants.ts

@@ -1,3 +1,6 @@
 
 export const FAILING_HEADER = "Fail-Header-Foo"
 export const FAILING_HEADER_VALUE = "fail-header-value-BAR"
+
+export const GET_HEADER = "Get-Header"
+export const GET_HEADER_VALUE = "get-header-value-FOO"

examples/ci/app/test-routes/call/third-party/route.ts

@@ -1,19 +1,24 @@
-import { FAILING_HEADER_VALUE, FAILING_HEADER } from "../constants";
+import { FAILING_HEADER_VALUE, FAILING_HEADER, GET_HEADER, GET_HEADER_VALUE } from "../constants";
 
 const thirdPartyResult = "third-party-result";
 
 export const GET = async (request: Request) => {
   return new Response(
     `called GET '${thirdPartyResult}' '${request.headers.get("get-header")}'`,
-    { status: 200 }
+    {
+      status: 200,
+      headers: {
+        [ GET_HEADER ]: GET_HEADER_VALUE
+      }
+    }
   )
 }
 
 export const POST = async (request: Request) => {
 
   return new Response(
     `called POST '${thirdPartyResult}' '${request.headers.get("post-header")}' '${await request.text()}'`,
-    { status: 200 }
+    { status: 201 }
   )
 }
 

examples/ci/app/test-routes/call/workflow/route.ts

@@ -2,7 +2,7 @@ import { serve } from "@upstash/workflow/nextjs";
 import { BASE_URL, TEST_ROUTE_PREFIX } from "app/ci/constants";
 import { testServe, expect } from "app/ci/utils";
 import { saveResult } from "app/ci/upstash/redis"
-import { FAILING_HEADER, FAILING_HEADER_VALUE } from "../constants";
+import { FAILING_HEADER, FAILING_HEADER_VALUE, GET_HEADER, GET_HEADER_VALUE } from "../constants";
 
 const testHeader = `test-header-foo`
 const headerValue = `header-foo`
@@ -28,7 +28,7 @@ export const { POST, GET } = testServe(
 
       expect(context.headers.get(testHeader)!, headerValue)
 
-      const { body: postResult } = await context.call("post call", {
+      const { body: postResult, status: postStatus } = await context.call("post call", {
         url: thirdPartyEndpoint,
         method: "POST",
         body: "post-payload",
@@ -37,18 +37,21 @@ export const { POST, GET } = testServe(
 
       // check payload after first step because we can't check above
       expect(input, payload);
-
+      expect(postStatus, 201)
+
       expect(postResult as string, 
         "called POST 'third-party-result' 'post-header-value-x' '\"post-payload\"'"
       );
-
+      
       await context.sleep("sleep 1", 2);
-
-      const { body: getResult } = await context.call("get call", {
+      
+      const { body: getResult, header: getHeaders, status: getStatus } = await context.call("get call", {
         url: thirdPartyEndpoint,
         headers: getHeader,
       });
-
+
+      expect(getStatus, 200)
+      expect(getHeaders[GET_HEADER][0], GET_HEADER_VALUE)
       expect(getResult as string, "called GET 'third-party-result' 'get-header-value-x'");
 
       const { body: patchResult, status, header } = await context.call("get call", {

examples/ci/app/test-routes/large-payload/README.md

@@ -0,0 +1,8 @@
+This directory has a endpoints testing the lazy fetch functionality:
+- `call-result`: endpoint called with context.call returns a large payload
+- `error`: a large error is thrown. failureFunction is called with the initial body.
+- `initial`: workflow is started with a large object
+- `step-result`: a step returns a large result
+- `step-result-parallel`: a parallel step returns a large result
+
+In `utils.ts`, you can find the large object used.

examples/ci/app/test-routes/large-payload/call-result/third-party/route.ts

@@ -0,0 +1,13 @@
+import { GET_HEADER, GET_HEADER_VALUE, largeObject } from "../../utils"
+
+export const GET = async () => {
+  return new Response(
+    largeObject,
+    {
+      status: 201,
+      headers: {
+        [ GET_HEADER ]: GET_HEADER_VALUE
+      }
+    }
+  )
+}

examples/ci/app/test-routes/large-payload/call-result/workflow/route.ts

@@ -0,0 +1,54 @@
+import { serve } from "@upstash/workflow/nextjs";
+import { BASE_URL, TEST_ROUTE_PREFIX } from "app/ci/constants";
+import { testServe, expect } from "app/ci/utils";
+import { saveResult } from "app/ci/upstash/redis"
+import { GET_HEADER, GET_HEADER_VALUE, largeObject, largeObjectLength } from "../../utils";
+
+const header = `test-header-foo`
+const headerValue = `header-bar`
+const payload = "“unicode-quotes”"
+
+const thirdPartyEndpoint = `${TEST_ROUTE_PREFIX}/large-payload/call-result/third-party`
+
+export const { POST, GET } = testServe(
+  serve<string>(
+    async (context) => {
+      const input = context.requestPayload;
+
+      expect(context.headers.get(header)!, headerValue)
+
+      const { body: result1, status, header: headers } = await context.call<string>("get large bod", {
+        url: thirdPartyEndpoint,
+        method: "GET"
+      })
+
+      expect(input, payload);
+
+      expect(status, 201)
+      expect(result1, largeObject)
+      expect(result1.length, largeObjectLength)
+      expect(headers[GET_HEADER][0], GET_HEADER_VALUE)
+
+      const result2 = await context.run("step2", () => {
+        return result1.length
+      });
+
+      expect(result2, largeObjectLength);
+
+      await saveResult(
+        context,
+        result2.toString()
+      )
+    }, {
+      baseUrl: BASE_URL,
+      retries: 0
+    }
+  ), {
+    expectedCallCount: 5,
+    expectedResult: largeObjectLength.toString(),
+    payload,
+    headers: {
+      [ header ]: headerValue
+    }
+  }
+)

examples/ci/app/test-routes/large-payload/error/route.ts

@@ -0,0 +1,51 @@
+import { serve } from "@upstash/workflow/nextjs";
+import { BASE_URL } from "app/ci/constants";
+import { testServe, expect } from "app/ci/utils";
+import { saveResult } from "app/ci/upstash/redis"
+import { largeObject } from "../utils";
+import { WorkflowContext } from "@upstash/workflow";
+
+const header = `test-header-foo`
+const headerValue = `header-bar`
+const payload = "“unicode-quotes”"
+
+export const { POST, GET } = testServe(
+  serve<string>(
+    async (context) => {
+      const input = context.requestPayload;
+
+      expect(input, payload);
+      expect(context.headers.get(header)!, headerValue)
+
+      const result1 = await context.run("step1", () => {
+        return input.length;
+      });
+
+      expect(result1, payload.length);
+
+      await context.run("step2", () => {
+        throw new Error(largeObject)
+      });
+    }, {
+      baseUrl: BASE_URL,
+      retries: 0,
+      async failureFunction({ context, failStatus, failResponse }) {
+        expect( failResponse, largeObject )
+        expect( failStatus, 500 )
+        expect( context.requestPayload as string, payload )
+
+        await saveResult(
+          context as WorkflowContext,
+          `super secret`
+        )
+      },
+    }
+  ), {
+    expectedCallCount: 4,
+    expectedResult: `super secret`,
+    payload,
+    headers: {
+      [ header ]: headerValue
+    }
+  }
+)

examples/ci/app/test-routes/large-payload/initial/route.ts

@@ -0,0 +1,56 @@
+import { serve } from "@upstash/workflow/nextjs";
+import { BASE_URL } from "app/ci/constants";
+import { testServe, expect } from "app/ci/utils";
+import { saveResult } from "app/ci/upstash/redis"
+import { largeObject } from "../utils";
+import { WorkflowContext } from "@upstash/workflow";
+
+const header = `test-header-foo`
+const headerValue = `header-bar`
+const throws = "throwing-foo"
+
+export const { POST, GET } = testServe(
+  serve<string>(
+    async (context) => {
+      const input = context.requestPayload;
+
+      expect(input, largeObject);
+      expect(context.headers.get(header)!, headerValue)
+
+      const result1 = await context.run("step1", () => {
+        return input.length;
+      });
+
+      expect(result1, largeObject.length);
+
+      const result2 = await context.run("step2", () => {
+        return input
+      });
+
+      expect(result2, largeObject);
+
+      await context.run("throws", () => {
+        throw new Error(throws)
+      })
+    }, {
+      baseUrl: BASE_URL,
+      retries: 0,
+      async failureFunction({ context, failResponse }) {
+        expect(context.requestPayload as string, largeObject)
+        expect(failResponse, throws)
+
+        await saveResult(
+          context as WorkflowContext,
+          throws
+        )
+      },
+    }
+  ), {
+    expectedCallCount: 5,
+    expectedResult: throws,
+    payload: largeObject,
+    headers: {
+      [ header ]: headerValue
+    }
+  }
+)

examples/ci/app/test-routes/large-payload/step-result-parallel/route.ts

@@ -0,0 +1,59 @@
+import { serve } from "@upstash/workflow/nextjs";
+import { BASE_URL } from "app/ci/constants";
+import { testServe, expect } from "app/ci/utils";
+import { saveResult } from "app/ci/upstash/redis"
+import { largeObject, largeObjectLength } from "../utils";
+
+const header = `test-header-foo`
+const headerValue = `header-bar`
+const payload = "foo"
+
+export const { POST, GET } = testServe(
+  serve<string>(
+    async (context) => {
+      const input = context.requestPayload;
+
+      expect(input, payload);
+      expect(context.headers.get(header)!, headerValue)
+
+      const [result1, largeResult1] = await Promise.all([
+        context.run("step 1", () => undefined),
+        context.run("step 2 - large", () => {
+          return largeObject;
+        })
+      ])
+
+      expect(typeof result1, "undefined");
+      expect(typeof largeResult1, "string");
+      expect(largeResult1.length, largeObjectLength);
+      expect(largeResult1, largeObject);
+
+      const [largeResult2, result2] = await Promise.all([
+        context.run("step 3 - large", () => {
+          return largeObject;
+        }),
+        context.run("step 4", () => undefined),
+      ])
+
+      expect(typeof result2, "undefined");
+      expect(typeof largeResult2, "string");
+      expect(largeResult2.length, largeObjectLength);
+      expect(largeResult2, largeObject);
+
+      await saveResult(
+        context,
+        `${largeResult1.length} - ${largeResult2.length}`
+      )
+    }, {
+      baseUrl: BASE_URL,
+      retries: 0,
+    }
+  ), {
+    expectedCallCount: 10,
+    expectedResult: `${largeObjectLength} - ${largeObjectLength}`,
+    payload,
+    headers: {
+      [ header ]: headerValue
+    }
+  }
+)