nested workflow

.github/workflows/node.js.yml

@@ -8,9 +8,9 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@08eba0b27e820071cde6df949e0beb9ba4906955 # v4
+      - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5
       - name: Use Node.js
-        uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4
+        uses: actions/setup-node@a0853c24544627f65ddf259abe73b1d18a591444 # v5
       - run: npm i
       - run: npm run build
       - run: npx pkg-pr-new publish

CHANGELOG.md

@@ -1,5 +1,20 @@
 # Changelog
 
+## 0.2.8 alpha
+
+- Adds asynchronous events - wait for an event in a workflow, send
+  events asynchronously - allows pause/resume, human-in-loop, etc.
+- Supports nested workflows with step.runWorkflow.
+- Surfaces return value of the workflow in the status
+- You can start a workflow directly from the CLI / dashboard without having to
+  make a mutation to call workflow.start:
+  - `{ fn: "path/to/file:workflowName", args: { ...your workflow args } }`
+- Reduces read bandwidth when reading the journal after running many steps in parallel.
+- Simplifies the onComplete type requirement so you can accept a workflowId as a string.
+  This helps when you have statically generated types which can't do branded strings.
+- Adds a /test entrypoint to make testing easier
+- Exports the `WorkflowCtx` and `WorkflowStep` types
+
 ## 0.2.7
 
 - Support for console logging & timing in workflows

CONTRIBUTING.md

@@ -3,14 +3,15 @@
 ## Running locally
 
 ```sh
-npm run setup
+npm i
 npm run dev
 ```
 
 ## Testing
 
 ```sh
 npm run clean
+npm run build
 npm run typecheck
 npm run lint
 npm run test

README.md

@@ -1,16 +1,18 @@
-# Convex Workflow
+# Convex Durable Workflows
 
 [![npm version](https://badge.fury.io/js/@convex-dev%2Fworkflow.svg?)](https://badge.fury.io/js/@convex-dev%2Fworkflow)
 
 <!-- START: Include on https://convex.dev/components -->
 
+The Workflow component enables you
+
 Have you ever wanted to run a series of functions reliably and durably, where
 each can have its own retry behavior, the overall workflow will survive server
 restarts, and you can have long-running workflows spanning months that can be
 canceled? Do you want to observe the status of a workflow reactively, as well as
 the results written from each step?
 
-And do you want to do this with code, instead of a DSL?
+And do you want to do this with code, instead of a static configuration?
 
 Welcome to the world of Convex workflows.
 
@@ -32,23 +34,39 @@ import { components } from "./_generated/api";
 
 export const workflow = new WorkflowManager(components.workflow);
 
-export const exampleWorkflow = workflow.define({
+export const userOnboarding = workflow.define({
   args: {
-    storageId: v.id("_storage"),
+    userId: v.id("users"),
   },
-  handler: async (step, args): Promise<number[]> => {
-    const transcription = await step.runAction(
-      internal.index.computeTranscription,
+  handler: async (ctx, args): Promise<void> => {
+    const status = await ctx.runMutation(
+      internal.emails.sendVerificationEmail,
       { storageId: args.storageId },
     );
 
-    const embedding = await step.runAction(
-      internal.index.computeEmbedding,
-      { transcription },
-      // Run this a month after the transcription is computed.
-      { runAfter: 30 * 24 * 60 * 60 * 1000 },
+    if (status === "needsVerification") {
+      // Waits until verification is completed asynchronously.
+      await ctx.awaitEvent({ name: "verificationEmail" });
+    }
+    const result = await ctx.runAction(
+      internal.llm.generateCustomContent,
+      { userId: args.userId },
+      // Retry this on transient errors with the default retry policy.
+      { retry: true },
+    );
+    if (result.needsHumanInput) {
+      // Run a whole workflow as a single step.
+      await ctx.runWorkflow(internal.llm.refineContentWorkflow, {
+        userId: args.userId,
+      });
+    }
+
+    await ctx.runMutation(
+      internal.emails.sendFollowUpEmailMaybe,
+      { userId: args.userId },
+      // Runs one day after the previous step.
+      { runAfter: 24 * 60 * 60 * 1000 },
     );
-    return embedding;
   },
 });
 ```
