use commitTS for workpool

CHANGELOG.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## Unreleased
+
+- Orders the pending-work queues by commit timestamp (`v.commitTs()`, Convex ≥
+  1.43). A commit timestamp is assigned when the transaction commits, so nothing
+  can appear behind a cursor the main loop has already read past. That removes
+  the 15-second cursor rewind buffer and the once-a-minute full rescan that
+  0.4.7 added to cope with out-of-order inserts — the loop now reads only rows
+  it hasn't seen. On a 5000-task saturation benchmark (parallelism 200, 20ms
+  tasks) this ran ~16% faster — 121 → 144 tasks/s, with p99 latency down from
+  ~32s to ~25s.
+- The `segment` fields keep their name and index, but now hold nanoseconds
+  rather than 100ms buckets. Work scheduled to start later stores its start time
+  there directly, which sorts after everything already committed, so one index
+  covers both ready and scheduled work. Entries an older version wrote have much
+  smaller values, so they sort first and get processed promptly on upgrade.
+- Requires `convex` 1.43 or later.
+
 ## 0.4.9
 
 - Runs actions and queries in batches of up to 32 from a single scheduled

README.md

@@ -253,6 +253,10 @@ can learn about [here](https://docs.convex.dev/get-started).
 Run `npm create convex` or follow any of the
 [quickstarts](https://docs.convex.dev/home) to set one up.
 
+Workpool orders its internal queues by
+[commit timestamp](https://docs.convex.dev/database/advanced/commit-timestamp),
+so it needs `convex` 1.43 or later.
+
 ### Install the component
 
 See [`example/`](./example/convex/) for a working demo.

example/convex/_generated/api.d.ts

@@ -24,6 +24,7 @@ import type * as test_scenarios_noisyNeighbor from "../test/scenarios/noisyNeigh
 import type * as test_scenarios_overhead from "../test/scenarios/overhead.js";
 import type * as test_scenarios_sustained from "../test/scenarios/sustained.js";
 import type * as test_scenarios_throughput from "../test/scenarios/throughput.js";
+import type * as test_scheduling from "../test/scheduling.js";
 import type * as test_work from "../test/work.js";
 
 import type {
@@ -49,6 +50,7 @@ declare const fullApi: ApiFromModules<{
   "test/scenarios/overhead": typeof test_scenarios_overhead;
   "test/scenarios/sustained": typeof test_scenarios_sustained;
   "test/scenarios/throughput": typeof test_scenarios_throughput;
+  "test/scheduling": typeof test_scheduling;
   "test/work": typeof test_work;
 }>;
 

example/convex/schema.ts

@@ -38,4 +38,11 @@ export default defineSchema({
     name: v.string(),
     value: v.number(),
   }).index("name", ["name"]),
+  // When each probe in test/scheduling.ts ran, to check delayed and retried
+  // work against a real deployment.
+  schedulingProbes: defineTable({
+    label: v.string(),
+    at: v.number(),
+    attempt: v.optional(v.number()),
+  }),
 });

example/convex/test/scheduling.ts

@@ -0,0 +1,147 @@
+import { v } from "convex/values";
+import {
+  internalAction,
+  internalMutation,
+  internalQuery,
+} from "../_generated/server";
+import { components, internal } from "../_generated/api";
+import { Workpool } from "@convex-dev/workpool";
+
+/**
+ * End-to-end checks for the two paths the throughput scenarios never touch:
+ * work scheduled for later, and work that retries. Both put a wall-clock time
+ * into the queue's commit-timestamp-ordered field instead of the placeholder,
+ * so they're worth exercising against a real deployment and not just
+ * convex-test.
+ *
+ * Run:
+ *   npx convex run test/scheduling:default
+ *   npx convex run test/scheduling:default '{"delayMs":400000}'  # past the
+ *                                          # safe-future threshold
+ */
+const pool = new Workpool(components.testWorkpool, { maxParallelism: 10 });
+
+const vProbe = v.object({
+  label: v.string(),
+  at: v.number(),
+  attempt: v.optional(v.number()),
+});
+
+export const record = internalMutation({
+  args: { label: v.string(), attempt: v.optional(v.number()) },
+  returns: v.null(),
+  handler: async (ctx, { label, attempt }) => {
+    await ctx.db.insert("schedulingProbes", { label, at: Date.now(), attempt });
+    return null;
+  },
+});
+
+export const probes = internalQuery({
+  args: {},
+  returns: v.array(vProbe),
+  handler: async (ctx) => {
+    const docs = await ctx.db.query("schedulingProbes").collect();
+    return docs
+      .map(({ label, at, attempt }) => ({ label, at, attempt }))
+      .sort((a, b) => a.at - b.at);
+  },
+});
+
+export const reset = internalMutation({
+  args: {},
+  returns: v.null(),
+  handler: async (ctx) => {
+    for (const doc of await ctx.db.query("schedulingProbes").collect()) {
+      await ctx.db.delete("schedulingProbes", doc._id);
+    }
+    return null;
+  },
+});
+
+/** Fails its first two attempts so the retry backoff path runs. */
+export const failTwice = internalAction({
+  args: { label: v.string() },
+  returns: v.null(),
+  handler: async (ctx, { label }) => {
+    const seen = (
+      await ctx.runQuery(internal.test.scheduling.probes, {})
+    ).filter((p) => p.label === label).length;
+    await ctx.runMutation(internal.test.scheduling.record, {
+      label,
+      attempt: seen + 1,
+    });
+    if (seen < 2) throw new Error(`attempt ${seen + 1} fails on purpose`);
+    return null;
+  },
+});
+
+export const enqueueBoth = internalMutation({
+  args: { delayMs: v.number() },
+  returns: v.null(),
+  handler: async (ctx, { delayMs }) => {
+    await pool.enqueueMutation(
+      ctx,
+      internal.test.scheduling.record,
+      { label: "delayed" },
+      { runAfter: delayMs },
+    );
+    await pool.enqueueAction(
+      ctx,
+      internal.test.scheduling.failTwice,
+      { label: "retry" },
+      { retry: { maxAttempts: 4, initialBackoffMs: 500, base: 2 } },
+    );
+    return null;
+  },
+});
+
+export default internalAction({
+  args: { delayMs: v.optional(v.number()) },
+  handler: async (ctx, { delayMs = 8_000 }) => {
+    await ctx.runMutation(internal.test.scheduling.reset, {});
+
+    const enqueuedAt = Date.now();
+    await ctx.runMutation(internal.test.scheduling.enqueueBoth, { delayMs });
+
+    const deadline = Date.now() + delayMs + 60_000;
+    // Keep the number of polls bounded — a long delay at a fixed short interval
+    // runs past an action's limit on how many functions it may call.
+    const pollMs = Math.max(250, Math.round(delayMs / 100));
+    let probes: { label: string; at: number; attempt?: number }[] = [];
+    let ranEarly = false;
+    while (Date.now() < deadline) {
+      probes = await ctx.runQuery(internal.test.scheduling.probes, {});
+      const delayed = probes.find((p) => p.label === "delayed");
+      if (delayed && delayed.at < enqueuedAt + delayMs) ranEarly = true;
+      const retriesDone = probes.filter((p) => p.label === "retry").length >= 3;
+      if (delayed && retriesDone) break;
+      await new Promise((r) => setTimeout(r, pollMs));
+    }
+
+    const delayed = probes.find((p) => p.label === "delayed");
+    const retries = probes.filter((p) => p.label === "retry");
+    const lateByMs = delayed ? delayed.at - (enqueuedAt + delayMs) : undefined;
+
+    console.log("=== scheduling results ===");
+    console.log(
+      delayed
+        ? `delayed (runAfter ${delayMs}ms): ran ${lateByMs}ms after its runAt` +
+            (ranEarly ? " — RAN EARLY" : "")
+        : `delayed (runAfter ${delayMs}ms): NEVER RAN`,
+    );
+    console.log(
+      `retry: ${retries.length} attempts, gaps ${
+        retries
+          .slice(1)
+          .map((p, i) => p.at - retries[i].at)
+          .join("/") || "n/a"
+      }ms`,
+    );
+    return {
+      delayedRan: !!delayed,
+      delayedRanEarly: ranEarly,
+      delayedLateByMs: lateByMs,
+      retryAttempts: retries.length,
+    };
+  },
+});