Merge branch 'complete'

Author: ianmacartney

.mdc/convex_rules.mdc

@@ -95,8 +95,8 @@ await pool.enqueueAction(ctx, internal.email.send, args, {
 export const emailSent = internalMutation({
   args: {
     workId: workIdValidator,
+    result: resultValidator,
     context: v.object({ emailType: v.string(), userId: v.id("users") }),
-    result: runResultValidator,
   },
   handler: async (ctx, args) => {
     if (args.result.kind === "canceled") return;
@@ -303,25 +303,16 @@ You can read the status of a function by calling `pool.status(id)`.
 
 The status will be one of:
 
-- `{ kind: "pending" }`: The function has not started yet.
-- `{ kind: "inProgress" }`: The function is currently running.
-- `{ kind: "completed"; completionStatus: CompletionStatus }`: The function has
-  finished.
-
-The `CompletionStatus` type is one of:
-
-- `"success"`: The function completed successfully.
-- `"error"`: The function threw an error.
-- `"canceled"`: The function was canceled.
-- `"timeout"`: The function timed out.
+- `{ kind: "pending"; previousAttempts: number }`: The function has not started yet.
+- `{ kind: "running"; previousAttempts: number }`: The function is currently running.
+- `{ kind: "finished" }`: The function has succeeded, failed, or been canceled.
 
 ## Canceling work
 
-You can cancel work by calling `pool.cancel(id)`.
+You can cancel work by calling `pool.cancel(id)` or all of them with
+`pool.cancelAll()`.
 
-This will remove the work from the queue and mark it as canceled.
-If the work has already started, it will wait for it to finish, but not allow
-retries.
+This will avoid starting or retrying, but will not stop in-progress work.
 
 <!-- END: Include on https://convex.dev/components -->
 

README.md

@@ -80,11 +80,14 @@ export declare const components: {
         "query",
         "internal",
         { id: string },
-        | { attempt: number; state: "pending" }
-        | { attempt: number; state: "running" }
+        | { previousAttempts: number; state: "pending" }
+        | { previousAttempts: number; state: "running" }
         | { state: "finished" }
       >;
     };
+    stats: {
+      queueLength: FunctionReference<"query", "internal", {}, number>;
+    };
   };
   bigPool: {
     lib: {
@@ -126,11 +129,14 @@ export declare const components: {
         "query",
         "internal",
         { id: string },
-        | { attempt: number; state: "pending" }
-        | { attempt: number; state: "running" }
+        | { previousAttempts: number; state: "pending" }
+        | { previousAttempts: number; state: "running" }
         | { state: "finished" }
       >;
     };
+    stats: {
+      queueLength: FunctionReference<"query", "internal", {}, number>;
+    };
   };
   serializedPool: {
     lib: {
@@ -172,10 +178,13 @@ export declare const components: {
         "query",
         "internal",
         { id: string },
-        | { attempt: number; state: "pending" }
-        | { attempt: number; state: "running" }
+        | { previousAttempts: number; state: "pending" }
+        | { previousAttempts: number; state: "running" }
         | { state: "finished" }
       >;
     };
+    stats: {
+      queueLength: FunctionReference<"query", "internal", {}, number>;
+    };
   };
 };

example/convex/_generated/api.d.ts

@@ -33,7 +33,7 @@ describe("workpool", () => {
     const id = await t.mutation(api.example.enqueueOneMutation, { data: 1 });
     expect(await t.query(api.example.status, { id })).toEqual({
       state: "pending",
-      attempt: 0,
+      previousAttempts: 0,
     });
     expect(await t.query(api.example.queryData, {})).toEqual([]);
     await t.finishAllScheduledFunctions(vi.runAllTimers);
@@ -49,9 +49,11 @@ describe("workpool", () => {
       await t.mutation(api.example.enqueueOneMutation, { data: i });
     }
     await t.finishAllScheduledFunctions(vi.runAllTimers);
-    expect(await t.query(api.example.queryData, {})).toEqual([
-      0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
-    ]);
+    expect(await t.query(api.example.queryData, {})).toEqual(
+      expect.arrayContaining([
+        0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
+      ])
+    );
   });
 
   test("enqueueMany with high parallelism", async () => {

example/convex/example.test.ts

@@ -10,7 +10,7 @@ import {
   WorkId,
   workIdValidator,
   Workpool,
-  runResultValidator,
+  resultValidator,
 } from "@convex-dev/workpool";
 import { v } from "convex/values";
 import { createLogger } from "../../src/component/logging";
@@ -24,19 +24,19 @@ const bigPool = new Workpool(components.bigPool, {
     base: 2,
   },
   retryActionsByDefault: true,
-  logLevel: "DEBUG",
+  logLevel: "WARN",
 });
 const smallPool = new Workpool(components.smallPool, {
   maxParallelism: 3,
   retryActionsByDefault: true,
-  logLevel: "INFO",
+  logLevel: "WARN",
 });
 const serializedPool = new Workpool(components.serializedPool, {
   maxParallelism: 1,
   retryActionsByDefault: true,
-  logLevel: "INFO",
+  logLevel: "WARN",
 });
-const console = createLogger("WARN");
+const console = createLogger("DEBUG");
 
 export const addMutation = mutation({
   args: { data: v.optional(v.number()) },
@@ -235,10 +235,10 @@ export const onComplete = internalMutation({
   args: {
     workId: workIdValidator,
     context: v.number(),
-    result: runResultValidator,
+    result: resultValidator,
   },
   handler: async (ctx, args) => {
-    console.warn("total", (Date.now() - args.context) / 1000);
+    console.info("total", (Date.now() - args.context) / 1000);
   },
 });
 
@@ -274,7 +274,7 @@ export const runPaced = internalAction({
 
 export const cancel = internalAction({
   args: {
-    id: workIdValidator,
+    id: v.optional(workIdValidator),
   },
   handler: async (ctx, args) => {
     console.debug("Canceling", args.id);

example/convex/example.ts

@@ -7,7 +7,7 @@
     "email": "support@convex.dev",
     "url": "https://github.com/get-convex/workpool/issues"
   },
-  "version": "0.2.0-beta.0",
+  "version": "0.2.0",
   "license": "Apache-2.0",
   "keywords": [
     "convex",
@@ -24,14 +24,15 @@
     "build:cjs": "tsc --project ./commonjs.json && echo '{\\n  \"type\": \"commonjs\"\\n}' > dist/commonjs/package.json",
     "dev": "cd example; npm run dev",
     "typecheck": "tsc --noEmit",
-    "prepare": "npm run build",
-    "prepack": "node node10stubs.mjs",
-    "postpack": "node node10stubs.mjs --cleanup",
-    "test": "vitest",
     "lint": "tsc --noEmit && eslint . && prettier --check .",
     "format": "prettier --write .",
+    "test": "vitest run",
+    "test:watch": "vitest",
     "test:debug": "vitest --inspect-brk --no-file-parallelism",
-    "test:coverage": "vitest run --coverage --coverage.reporter=text"
+    "test:coverage": "vitest run --coverage --coverage.reporter=text",
+    "prepare": "npm run build",
+    "prepack": "node node10stubs.mjs",
+    "postpack": "node node10stubs.mjs --cleanup"
   },
   "files": [
     "dist",

package.json

@@ -8,19 +8,22 @@ import {
 import { v, VString } from "convex/values";
 import { Mounts } from "../component/_generated/api.js";
 import {
+  DEFAULT_LOG_LEVEL,
+  type LogLevel,
+  logLevel,
+} from "../component/logging.js";
+import {
+  Config,
+  DEFAULT_MAX_PARALLELISM,
   OnComplete,
-  runResult as runResultValidator,
-  RunResult,
+  runResult as resultValidator,
   type RetryBehavior,
+  RunResult,
   OnCompleteArgs as SharedOnCompleteArgs,
   Status,
-  Config,
 } from "../component/shared.js";
-import { type LogLevel, logLevel } from "../component/logging.js";
 import { RunMutationCtx, RunQueryCtx, UseApi } from "./utils.js";
-import { DEFAULT_LOG_LEVEL } from "../component/logging.js";
-import { DEFAULT_MAX_PARALLELISM } from "../component/kick.js";
-export { runResultValidator, type RunResult };
+export { resultValidator, type RunResult };
 
 // Attempts will run with delay [0, 250, 500, 1000, 2000] (ms)
 export const DEFAULT_RETRY_BEHAVIOR: RetryBehavior = {
@@ -133,11 +136,18 @@ export class Workpool {
     fnArgs: Args,
     options?: CallbackOptions & SchedulerOptions
   ): Promise<WorkId> {
+    const onComplete: OnComplete | undefined = options?.onComplete
+      ? {
+          fnHandle: await createFunctionHandle(options.onComplete),
+          context: options.context,
+        }
+      : undefined;
     const id = await ctx.runMutation(this.component.lib.enqueue, {
       ...(await defaultEnqueueArgs(fn, this.options)),
       fnArgs,
       fnType: "mutation",
       runAt: getRunAt(options),
+      onComplete,
     });
     return id as WorkId;
   }
@@ -215,7 +225,7 @@ export type CallbackOptions = {
    *  args: {
    *    workId: workIdValidator,
    *    context: v.any(),
-   *    result: runResult,
+   *    result: resultValidator,
    *  },
    *  handler: async (ctx, args) => {
    *    console.log(args.result, "Got Context back -> ", args.context, Date.now() - args.context);

src/client/index.ts

@@ -25,19 +25,19 @@ Concepts:
 flowchart LR
     Client -->|enqueue| pendingStart
     Client -->|cancel| pendingCancelation
-    Recovery-->|recover| pendingCancelation
-    Recovery-->|recover| pendingCompletion
-    Worker-->|"saveResult"| pendingCompletion
-    pendingStart -->|main| workerRunning["internalState.running"]
-    workerRunning-->|"main(pendingCompletion)"| Retry{"Needs retry?"}
-    Retry-->|"no / canceled"| complete
-    Retry-->|yes| pendingStart
-    pendingStart-->|"main(pendingCancelation)"| complete
+    complete --> |success or failure| pendingCompletion
+    pendingCompletion -->|retry| pendingStart
+    pendingStart --> workerRunning["worker running"]
+    workerRunning -->|worker finished| complete
+    workerRunning --> |recovery| complete
+    successfulCancel["AND"]@{shape: delay} --> |canceled| complete
+    pendingStart --> successfulCancel
+    pendingCancelation --> successfulCancel
 ```
 
 Notably:
 
-- The pending\* states are only written by other sources.
+- The pending\* states are written by outside sources.
 - The main loop federates changes to/from "running"
 - Canceling only impacts pending and retrying jobs.
 
@@ -53,12 +53,12 @@ flowchart TD
     running-->|"all done"| idle
 ```
 
-- While the loop is running, clients won't see database conflicts with the
-  state changing.
-- The "saturated" state is concretely "running" or "scheduled" with a boolean
-  set, to avoid clients from kicking the main loop on enqueueing, which is
-  unlikely to be productive, since the next action needs to be something
-  terminating.
+- While the loop is running, the runStatus doesn't change, making it safer to
+  read from clients without database conflicts.
+- The "saturated" state is concretely "running" or "scheduled" at max
+  parallelism. There is a boolean set on "scheduled" to avoid clients from
+  kicking the main loop on enqueueing, which is unlikely to be productive, since
+  the next action needs to be something terminating.
 
 ## Retention optimization strategy
 

src/component/README.md

@@ -8,6 +8,7 @@
  * @module
  */
 
+import type * as complete from "../complete.js";
 import type * as kick from "../kick.js";
 import type * as lib from "../lib.js";
 import type * as logging from "../logging.js";
@@ -31,6 +32,7 @@ import type {
  * ```
  */
 declare const fullApi: ApiFromModules<{
+  complete: typeof complete;
   kick: typeof kick;
   lib: typeof lib;
   logging: typeof logging;
@@ -80,11 +82,14 @@ export type Mounts = {
       "query",
       "public",
       { id: string },
-      | { attempt: number; state: "pending" }
-      | { attempt: number; state: "running" }
+      | { previousAttempts: number; state: "pending" }
+      | { previousAttempts: number; state: "running" }
       | { state: "finished" }
     >;
   };
+  stats: {
+    queueLength: FunctionReference<"query", "public", {}, number>;
+  };
 };
 // For now fullApiWithMounts is only fullApi which provides
 // jump-to-definition in component client code.