@@ -97,17 +115,19 @@ is designed to feel like a Convex action but with a few restrictions:
 
 1. The workflow runs in the background, so it can't return a value.
 2. The workflow must be _deterministic_, so it should implement most of its logic
-   by calling out to other Convex functions. We will be lifting some of these
-   restrictions over time by implementing `Math.random()`, `Date.now()`, and
-   `fetch` within our workflow environment.
+   by calling out to other Convex functions. We restrict access to some
+   non-deterministic functions like `Math.random()` and `fetch`. Others we
+   patch, such as `console` for logging and `Date` for time.
 
 Note: To help avoid type cycles, always annotate the return type of the `handler`
 with the return type of the workflow.
 
 ```ts
 export const exampleWorkflow = workflow.define({
   args: { name: v.string() },
+  returns: v.string(),
   handler: async (step, args): Promise<string> => {
+    //                         ^ Specify the return type of the handler
     const queryResult = await step.runQuery(
       internal.example.exampleQuery,
       args,
@@ -283,11 +303,13 @@ export const exampleWorkflow = workflow.define({
 });
 ```
 
-### Specifying how many workflows can run in parallel
+### Specifying step parallelism
 
 You can specify how many steps can run in parallel by setting the
 `maxParallelism` workpool option. It has a reasonable default.
-On the free tier, you should not exceed 20.
+On the free tier, you should not exceed 20, otherwise your other scheduled
+functions may become delayed while competing for available functions with your
+workflow steps.
 On a Pro account, you should not exceed 100 across all your workflows and workpools.
 If you want to do a lot of work in parallel, you should employ batching, where
 each workflow operates on a batch of work, e.g. scraping a list of links instead

example/convex/_generated/api.d.ts

@@ -10,6 +10,8 @@
 
 import type * as admin from "../admin.js";
 import type * as example from "../example.js";
+import type * as nestedWorkflow from "../nestedWorkflow.js";
+import type * as passingSignals from "../passingSignals.js";
 import type * as transcription from "../transcription.js";
 import type * as userConfirmation from "../userConfirmation.js";
 
