context.cancel & rename errors & lazy fetch

src/client/index.ts

@@ -1,6 +1,6 @@
 import { NotifyResponse, Waiter } from "../types";
 import { Client as QStashClient } from "@upstash/qstash";
-import { makeGetWaitersRequest, makeNotifyRequest } from "./utils";
+import { makeCancelRequest, makeGetWaitersRequest, makeNotifyRequest } from "./utils";
 
 type ClientConfig = ConstructorParameters<typeof QStashClient>[0];
 
@@ -37,12 +37,7 @@ export class Client {
    * @returns true if workflow is succesfully deleted. Otherwise throws QStashError
    */
   public async cancel({ workflowRunId }: { workflowRunId: string }) {
-    const result = (await this.client.http.request({
-      path: ["v2", "workflows", "runs", `${workflowRunId}?cancel=true`],
-      method: "DELETE",
-      parseResponseAsJson: false,
-    })) as { error: string } | undefined;
-    return result ?? true;
+    return await makeCancelRequest(this.client.http, workflowRunId);
   }
 
   /**

src/client/utils.ts

@@ -1,5 +1,6 @@
 import { Client } from "@upstash/qstash";
-import { NotifyResponse, Waiter } from "../types";
+import { NotifyResponse, RawStep, Waiter } from "../types";
+import { WorkflowLogger } from "../logger";
 
 export const makeNotifyRequest = async (
   requester: Client["http"],
@@ -25,3 +26,24 @@ export const makeGetWaitersRequest = async (
   })) as Required<Waiter>[];
   return result;
 };
+
+export const makeCancelRequest = async (requester: Client["http"], workflowRunId: string) => {
+  const result = (await requester.request({
+    path: ["v2", "workflows", "runs", `${workflowRunId}?cancel=true`],
+    method: "DELETE",
+    parseResponseAsJson: false,
+  })) as { error: string } | undefined;
+  return result ?? true;
+};
+
+export const getSteps = async (
+  requester: Client["http"],
+  workflowRunId: string,
+  debug?: WorkflowLogger
+): Promise<RawStep[]> => {
+  await debug?.log("INFO", "ENDPOINT_START", "Pulling steps from QStash.");
+  return (await requester.request({
+    path: ["v2", "workflows", "runs", workflowRunId],
+    parseResponseAsJson: true,
+  })) as RawStep[];
+};

src/context/auto-executor.test.ts

@@ -6,7 +6,7 @@ import { MOCK_QSTASH_SERVER_URL, mockQStashServer, WORKFLOW_ENDPOINT } from "../
 import { nanoid } from "../utils";
 import { AutoExecutor } from "./auto-executor";
 import type { Step } from "../types";
-import { QStashWorkflowAbort, QStashWorkflowError } from "../error";
+import { WorkflowAbort, WorkflowError } from "../error";
 
 class SpyAutoExecutor extends AutoExecutor {
   public declare getParallelCallState;
@@ -106,7 +106,7 @@ describe("auto-executor", () => {
           const throws = context.run("attemptCharge", () => {
             return { input: context.requestPayload, success: false };
           });
-          expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+          expect(throws).rejects.toThrowError(WorkflowAbort);
         },
         responseFields: {
           status: 200,
@@ -195,7 +195,7 @@ describe("auto-executor", () => {
             context.sleep("sleep for some time", 123),
             context.sleepUntil("sleep until next day", 123_123),
           ]);
-          expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+          expect(throws).rejects.toThrowError(WorkflowAbort);
         },
         responseFields: {
           status: 200,
@@ -262,7 +262,7 @@ describe("auto-executor", () => {
             context.sleep("sleep for some time", 123),
             context.sleepUntil("sleep until next day", 123_123),
           ]);
-          expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+          expect(throws).rejects.toThrowError(WorkflowAbort);
         },
         responseFields: {
           status: 200,
@@ -314,7 +314,7 @@ describe("auto-executor", () => {
             context.sleep("sleep for some time", 123),
             context.sleepUntil("sleep until next day", 123_123),
           ]);
-          expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+          expect(throws).rejects.toThrowError(WorkflowAbort);
         },
         responseFields: {
           status: 200,
@@ -366,7 +366,7 @@ describe("auto-executor", () => {
             context.sleep("sleep for some time", 123),
             context.sleepUntil("sleep until next day", 123_123),
           ]);
-          expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+          expect(throws).rejects.toThrowError(WorkflowAbort);
         },
         responseFields: {
           status: 200,
@@ -434,7 +434,7 @@ describe("auto-executor", () => {
           return true;
         });
         expect(throws).rejects.toThrow(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible step name. Expected 'wrongName', got 'attemptCharge' from the request"
           )
         );
@@ -443,7 +443,7 @@ describe("auto-executor", () => {
         const context = getContext([initialStep, singleStep]);
         const throws = context.sleep("attemptCharge", 10);
         expect(throws).rejects.toThrow(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible step type. Expected 'SleepFor', got 'Run' from the request"
           )
         );
@@ -460,7 +460,7 @@ describe("auto-executor", () => {
           context.sleepUntil("sleep until next day", 123_123),
         ]);
         expect(throws).rejects.toThrow(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible step name. Expected 'wrongName', got 'sleep for some time' from the request"
           )
         );
@@ -474,7 +474,7 @@ describe("auto-executor", () => {
           context.sleepUntil("sleep until next day", 123_123),
         ]);
         expect(throws).rejects.toThrow(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible step type. Expected 'SleepUntil', got 'SleepFor' from the request"
           )
         );
@@ -490,7 +490,7 @@ describe("auto-executor", () => {
           context.sleep("wrongName", 10), // wrong step name
           context.sleepUntil("sleep until next day", 123_123),
         ]);
-        expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+        expect(throws).rejects.toThrowError(WorkflowAbort);
       });
       test("step type", () => {
         const context = getContext([initialStep, ...parallelSteps.slice(0, 3)]);
@@ -500,7 +500,7 @@ describe("auto-executor", () => {
           context.sleepUntil("sleep for some time", 10), // wrong step type
           context.sleepUntil("sleep until next day", 123_123),
         ]);
-        expect(throws).rejects.toThrowError(QStashWorkflowAbort);
+        expect(throws).rejects.toThrowError(WorkflowAbort);
       });
     });
 
@@ -514,7 +514,7 @@ describe("auto-executor", () => {
           context.sleepUntil("sleep until next day", 123_123),
         ]);
         expect(throws).rejects.toThrowError(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible steps detected in parallel execution: Incompatible step name. Expected 'wrongName', got 'sleep for some time' from the request\n" +
               '  > Step Names from the request: ["sleep for some time","sleep until next day"]\n' +
               '    Step Types from the request: ["SleepFor","SleepUntil"]\n' +
@@ -532,7 +532,7 @@ describe("auto-executor", () => {
           context.sleepUntil("sleep until next day", 123_123),
         ]);
         expect(throws).rejects.toThrowError(
-          new QStashWorkflowError(
+          new WorkflowError(
             "Incompatible steps detected in parallel execution: Incompatible step type. Expected 'SleepUntil', got 'SleepFor' from the request\n" +
               '  > Step Names from the request: ["sleep for some time","sleep until next day"]\n' +
               '    Step Types from the request: ["SleepFor","SleepUntil"]\n' +

src/context/auto-executor.ts

@@ -1,4 +1,4 @@
-import { QStashWorkflowAbort, QStashWorkflowError } from "../error";
+import { WorkflowAbort, WorkflowError } from "../error";
 import type { WorkflowContext } from "./context";
 import type { StepFunction, ParallelCallState, Step, WaitRequest } from "../types";
 import { type BaseLazyStep } from "./steps";
@@ -39,14 +39,14 @@ export class AutoExecutor {
    *
    * If a function is already executing (this.executingStep), this
    * means that there is a nested step which is not allowed. In this
-   * case, addStep throws QStashWorkflowError.
+   * case, addStep throws WorkflowError.
    *
    * @param stepInfo step plan to add
    * @returns result of the step function
    */
   public async addStep<TResult>(stepInfo: BaseLazyStep<TResult>) {
     if (this.executingStep) {
-      throw new QStashWorkflowError(
+      throw new WorkflowError(
         "A step can not be run inside another step." +
           ` Tried to run '${stepInfo.stepName}' inside '${this.executingStep}'`
       );
@@ -159,7 +159,7 @@ export class AutoExecutor {
 
     if (parallelCallState !== "first" && plannedParallelStepCount !== parallelSteps.length) {
       // user has added/removed a parallel step
-      throw new QStashWorkflowError(
+      throw new WorkflowError(
         `Incompatible number of parallel steps when call state was '${parallelCallState}'.` +
           ` Expected ${parallelSteps.length}, got ${plannedParallelStepCount} from the request.`
       );
@@ -194,7 +194,7 @@ export class AutoExecutor {
          */
         const planStep = this.steps.at(-1);
         if (!planStep || planStep.targetStep === undefined) {
-          throw new QStashWorkflowError(
+          throw new WorkflowError(
             `There must be a last step and it should have targetStep larger than 0.` +
               `Received: ${JSON.stringify(planStep)}`
           );
@@ -217,10 +217,10 @@ export class AutoExecutor {
           );
           await this.submitStepsToQStash([resultStep]);
         } catch (error) {
-          if (error instanceof QStashWorkflowAbort) {
+          if (error instanceof WorkflowAbort) {
             throw error;
           }
-          throw new QStashWorkflowError(
+          throw new WorkflowError(
             `Error submitting steps to QStash in partial parallel step execution: ${error}`
           );
         }
@@ -235,7 +235,7 @@ export class AutoExecutor {
          * This call to the API should be discarded: no operations are to be made. Parallel steps which are still
          * running will finish and call QStash eventually.
          */
-        throw new QStashWorkflowAbort("discarded parallel");
+        throw new WorkflowAbort("discarded parallel");
       }
       case "last": {
         /**
@@ -312,7 +312,7 @@ export class AutoExecutor {
   private async submitStepsToQStash(steps: Step[]) {
     // if there are no steps, something went wrong. Raise exception
     if (steps.length === 0) {
-      throw new QStashWorkflowError(
+      throw new WorkflowError(
         `Unable to submit steps to QStash. Provided list is empty. Current step: ${this.stepCount}`
       );
     }
@@ -359,7 +359,7 @@ export class AutoExecutor {
         parseResponseAsJson: false,
       });
 
-      throw new QStashWorkflowAbort(steps[0].stepName, steps[0]);
+      throw new WorkflowAbort(steps[0].stepName, steps[0]);
     }
 
     const result = await this.context.qstashClient.batchJSON(
@@ -415,7 +415,7 @@ export class AutoExecutor {
     });
 
     // if the steps are sent successfully, abort to stop the current request
-    throw new QStashWorkflowAbort(steps[0].stepName, steps[0]);
+    throw new WorkflowAbort(steps[0].stepName, steps[0]);
   }
 
   /**
@@ -448,7 +448,7 @@ export class AutoExecutor {
     ) {
       return result[index] as TResult;
     } else {
-      throw new QStashWorkflowError(
+      throw new WorkflowError(
         `Unexpected parallel call result while executing step ${index}: '${result}'. Expected ${lazyStepList.length} many items`
       );
     }
@@ -465,22 +465,22 @@ export class AutoExecutor {
  * from the incoming request; compare the step names and types to make sure
  * that they are the same.
  *
- * Raises `QStashWorkflowError` if there is a difference.
+ * Raises `WorkflowError` if there is a difference.
  *
  * @param lazyStep lazy step created during execution
  * @param stepFromRequest step parsed from incoming request
  */
 const validateStep = (lazyStep: BaseLazyStep, stepFromRequest: Step): void => {
   // check step name
   if (lazyStep.stepName !== stepFromRequest.stepName) {
-    throw new QStashWorkflowError(
+    throw new WorkflowError(
       `Incompatible step name. Expected '${lazyStep.stepName}',` +
         ` got '${stepFromRequest.stepName}' from the request`
     );
   }
   // check type name
   if (lazyStep.stepType !== stepFromRequest.stepType) {
-    throw new QStashWorkflowError(
+    throw new WorkflowError(
       `Incompatible step type. Expected '${lazyStep.stepType}',` +
         ` got '${stepFromRequest.stepType}' from the request`
     );
@@ -491,7 +491,7 @@ const validateStep = (lazyStep: BaseLazyStep, stepFromRequest: Step): void => {
  * validates that each lazy step and step from request has the same step
  * name and type using `validateStep` method.
  *
- * If there is a difference, raises `QStashWorkflowError` with information
+ * If there is a difference, raises `WorkflowError` with information
  * about the difference.
  *
  * @param lazySteps list of lazy steps created during parallel execution
@@ -503,12 +503,12 @@ const validateParallelSteps = (lazySteps: BaseLazyStep[], stepsFromRequest: Step
       validateStep(lazySteps[index], stepFromRequest);
     }
   } catch (error) {
-    if (error instanceof QStashWorkflowError) {
+    if (error instanceof WorkflowError) {
       const lazyStepNames = lazySteps.map((lazyStep) => lazyStep.stepName);
       const lazyStepTypes = lazySteps.map((lazyStep) => lazyStep.stepType);
       const requestStepNames = stepsFromRequest.map((step) => step.stepName);
       const requestStepTypes = stepsFromRequest.map((step) => step.stepType);
-      throw new QStashWorkflowError(
+      throw new WorkflowError(
         `Incompatible steps detected in parallel execution: ${error.message}` +
           `\n  > Step Names from the request: ${JSON.stringify(requestStepNames)}` +
           `\n    Step Types from the request: ${JSON.stringify(requestStepTypes)}` +