diff --git a/CHANGELOG.md b/CHANGELOG.md index 41abe3a3..c1dab062 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## 0.4.15 (Unreleased) + +- Upgrades @convex-dev/batch-worker to 0.3. Requires Convex 1.43 or later. +- Records each pendingStart ID on its work document, so the next minor version + can report status accurately for work already in flight. **Not backwards + compatible due to this schema change.** +- Adds a one-off migration that records the queue entry on work enqueued before + this version and removes queue entries whose work is gone. Before upgrading to + 0.5, deploy this version and run + `npx convex run --component workpool migrations:backfillPendingStartIds`, + replacing `workpool` with each component instance's name. + ## 0.4.14 - Updates the /test entrypoint for compatibility with convex-test's new diff --git a/package-lock.json b/package-lock.json index 3a92c33e..5023ed28 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "0.4.14", "license": "Apache-2.0", "dependencies": { - "@convex-dev/batch-worker": "^0.2.2" + "@convex-dev/batch-worker": "^0.3.6" }, "devDependencies": { "@convex-dev/eslint-plugin": "^5.0.0", @@ -42,7 +42,7 @@ "vitest": "5.0.0" }, "peerDependencies": { - "convex": "^1.36.1", + "convex": "^1.43.0", "convex-helpers": "^0.1.94" } }, @@ -378,13 +378,15 @@ } }, "node_modules/@convex-dev/batch-worker": { - "version": "0.2.2", - "resolved": "https://registry.npmjs.org/@convex-dev/batch-worker/-/batch-worker-0.2.2.tgz", - "integrity": "sha512-3knigElzTVhWw1sN28EYkGBTdrJZ03vMj2PJPuFq3V0Tg41i7tcv/EWDBFQWcUaKzBWWVw693hcJtxCF3CRtEw==", + "version": "0.3.6", + "resolved": "https://registry.npmjs.org/@convex-dev/batch-worker/-/batch-worker-0.3.6.tgz", + "integrity": "sha512-tR+9DaHR3jG3oqkIFX0ok/6944h7UHEQgT6HTmiuxU+lERYuYMsxA2IWwFD5yTFbCpO3YJhqXU4QdGY90sgUqQ==", "license": "Apache-2.0", + "dependencies": { + "convex-helpers": "^0.1.127" + }, "peerDependencies": { - "convex": "^1.36.1", - "react": "^18.3.1 || ^19.0.0" + "convex": "^1.43.0" } }, "node_modules/@convex-dev/eslint-plugin": { @@ -2066,7 +2068,7 @@ "version": "1.1.0", "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz", "integrity": "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==", - "dev": true, + "devOptional": true, "license": "MIT" }, "node_modules/@standard-schema/utils": { @@ -3074,7 +3076,6 @@ "version": "0.1.128", "resolved": "https://registry.npmjs.org/convex-helpers/-/convex-helpers-0.1.128.tgz", "integrity": "sha512-5n7uKsJrYnnybvuSAWgDQ9d5Avqk4CzFsJ22SXow4tHmVxSUH6qiJCeomWWxcLSfN9t9jpKIXNJiGJrjddS4ng==", - "dev": true, "license": "Apache-2.0", "bin": { "convex-helpers": "bin.cjs" @@ -4561,6 +4562,7 @@ "version": "19.2.4", "resolved": "https://registry.npmjs.org/react/-/react-19.2.4.tgz", "integrity": "sha512-9nfp2hYpCwOjAN+8TZFGhtWEwgvWHXqESH8qT89AT/lWklpLON22Lc8pEtnpsZz7VmawabSU0gCjnj8aC0euHQ==", + "devOptional": true, "license": "MIT", "engines": { "node": ">=0.10.0" @@ -5051,7 +5053,7 @@ "version": "5.9.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", - "dev": true, + "devOptional": true, "license": "Apache-2.0", "bin": { "tsc": "bin/tsc", @@ -5655,7 +5657,7 @@ "version": "4.3.6", "resolved": "https://registry.npmjs.org/zod/-/zod-4.3.6.tgz", "integrity": "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==", - "dev": true, + "devOptional": true, "license": "MIT", "funding": { "url": "https://github.com/sponsors/colinhacks" diff --git a/package.json b/package.json index 7d93645e..66b8ecb0 100644 --- a/package.json +++ b/package.json @@ -72,7 +72,7 @@ } }, "peerDependencies": { - "convex": "^1.36.1", + "convex": "^1.43.0", "convex-helpers": "^0.1.94" }, "devDependencies": { @@ -108,6 +108,6 @@ "types": "./dist/client/index.d.ts", "module": "./dist/client/index.js", "dependencies": { - "@convex-dev/batch-worker": "^0.2.2" + "@convex-dev/batch-worker": "^0.3.6" } } diff --git a/src/component/_generated/api.ts b/src/component/_generated/api.ts index 6bb2fa2c..684662b0 100644 --- a/src/component/_generated/api.ts +++ b/src/component/_generated/api.ts @@ -16,6 +16,7 @@ import type * as kick from "../kick.js"; import type * as lib from "../lib.js"; import type * as logging from "../logging.js"; import type * as loop from "../loop.js"; +import type * as migrations from "../migrations.js"; import type * as recovery from "../recovery.js"; import type * as shared from "../shared.js"; import type * as stats from "../stats.js"; @@ -37,6 +38,7 @@ const fullApi: ApiFromModules<{ lib: typeof lib; logging: typeof logging; loop: typeof loop; + migrations: typeof migrations; recovery: typeof recovery; shared: typeof shared; stats: typeof stats; diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index bf65459d..5c7cae32 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -129,6 +129,24 @@ describe("lib", () => { expect(status).toEqual({ state: "pending", previousAttempts: 0 }); }); + it("records the queue entry on the work", async () => { + const id = await t.mutation(api.lib.enqueue, { + fnHandle: "testHandle", + fnName: "testFunction", + fnArgs: {}, + fnType: "mutation", + runAt: Date.now() + 60_000, + config: {}, + }); + await t.run(async (ctx) => { + const pending = await ctx.db.query("pendingStart").unique(); + assert(pending); + expect((await ctx.db.get("work", id))?.pendingStartId).toBe( + pending._id, + ); + }); + }); + it("should throw error if maxParallelism is too high", async () => { await expect( t.mutation(api.lib.enqueue, { diff --git a/src/component/lib.ts b/src/component/lib.ts index a6005b63..ab05c1a0 100644 --- a/src/component/lib.ts +++ b/src/component/lib.ts @@ -112,10 +112,11 @@ async function enqueueHandler( // Store the work item const workId = await ctx.db.insert("work", workItem); - await ctx.db.insert("pendingStart", { + const pendingStartId = await ctx.db.insert("pendingStart", { workId, segment: max(toSegment(runAt), getCurrentSegment()), }); + await ctx.db.patch("work", workId, { pendingStartId }); recordEnqueued(console, { workId, fnName: workArgs.fnName, runAt }); return workId; } diff --git a/src/component/loop.test.ts b/src/component/loop.test.ts index 1f3bff4a..67d0d539 100644 --- a/src/component/loop.test.ts +++ b/src/component/loop.test.ts @@ -188,6 +188,17 @@ describe("loop", () => { expect(await statusOf(workId)).toMatchObject({ state: "running" }); }); + it("records the queue entry on older work when starting it", async () => { + await initialize(); + const workId = await enqueueWork(); + const [pending] = (await observe()).pendingStart; + + await runLoop(); + + const work = await t.run((ctx) => ctx.db.get("work", workId)); + expect(work?.pendingStartId).toBe(pending._id); + }); + it("removes work from running once a successful completion is processed", async () => { await initialize(); const workId = await enqueueWork(); @@ -391,6 +402,8 @@ describe("loop", () => { const o = await observe(); expect(o.pendingStart).toHaveLength(1); expect(o.pendingStart[0].workId).toBe(workId); + const work = await t.run((ctx) => ctx.db.get("work", workId)); + expect(work?.pendingStartId).toBe(o.pendingStart[0]._id); expect(await statusOf(workId)).toMatchObject({ state: "pending", previousAttempts: 1, diff --git a/src/component/loop.ts b/src/component/loop.ts index e446b295..45511c19 100644 --- a/src/component/loop.ts +++ b/src/component/loop.ts @@ -1,6 +1,9 @@ import type { WithoutSystemFields } from "convex/server"; import { type Infer, v } from "convex/values"; -import { type BatchResult, vBatchResult } from "@convex-dev/batch-worker"; +import { + type BatchResult, + defineBatchWorkerValidators, +} from "@convex-dev/batch-worker"; import { internal } from "./_generated/api.js"; import { kickMainLoop } from "./kick.js"; import type { Doc, Id } from "./_generated/dataModel.js"; @@ -115,6 +118,7 @@ const batchFields = { starts: v.array(vStart), }; type Batch = Infer>>; +const batchValidators = defineBatchWorkerValidators({ batch: batchFields }); /** * The work query (batch-worker contract). Decides whether there's work to do, @@ -126,8 +130,8 @@ type Batch = Infer>>; * a real dependency before going idle, so we just read the tables directly. */ export const getBatch = internalQuery({ - args: { name: v.string() }, - returns: vBatchResult(v.object(batchFields)), + args: batchValidators.vQueryArgs, + returns: batchValidators.vQueryReturns, handler: async (ctx): Promise> => { const globals = await getGlobals(ctx); const state = await ctx.db.query("internalState").unique(); @@ -232,7 +236,7 @@ export const getBatch = internalQuery({ * Returning `null` tells batch-worker to re-run immediately to keep draining. */ export const run = internalMutation({ - args: batchFields, + args: batchValidators.vMutationArgs, returns: v.null(), handler: async (ctx, batch) => { const state = await getOrCreateState(ctx); @@ -562,6 +566,9 @@ async function handleStart( console.error(`Trying to start, but work not found: ${workId}`); return null; } + if (work.pendingStartId === undefined) { + await ctx.db.patch("work", workId, { pendingStartId: _id }); + } return { work, lagMs: Date.now() - fromSegment(segment), @@ -698,10 +705,11 @@ async function rescheduleJob( const nextAttempt = withJitter(backoffMs); const startTime = Date.now() + nextAttempt; const segment = toSegment(startTime); - await ctx.db.insert("pendingStart", { + const pendingStartId = await ctx.db.insert("pendingStart", { workId: work._id, segment, }); + await ctx.db.patch("work", work._id, { pendingStartId }); return true; } diff --git a/src/component/migrations.test.ts b/src/component/migrations.test.ts new file mode 100644 index 00000000..79381491 --- /dev/null +++ b/src/component/migrations.test.ts @@ -0,0 +1,94 @@ +import { convexTest } from "convex-test"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import batchWorker from "@convex-dev/batch-worker/test"; +import { internal } from "./_generated/api.js"; +import type { Id } from "./_generated/dataModel.js"; +import schema from "./schema.js"; +import { getCurrentSegment } from "./shared.js"; + +const modules = import.meta.glob("./**/*.ts"); + +describe("backfillPendingStartIds", () => { + let t: ReturnType; + + beforeEach(() => { + vi.useFakeTimers(); + t = convexTest(schema, modules); + batchWorker.register(t); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + /** Insert work the way versions before the pointer did. */ + async function insertLegacy(count = 1) { + return t.run(async (ctx) => { + const inserted = []; + for (let i = 0; i < count; i++) { + const workId = await ctx.db.insert("work", { + fnType: "action", + fnHandle: "handle", + fnName: "fn", + fnArgs: {}, + attempts: 0, + }); + const pendingId = await ctx.db.insert("pendingStart", { + workId, + segment: getCurrentSegment() + BigInt(i % 3) - 1n, + }); + inserted.push({ workId, pendingId }); + } + return inserted; + }); + } + + async function pointerOf(workId: Id<"work">) { + return t.run( + async (ctx) => (await ctx.db.get("work", workId))?.pendingStartId ?? null, + ); + } + + async function migrate(before?: number) { + await t.mutation(internal.migrations.backfillPendingStartIds, { before }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + } + + it("records entries created before the run across pages", async () => { + const legacy = await insertLegacy(600); + // Under frozen time, convex-test spaces creation times 1µs apart. + vi.advanceTimersByTime(1000); + const before = Date.now(); + vi.advanceTimersByTime(1000); + const [later] = await insertLegacy(); + + await migrate(before); + + for (const { workId, pendingId } of legacy) { + expect(await pointerOf(workId)).toBe(pendingId); + } + expect(await pointerOf(later.workId)).toBeNull(); + }); + + it("keeps existing pointers and deletes entries whose work is gone", async () => { + const [recorded, orphan] = await insertLegacy(2); + // Visited first, but the work already points at its own entry. + await t.run(async (ctx) => { + await ctx.db.insert("pendingStart", { + workId: recorded.workId, + segment: getCurrentSegment(), + }); + await ctx.db.patch("work", recorded.workId, { + pendingStartId: recorded.pendingId, + }); + await ctx.db.delete("work", orphan.workId); + }); + + await migrate(); + + expect(await pointerOf(recorded.workId)).toBe(recorded.pendingId); + expect( + await t.run((ctx) => ctx.db.get("pendingStart", orphan.pendingId)), + ).toBeNull(); + }); +}); diff --git a/src/component/migrations.ts b/src/component/migrations.ts new file mode 100644 index 00000000..0be665fb --- /dev/null +++ b/src/component/migrations.ts @@ -0,0 +1,66 @@ +import { v } from "convex/values"; +import { internal } from "./_generated/api.js"; +import { internalMutation } from "./_generated/server.js"; + +const BATCH_SIZE = 64; + +/** + * One-off backfill of `work.pendingStartId` for work enqueued before it was + * recorded, deleting entries whose work is gone. Walks entries created before + * it started, newest first. Run it with + * `npx convex run --component workpool migrations:backfillPendingStartIds`. + */ +export const backfillPendingStartIds = internalMutation({ + args: { + // Fixed for the whole run so the paginated query, and its cursor, match. + before: v.optional(v.number()), + cursor: v.optional(v.string()), + }, + returns: v.null(), + handler: async (ctx, args): Promise => { + const before = args.before; + if (before === undefined) { + // Immediately pin the before value, so on OCCs we don't keep moving + // the target start. + // Eventually, old transactions will settle and we'll only be looking at + // older docs. + await ctx.scheduler.runAfter( + 0, + internal.migrations.backfillPendingStartIds, + { before: Date.now() }, + ); + return; + } + const page = await ctx.db + .query("pendingStart") + .withIndex("by_creation_time", (q) => q.lte("_creationTime", before)) + .order("desc") + .paginate({ cursor: args.cursor ?? null, numItems: BATCH_SIZE }); + const results = await Promise.all( + page.page.map(async (entry) => { + const work = await ctx.db.get("work", entry.workId); + if (!work) { + await ctx.db.delete("pendingStart", entry._id); + return "deleted"; + } + if (work.pendingStartId !== undefined) return "skipped"; + await ctx.db.patch("work", work._id, { pendingStartId: entry._id }); + return "updated"; + }), + ); + const updated = results.filter((r) => r === "updated").length; + const deleted = results.filter((r) => r === "deleted").length; + if (page.isDone) { + console.log( + `Done backfilling. Last batch: ${updated} updated, ${deleted} deleted.`, + ); + } else { + console.log(`Continuing after ${updated} updated, ${deleted} deleted.`); + await ctx.scheduler.runAfter( + 0, + internal.migrations.backfillPendingStartIds, + { before, cursor: page.continueCursor }, + ); + } + }, +}); diff --git a/src/component/schema.ts b/src/component/schema.ts index bf2d16c9..8a972170 100644 --- a/src/component/schema.ts +++ b/src/component/schema.ts @@ -57,6 +57,8 @@ export default defineSchema({ onComplete: v.optional(vOnCompleteFnContext), retryBehavior: v.optional(retryBehavior), canceled: v.optional(v.boolean()), + // May point to a deleted entry after starting, or be absent on older work. + pendingStartId: v.optional(v.id("pendingStart")), }), // Written on enqueue & rescheduled for retry, read & deleted by `main`.