remove runAtTime

Author: ldanilek

eslint.config.js

@@ -38,7 +38,7 @@ export default [
         {
           argsIgnorePattern: "^_",
           varsIgnorePattern: "^_",
-          "caughtErrorsIgnorePattern": "^_",
+          caughtErrorsIgnorePattern: "^_",
         },
       ],
     },

example/convex/_generated/api.d.ts

@@ -56,7 +56,7 @@ export declare const components: {
           fnArgs: any;
           fnHandle: string;
           fnName: string;
-          fnType: "action" | "mutation" | "unknown";
+          fnType: "action" | "mutation";
           options: {
             actionTimeoutMs?: number;
             debounceMs?: number;
@@ -66,9 +66,7 @@ export declare const components: {
             mutationTimeoutMs?: number;
             slowHeartbeatMs?: number;
             ttl?: number;
-            unknownTimeoutMs?: number;
           };
-          runAtTime: number;
         },
         string
       >;
@@ -104,7 +102,7 @@ export declare const components: {
           fnArgs: any;
           fnHandle: string;
           fnName: string;
-          fnType: "action" | "mutation" | "unknown";
+          fnType: "action" | "mutation";
           options: {
             actionTimeoutMs?: number;
             debounceMs?: number;
@@ -114,9 +112,7 @@ export declare const components: {
             mutationTimeoutMs?: number;
             slowHeartbeatMs?: number;
             ttl?: number;
-            unknownTimeoutMs?: number;
           };
-          runAtTime: number;
         },
         string
       >;

src/client/index.ts

@@ -79,7 +79,6 @@ export class WorkPool {
       fnName: getFunctionName(fn),
       fnArgs,
       fnType: "action",
-      runAtTime: Date.now(),
       options: this.options,
     });
     return id as WorkId;
@@ -95,31 +94,6 @@ export class WorkPool {
       fnName: getFunctionName(fn),
       fnArgs,
       fnType: "mutation",
-      runAtTime: Date.now(),
-      options: this.options,
-    });
-    return id as WorkId;
-  }
-  // Unknown is if you don't know at runtime whether it's an action or mutation,
-  // which can happen if it comes from `runAt` or `runAfter`.
-  async enqueueUnknown<Args extends DefaultFunctionArgs>(
-    ctx: RunMutationCtx,
-    fn: FunctionReference<
-      "action" | "mutation",
-      FunctionVisibility,
-      Args,
-      null
-    >,
-    fnArgs: Args,
-    runAtTime: number
-  ): Promise<WorkId> {
-    const fnHandle = await createFunctionHandle(fn);
-    const id = await ctx.runMutation(this.component.lib.enqueue, {
-      fnHandle,
-      fnName: getFunctionName(fn),
-      fnArgs,
-      fnType: "unknown",
-      runAtTime,
       options: this.options,
     });
     return id as WorkId;

src/component/_generated/api.d.ts

@@ -41,7 +41,7 @@ export type Mounts = {
         fnArgs: any;
         fnHandle: string;
         fnName: string;
-        fnType: "action" | "mutation" | "unknown";
+        fnType: "action" | "mutation";
         options: {
           actionTimeoutMs?: number;
           debounceMs?: number;
@@ -51,9 +51,7 @@ export type Mounts = {
           mutationTimeoutMs?: number;
           slowHeartbeatMs?: number;
           ttl?: number;
-          unknownTimeoutMs?: number;
         };
-        runAtTime: number;
       },
       string
     >;

src/component/lib.ts

@@ -25,17 +25,11 @@ export const enqueue = mutation({
     fnHandle: v.string(),
     fnName: v.string(),
     fnArgs: v.any(),
-    fnType: v.union(
-      v.literal("action"),
-      v.literal("mutation"),
-      v.literal("unknown")
-    ),
-    runAtTime: v.number(),
+    fnType: v.union(v.literal("action"), v.literal("mutation")),
     options: v.object({
       maxParallelism: v.number(),
       actionTimeoutMs: v.optional(v.number()),
       mutationTimeoutMs: v.optional(v.number()),
-      unknownTimeoutMs: v.optional(v.number()),
       debounceMs: v.optional(v.number()),
       fastHeartbeatMs: v.optional(v.number()),
       slowHeartbeatMs: v.optional(v.number()),
@@ -44,16 +38,12 @@ export const enqueue = mutation({
     }),
   },
   returns: v.id("pendingWork"),
-  handler: async (
-    ctx,
-    { fnHandle, fnName, options, fnArgs, fnType, runAtTime }
-  ) => {
+  handler: async (ctx, { fnHandle, fnName, options, fnArgs, fnType }) => {
     const debounceMs = options.debounceMs ?? 50;
     await ensurePoolExists(ctx, {
       maxParallelism: options.maxParallelism,
       actionTimeoutMs: options.actionTimeoutMs ?? 15 * 60 * 1000,
       mutationTimeoutMs: options.mutationTimeoutMs ?? 30 * 1000,
-      unknownTimeoutMs: options.unknownTimeoutMs ?? 15 * 60 * 1000,
       debounceMs,
       fastHeartbeatMs: options.fastHeartbeatMs ?? 10 * 1000,
       slowHeartbeatMs: options.slowHeartbeatMs ?? 2 * 60 * 60 * 1000,
@@ -65,10 +55,8 @@ export const enqueue = mutation({
       fnName,
       fnArgs,
       fnType,
-      runAtTime,
     });
-    const delay = Math.max(runAtTime - Date.now(), debounceMs);
-    await kickMainLoop(ctx, delay, false);
+    await kickMainLoop(ctx, debounceMs, false);
     return workId;
   },
 });
@@ -150,10 +138,7 @@ export const mainLoop = internalMutation({
       BATCH_SIZE
     );
     let didSomething = false;
-    const pending = await ctx.db
-      .query("pendingWork")
-      .withIndex("runAtTime", (q) => q.lte("runAtTime", Date.now()))
-      .take(toSchedule);
+    const pending = await ctx.db.query("pendingWork").take(toSchedule);
     console_.debug(`scheduling ${pending.length} pending work`);
     await Promise.all(
       pending.map(async (work) => {
@@ -254,12 +239,9 @@ export const mainLoop = internalMutation({
     } else {
       // Decide when to wake up.
       const allInProgressWork = await ctx.db.query("inProgressWork").collect();
-      const nextPending = await ctx.db
-        .query("pendingWork")
-        .withIndex("runAtTime")
-        .first();
+      const nextPending = await ctx.db.query("pendingWork").first();
       const nextPendingTime = nextPending
-        ? nextPending.runAtTime
+        ? nextPending._creationTime
         : slowHeartbeatMs + Date.now();
       const nextInProgress = allInProgressWork.length
         ? Math.min(
@@ -285,8 +267,8 @@ async function beginWork(
   if (!options) {
     throw new Error("cannot begin work with no pool");
   }
-  recordStarted(work._id, work.fnName, work._creationTime, work.runAtTime);
-  const { mutationTimeoutMs, actionTimeoutMs, unknownTimeoutMs } = options;
+  recordStarted(work._id, work.fnName, work._creationTime);
+  const { mutationTimeoutMs, actionTimeoutMs } = options;
   if (work.fnType === "action") {
     return {
       scheduledId: await ctx.scheduler.runAfter(
@@ -313,12 +295,6 @@ async function beginWork(
       ),
       timeoutMs: mutationTimeoutMs,
     };
-  } else if (work.fnType === "unknown") {
-    const fnHandle = work.fnHandle as FunctionHandle<"action" | "mutation">;
-    return {
-      scheduledId: await ctx.scheduler.runAfter(0, fnHandle, work.fnArgs),
-      timeoutMs: unknownTimeoutMs,
-    };
   } else {
     throw new Error(`Unexpected fnType ${work.fnType}`);
   }

src/component/schema.ts

@@ -48,7 +48,6 @@ export default defineSchema({
     maxParallelism: v.number(),
     actionTimeoutMs: v.number(),
     mutationTimeoutMs: v.number(),
-    unknownTimeoutMs: v.number(),
     debounceMs: v.number(),
     fastHeartbeatMs: v.number(),
     slowHeartbeatMs: v.number(),
@@ -69,16 +68,11 @@ export default defineSchema({
   }).index("runAtTime", ["runAtTime"]),
 
   pendingWork: defineTable({
-    fnType: v.union(
-      v.literal("action"),
-      v.literal("mutation"),
-      v.literal("unknown")
-    ),
+    fnType: v.union(v.literal("action"), v.literal("mutation")),
     fnHandle: v.string(),
     fnName: v.string(),
     fnArgs: v.any(),
-    runAtTime: v.number(),
-  }).index("runAtTime", ["runAtTime"]),
+  }),
   pendingCompletion: defineTable({
     completionStatus,
     workId: v.id("pendingWork"),

src/component/stats.ts

@@ -22,16 +22,14 @@ workpool
 export function recordStarted(
   workId: Id<"pendingWork">,
   fnName: string,
-  enqueuedAt: number,
-  runAtTime: number
+  enqueuedAt: number
 ) {
   console.log(
     JSON.stringify({
       workId,
       event: "started",
       fnName,
       enqueuedAt,
-      runAtTime,
       startedAt: Date.now(),
       lagSinceEnqueued: Date.now() - enqueuedAt,
     })