Merge pull request #145 from danthegoodman1/main

Support maxParallelism of 0 to enable pausing workpools

Author: ianmacartney

src/client/index.ts

@@ -256,6 +256,22 @@ export class Workpool {
       ...options,
     });
   }
+
+  /**
+   * Kicks the workpool to wake it up and sync config (e.g. maxParallelism).
+   * Useful for resuming after pausing (setting maxParallelism to 0).
+   *
+   * @param ctx - The mutation or action context that can call ctx.runMutation.
+   */
+  async kick(ctx: RunMutationCtx): Promise<void> {
+    await ctx.runMutation(this.component.lib.kick, {
+      config: {
+        logLevel: this.options.logLevel ?? DEFAULT_LOG_LEVEL,
+        maxParallelism: this.options.maxParallelism ?? DEFAULT_MAX_PARALLELISM,
+      },
+    });
+  }
+
   /**
    * Gets the status of a work item.
    *
@@ -360,7 +376,7 @@ export type RetryOption = {
 
 export type WorkpoolOptions = {
   /** How many actions/mutations can be running at once within this pool.
-   * Min 1, Suggested max: 100 on Pro, 20 on the free plan.
+   * Min 0 (no new work starts), Suggested max: 100 on Pro, 20 on free plan.
    */
   maxParallelism?: number;
   /** How much to log. This is updated on each call to `enqueue*`,

src/component/_generated/component.ts

@@ -93,6 +93,18 @@ export type ComponentApi<Name extends string | undefined = string | undefined> =
         Array<string>,
         Name
       >;
+      kick: FunctionReference<
+        "mutation",
+        "internal",
+        {
+          config: {
+            logLevel: "DEBUG" | "TRACE" | "INFO" | "REPORT" | "WARN" | "ERROR";
+            maxParallelism: number;
+          };
+        },
+        null,
+        Name
+      >;
       status: FunctionReference<
         "query",
         "internal",

src/component/kick.ts

@@ -110,7 +110,7 @@ async function getOrUpdateGlobals(ctx: MutationCtx, config?: Partial<Config>) {
   } else if (config) {
     let updated = false;
     if (
-      config.maxParallelism &&
+      config.maxParallelism !== undefined &&
       config.maxParallelism !== globals.maxParallelism
     ) {
       globals.maxParallelism = config.maxParallelism;

src/component/lib.test.ts

@@ -78,11 +78,11 @@ describe("lib", () => {
           fnType: "mutation",
           runAt: Date.now(),
           config: {
-            maxParallelism: 0, // Less than minimum
+            maxParallelism: -1, // Less than minimum
             logLevel: "WARN",
           },
         }),
-      ).rejects.toThrow("maxParallelism must be >= 1");
+      ).rejects.toThrow("maxParallelism must be >= 0");
     });
   });
 

src/component/lib.ts

@@ -81,8 +81,8 @@ function validateConfig(config: Config) {
     createLogger(config.logLevel).warn(
       `maxParallelism should be <= ${MAX_PARALLELISM_SOFT_LIMIT}, but is set to ${config.maxParallelism}. This will be an error in a future version.`,
     );
-  } else if (config.maxParallelism < 1) {
-    throw new Error("maxParallelism must be >= 1");
+  } else if (config.maxParallelism < 0) {
+    throw new Error("maxParallelism must be >= 0");
   }
 }
 
@@ -163,6 +163,16 @@ export const cancelAll = mutation({
   },
 });
 
+export const kick = mutation({
+  args: { config },
+  returns: v.null(),
+  handler: async (ctx, { config }) => {
+    validateConfig(config);
+    await kickMainLoop(ctx, "kick", config);
+    return null;
+  },
+});
+
 export const status = query({
   args: { id: v.id("work") },
   returns: statusValidator,