diff --git a/src/component/_generated/server.ts b/src/component/_generated/server.ts index 1a7688a..481cb3a 100644 --- a/src/component/_generated/server.ts +++ b/src/component/_generated/server.ts @@ -39,6 +39,8 @@ import type { DataModel } from "./dataModel.js"; type Env = { readonly CONVEX_CLOUD_URL: string; readonly CONVEX_SITE_URL: string; + readonly WORKER_IDLE_COOLDOWN_MS: string | undefined; + readonly WORKER_POLL_INTERVAL_MS: string | undefined; }; /** diff --git a/src/component/convex.config.ts b/src/component/convex.config.ts index 86ec81d..90127ea 100644 --- a/src/component/convex.config.ts +++ b/src/component/convex.config.ts @@ -1,7 +1,13 @@ import { defineComponent } from "convex/server"; +import { v } from "convex/values"; import batchWorker from "@convex-dev/batch-worker/convex.config.js"; -const component = defineComponent("aggregate"); +const component = defineComponent("aggregate", { + env: { + WORKER_IDLE_COOLDOWN_MS: v.optional(v.string()), + WORKER_POLL_INTERVAL_MS: v.optional(v.string()), + }, +}); component.use(batchWorker); export default component; diff --git a/src/component/worker.test.ts b/src/component/worker.test.ts index b233dec..c844139 100644 --- a/src/component/worker.test.ts +++ b/src/component/worker.test.ts @@ -32,6 +32,7 @@ import { } from "./btree.js"; import { BATCH_MAX_OPERATIONS, + parseDuration, MAX_BYTES_PER_ENTRY, enqueueOperations, hasHeadroomToFinish, @@ -104,6 +105,13 @@ async function drainViaWorker(t: TestConvex): Promise { await t.finishAllScheduledFunctions(vi.runAllTimers); } +// The batch worker's run state for our queue, or null before it is registered. +async function workerStatus(t: TestConvex) { + return await t.run(async (ctx) => + ctx.runQuery(components.batchWorker.lib.status, { name: OPS_WORKER_NAME }), + ); +} + // The cursor the batch worker has committed for our queue, or null before it // has drained anything. async function workerCursor( @@ -901,6 +909,87 @@ describe("guards", () => { }); }); +describe("staying awake between bursts", () => { + // Like `drainViaWorker`, but without running the clock past the worker's + // next poll, so its run state stays observable mid-gap. + async function advance(t: TestConvex, ms: number) { + for (let elapsed = 0; elapsed < ms; elapsed += 200) { + vi.advanceTimersByTime(200); + await t.finishInProgressScheduledFunctions(); + } + } + + afterEach(() => { + vi.unstubAllEnvs(); + }); + + const COOLDOWN_MS = 60_000; + + function configureWorker() { + vi.stubEnv("WORKER_IDLE_COOLDOWN_MS", String(COOLDOWN_MS)); + vi.stubEnv("WORKER_POLL_INTERVAL_MS", "1000"); + } + + test("an unusable env var is left unset", () => { + expect(parseDuration("5000")).toBe(5_000); + expect(parseDuration(undefined)).toBeUndefined(); + expect(parseDuration("")).toBeUndefined(); + expect(parseDuration("soon")).toBeUndefined(); + expect(parseDuration("0")).toBeUndefined(); + expect(parseDuration("-1000")).toBeUndefined(); + }); + + test("an empty queue defers to the batch worker unless configured", async () => { + const t = initConvexTest(); + expect( + await t.query(internal.worker.getBatch, { name: OPS_WORKER_NAME }), + ).toEqual({ kind: "idle" }); + + configureWorker(); + expect( + await t.query(internal.worker.getBatch, { name: OPS_WORKER_NAME }), + ).toEqual({ kind: "idle", cooldownMs: COOLDOWN_MS, pollIntervalMs: 1_000 }); + }); + + test("the worker stays running across a gap shorter than the cooldown", async () => { + configureWorker(); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await advance(t, COOLDOWN_MS / 2); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + await advance(t, 2_000); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ + k: 2, + v: "b", + s: 0, + }); + }); + }); + + test("it still parks once the queue has been empty for the cooldown", async () => { + configureWorker(); + const t = initConvexTest(); + await enqueue(t, { type: "insert", key: 1, value: "a" }); + await advance(t, COOLDOWN_MS + 2_000); + expect(await workerStatus(t)).toEqual({ kind: "idle" }); + + await enqueue(t, { type: "insert", key: 2, value: "b" }); + expect(await workerStatus(t)).toEqual({ kind: "running" }); + await drainViaWorker(t); + await t.run(async (ctx) => { + expect(await getHandler(ctx, { key: 2 })).toEqual({ + k: 2, + v: "b", + s: 0, + }); + expect(await ctx.db.query("pendingOperations").collect()).toEqual([]); + }); + }); +}); + describe("pinging the worker", () => { beforeEach(() => { vi.mocked(ping).mockClear(); diff --git a/src/component/worker.ts b/src/component/worker.ts index 604c4b5..7f2352d 100644 --- a/src/component/worker.ts +++ b/src/component/worker.ts @@ -9,6 +9,7 @@ import { import type { TransactionMetrics, WithoutSystemFields } from "convex/server"; import { type DatabaseWriter, + env, internalMutation, internalQuery, type MutationCtx, @@ -35,6 +36,13 @@ export const MAX_BYTES_PER_ENTRY = 100 * 1024; export const OPS_WORKER_NAME = "ops"; +export function parseDuration(value: string | undefined): number | undefined { + const parsed = Number(value); + return value !== undefined && Number.isFinite(parsed) && parsed > 0 + ? parsed + : undefined; +} + export async function enqueueOperations( ctx: MutationCtx, operations: Operation[], @@ -144,7 +152,11 @@ export const getBatch = internalQuery({ operations += row.operations.length; } if (entries.length === 0) { - return { kind: "idle" as const }; + return { + kind: "idle" as const, + cooldownMs: parseDuration(env.WORKER_IDLE_COOLDOWN_MS), + pollIntervalMs: parseDuration(env.WORKER_POLL_INTERVAL_MS), + }; } return { kind: "work" as const, batch: { entries } }; }, diff --git a/tsconfig.json b/tsconfig.json index d238155..be83fe9 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -6,6 +6,7 @@ "target": "ESNext", "lib": ["ES2021", "dom", "DOM.Iterable"], + "types": ["node"], "jsx": "react-jsx", "forceConsistentCasingInFileNames": true, "allowSyntheticDefaultImports": true,