@@ -30,6 +32,8 @@ import type {
 declare const fullApi: ApiFromModules<{
   admin: typeof admin;
   example: typeof example;
+  nestedWorkflow: typeof nestedWorkflow;
+  passingSignals: typeof passingSignals;
   transcription: typeof transcription;
   userConfirmation: typeof userConfirmation;
 }>;
@@ -63,7 +67,7 @@ export declare const components: {
             | { kind: "success"; returnValue: any }
             | { error: string; kind: "failed" }
             | { kind: "canceled" };
-          workflowId: string;
+          workflowId?: string;
           workpoolOptions?: {
             defaultRetryBehavior?: {
               base: number;
@@ -105,6 +109,21 @@ export declare const components: {
                   startedAt: number;
                   workId?: string;
                 }
+              | {
+                  args: any;
+                  argsSize: number;
+                  completedAt?: number;
+                  handle: string;
+                  inProgress: boolean;
+                  kind: "workflow";
+                  name: string;
+                  runResult?:
+                    | { kind: "success"; returnValue: any }
+                    | { error: string; kind: "failed" }
+                    | { kind: "canceled" };
+                  startedAt: number;
+                  workflowId?: string;
+                }
               | {
                   args: { eventId?: string };
                   argsSize: number;
@@ -118,7 +137,6 @@ export declare const components: {
                     | { error: string; kind: "failed" }
                     | { kind: "canceled" };
                   startedAt: number;
-                  workId?: string;
                 };
             stepNumber: number;
             workflowId: string;
@@ -170,6 +188,21 @@ export declare const components: {
                   startedAt: number;
                   workId?: string;
                 }
+              | {
+                  args: any;
+                  argsSize: number;
+                  completedAt?: number;
+                  handle: string;
+                  inProgress: boolean;
+                  kind: "workflow";
+                  name: string;
+                  runResult?:
+                    | { kind: "success"; returnValue: any }
+                    | { error: string; kind: "failed" }
+                    | { kind: "canceled" };
+                  startedAt: number;
+                  workflowId?: string;
+                }
               | {
                   args: { eventId?: string };
                   argsSize: number;
@@ -183,7 +216,6 @@ export declare const components: {
                     | { error: string; kind: "failed" }
                     | { kind: "canceled" };
                   startedAt: number;
-                  workId?: string;
                 };
           }>;
           workflowId: string;
@@ -218,6 +250,21 @@ export declare const components: {
                 startedAt: number;
                 workId?: string;
               }
+            | {
+                args: any;
+                argsSize: number;
+                completedAt?: number;
+                handle: string;
+                inProgress: boolean;
+                kind: "workflow";
+                name: string;
+                runResult?:
+                  | { kind: "success"; returnValue: any }
+                  | { error: string; kind: "failed" }
+                  | { kind: "canceled" };
+                startedAt: number;
+                workflowId?: string;
+              }
             | {
                 args: { eventId?: string };
                 argsSize: number;
@@ -231,7 +278,6 @@ export declare const components: {
                   | { error: string; kind: "failed" }
                   | { kind: "canceled" };
                 startedAt: number;
-                workId?: string;
               };
           stepNumber: number;
           workflowId: string;
@@ -302,6 +348,21 @@ export declare const components: {
                   startedAt: number;
                   workId?: string;
                 }
+              | {
+                  args: any;
+                  argsSize: number;
+                  completedAt?: number;
+                  handle: string;
+                  inProgress: boolean;
+                  kind: "workflow";
+                  name: string;
+                  runResult?:
+                    | { kind: "success"; returnValue: any }
+                    | { error: string; kind: "failed" }
+                    | { kind: "canceled" };
+                  startedAt: number;
+                  workflowId?: string;
+                }
               | {
                   args: { eventId?: string };
                   argsSize: number;
@@ -315,7 +376,6 @@ export declare const components: {
                     | { error: string; kind: "failed" }
                     | { kind: "canceled" };
                   startedAt: number;
-                  workId?: string;
                 };
             stepNumber: number;
             workflowId: string;
@@ -339,6 +399,45 @@ export declare const components: {
           };
         }
       >;
+      listSteps: FunctionReference<
+        "query",
+        "internal",
+        {
+          order: "asc" | "desc";
+          paginationOpts: {
+            cursor: string | null;
+            endCursor?: string | null;
+            id?: number;
+            maximumBytesRead?: number;
+            maximumRowsRead?: number;
+            numItems: number;
+          };
+          workflowId: string;
+        },
+        {
+          continueCursor: string;
+          isDone: boolean;
+          page: Array<{
+            args: any;
+            completedAt?: number;
+            eventId?: string;
+            kind: "function" | "workflow" | "event";
+            name: string;
+            nestedWorkflowId?: string;
+            runResult?:
+              | { kind: "success"; returnValue: any }
+              | { error: string; kind: "failed" }
+              | { kind: "canceled" };
+            startedAt: number;
+            stepId: string;
+            stepNumber: number;
+            workId?: string;
+            workflowId: string;
+          }>;
+          pageStatus?: "SplitRecommended" | "SplitRequired" | null;
+          splitCursor?: string | null;
+        }
+      >;
     };
   };
 };

example/convex/nestedWorkflow.ts

@@ -0,0 +1,37 @@
+import { v } from "convex/values";
+import { workflow } from "./example";
+import { internal } from "./_generated/api";
+import { internalMutation } from "./_generated/server";
+
+export const parentWorkflow = workflow.define({
+  args: { prompt: v.string() },
+  handler: async (ctx, args) => {
+    console.log("Starting confirmation workflow");
+    const length = await ctx.runWorkflow(
+      internal.nestedWorkflow.childWorkflow,
+      { foo: args.prompt },
+    );
+    console.log("Length:", length);
+    const stepResult = await ctx.runMutation(internal.nestedWorkflow.step, {
+      foo: args.prompt,
+    });
+    console.log("Step result:", stepResult);
+  },
+});
+
+export const childWorkflow = workflow.define({
+  args: { foo: v.string() },
+  returns: v.number(),
+  handler: async (_ctx, args) => {
+    console.log("Starting nested workflow");
+    return args.foo.length;
+  },
+});
+
+export const step = internalMutation({
+  args: { foo: v.string() },
+  handler: async (_ctx, args) => {
+    console.log("Starting step");
+    return args.foo.length;
+  },
+});