diff --git a/README.md b/README.md index 05fec59e..2cfdc670 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,28 @@ export const emailSent = pool.defineOnComplete({ }); ``` +To skip some outcomes, add `onCompleteExcludeKinds`. For example, the same +handler can handle failures and cancellations without being invoked on success: + +```ts +await pool.enqueueAction(ctx, internal.email.send, args, { + onComplete: internal.email.emailSent, + onCompleteExcludeKinds: ["success"], + context: { emailType: args.emailType, userId: args.userId }, +}); +``` + +The options are `"success"`, `"failed"`, and `"canceled"`. + +Filtering applies to the final result. Failed attempts that will be retried do +not invoke the callback. A terminal failure, including a `NonRetryableError`, +invokes it unless `"failed"` is excluded. Cancelation through the workpool +yields `"canceled"` when it lands before the work starts or before a retry is +scheduled; an attempt already in progress runs to its own outcome, so a race +with `cancel` can still report `"success"` or a terminal `"failed"`. Direct +scheduler cancelation (for example, from the dashboard) is treated as a failure +and can trigger retries instead. + ### Idempotency Idempotent actions are actions that can be run multiple times safely. This @@ -310,6 +332,9 @@ options include: set to `true`, it will use the `defaultRetryBehavior`. If it's set to a custom config, it will use that (and do retries). - `onComplete`: A mutation to run after the function finishes. +- `onCompleteExcludeKinds`: Optional list of result kinds to skip: `"success"`, + `"failed"`, and `"canceled"`. Defaults to excluding none, so every outcome + invokes the callback. - `context`: Any data you want to pass to the `onComplete` mutation. - `runAt` and `runAfter`: Similar to `ctx.scheduler.run*`, allows you to schedule the work to run later. By default it's immediate. @@ -331,8 +356,8 @@ You can override the retry behavior per-call with the `retry` option. If an action has retries enabled but hits a terminal failure, throw a `NonRetryableError`. Workpool will treat the attempt as failed, call -`onComplete` with the normal `{ kind: "failed", error }` result, and skip any -remaining retries. +`onComplete` with the normal `{ kind: "failed", error }` result unless +`"failed"` is excluded, and skip any remaining retries. ```ts import { NonRetryableError } from "@convex-dev/workpool"; @@ -407,7 +432,12 @@ export const cancelWork = mutation({ }); ``` -This will avoid starting or retrying, but will not stop in-progress work. +This will avoid starting or retrying, but will not stop in-progress work. If an +in-progress attempt succeeds, its result is still success. Failures remain +failures when retries are disabled or exhausted, or the error is a +`NonRetryableError`. If cancellation prevents work from starting or being +retried, the final result is `canceled`. The callback runs unless that final +result is listed in `onCompleteExcludeKinds`. ## Monitoring the workpool diff --git a/src/client/index.test.ts b/src/client/index.test.ts index c6bda3c3..e64d80f1 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -1,18 +1,770 @@ import { + anyApi, + componentsGeneric, + defineSchema, + defineTable, + internalActionGeneric, + internalMutationGeneric, + internalQueryGeneric, makeFunctionReference, + type ApiFromModules, + type DataModelFromSchemaDefinition, + type FunctionArgs, type GenericDataModel, type GenericMutationCtx, } from "convex/server"; -import { type Infer, v } from "convex/values"; -import { expectTypeOf, test } from "vitest"; +import { type Infer, type ObjectType, v } from "convex/values"; +import { convexTest } from "convex-test"; +import { + afterEach, + beforeEach, + describe, + expect, + expectTypeOf, + test, + vi, +} from "vitest"; +import workpool from "../test.js"; +import type { api } from "../component/_generated/api.js"; import { type EnqueueOptions, + NonRetryableError, type OnCompleteArgs, + type RetryOption, + Workpool, type WorkId, - type Workpool, + type WorkpoolComponent, vOnCompleteArgs, + vResult, + vWorkId, } from "./index.js"; +const schema = defineSchema({ + events: defineTable({ + key: v.string(), + kind: v.union( + v.literal("attempt"), + v.literal("work"), + v.literal("callback"), + ), + workId: v.optional(vWorkId), + result: v.optional(vResult), + context: v.optional(v.any()), + }).index("key", ["key"]), +}); +const runArgs = { + key: v.string(), + fail: v.optional(v.boolean()), + failures: v.optional(v.number()), + nonRetryable: v.optional(v.boolean()), + payload: v.optional(v.string()), +}; +type RunArgs = ObjectType; +type Context = { key: string; padding?: string; throw?: boolean }; + +// Observe real callback execution while retaining Convex validation and rollback. +const callback = vi.fn( + async ( + ctx: GenericMutationCtx>, + args: OnCompleteArgs & { workId: WorkId }, + ) => { + await ctx.db.insert("events", { + key: args.context?.key ?? "default", + kind: "callback", + workId: args.workId, + context: args.context, + result: args.result, + }); + if (args.context?.throw) { + throw new Error("callback failed"); + } + return null; + }, +); +let actionGate: Promise | undefined; +const fixtures = { + attempt: internalMutationGeneric({ + args: { key: v.string() }, + returns: v.number(), + handler: async (ctx, { key }) => { + await ctx.db.insert("events", { key, kind: "attempt" }); + const events = await ctx.db + .query("events") + .withIndex("key", (q) => q.eq("key", key)) + .collect(); + return events.filter((e) => e.kind === "attempt").length; + }, + }), + action: internalActionGeneric({ + args: runArgs, + returns: v.string(), + handler: async (ctx, args): Promise => { + const attempt = await ctx.runMutation(refs.attempt, { key: args.key }); + await actionGate; + if (args.fail || attempt <= (args.failures ?? 0)) { + if (args.nonRetryable) throw new NonRetryableError("work failed"); + throw new Error("work failed"); + } + return "action result"; + }, + }), + mutation: internalMutationGeneric({ + args: runArgs, + returns: v.string(), + handler: async (ctx, args) => { + await ctx.db.insert("events", { key: args.key, kind: "work" }); + if (args.fail) throw new Error("work failed"); + return "mutation result"; + }, + }), + query: internalQueryGeneric({ + args: runArgs, + returns: v.string(), + handler: async (_ctx, args) => { + if (args.fail) throw new Error("work failed"); + return "query result"; + }, + }), + complete: internalMutationGeneric({ + args: vOnCompleteArgs( + v.object({ + key: v.string(), + padding: v.optional(v.string()), + throw: v.optional(v.boolean()), + }), + ), + returns: v.null(), + handler: callback, + }), + optionalComplete: internalMutationGeneric({ + args: vOnCompleteArgs(), + returns: v.null(), + handler: callback, + }), +}; +const refs = ( + anyApi as unknown as ApiFromModules<{ callbacks: typeof fixtures }> +).callbacks; +// An application with real callback functions and a nested workpool component. +// The generated path provides the module root expected by convex-test. +const modules = { + "./_generated/server.ts": async () => ({}), + "./callbacks.ts": async () => fixtures, +}; +const component = componentsGeneric().workpool as unknown as WorkpoolComponent; +const pool = new Workpool(component, { maxParallelism: 3, logLevel: "ERROR" }); +const retries = { maxAttempts: 3, initialBackoffMs: 1, base: 2 }; +type Kind = "action" | "mutation" | "query"; +type Mode = "all" | "failed" | "canceled"; +const ALL_KINDS = ["success", "failed", "canceled"] as const; +const excludeFilters = [ + undefined, + [], + ["success"], + ["failed"], + ["canceled"], + ["success", "failed"], + ["success", "canceled"], + ["failed", "canceled"], + ["success", "failed", "canceled"], + ["failed", "failed"], +] as const; + +describe("completion callbacks through the client and scheduler", () => { + let t: ReturnType; + function setup() { + const t = convexTest(schema, modules); + workpool.register(t); + return t; + } + beforeEach(() => { + vi.useFakeTimers(); + callback.mockClear(); + actionGate = undefined; + t = setup(); + }); + afterEach(async () => { + await t.finishAllScheduledFunctions(vi.runAllTimers); + vi.useRealTimers(); + }); + const drain = () => t.finishAllScheduledFunctions(vi.runAllTimers); + const events = (key: string) => + t.query((ctx) => + ctx.db + .query("events") + .withIndex("key", (q) => q.eq("key", key)) + .collect(), + ); + + describe.each(excludeFilters.map((excludeKinds) => ({ excludeKinds })))( + "exclude filter $excludeKinds", + ({ excludeKinds }) => { + test.each(["success", "failed", "canceled"] as const)( + "dispatches a %s result at most once and finishes the job", + async (result) => { + const id = await t.mutation((ctx) => + pool.enqueueMutation( + ctx, + refs.mutation, + { key: "filtered", fail: result === "failed" }, + { + onComplete: refs.complete, + onCompleteExcludeKinds: excludeKinds, + context: { key: "filtered" }, + runAt: result === "canceled" ? Date.now() + 60_000 : undefined, + }, + ), + ); + if (result === "canceled") + await t.mutation((ctx) => pool.cancel(ctx, id)); + await drain(); + const selected = !( + excludeKinds as readonly string[] | undefined + )?.includes(result); + expect(callback).toHaveBeenCalledTimes(selected ? 1 : 0); + const recorded = (await events("filtered")).filter( + (e) => e.kind === "callback", + ); + expect(recorded).toMatchObject( + selected ? [{ workId: id, result: { kind: result } }] : [], + ); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }, + ); + }, + ); + + function enqueue( + kind: Kind, + args: RunArgs, + options: { + mode?: Mode; + context?: Context; + retry?: RetryOption["retry"]; + runAt?: number; + } = {}, + ) { + const { mode = "failed", context = { key: args.key }, ...rest } = options; + const opts = { + onComplete: refs.complete, + onCompleteExcludeKinds: + mode === "all" ? undefined : ALL_KINDS.filter((k) => k !== mode), + context, + ...rest, + }; + return t.mutation((ctx) => { + switch (kind) { + case "action": + return pool.enqueueAction(ctx, refs.action, args, opts); + case "mutation": + return pool.enqueueMutation(ctx, refs.mutation, args, opts); + case "query": + return pool.enqueueQuery(ctx, refs.query, args, opts); + } + }); + } + + test.each(["action", "mutation", "query"] as const)( + "a failure filter skips successful %s work", + async (kind) => { + const id = await enqueue(kind, { key: kind }); + await drain(); + expect(callback).not.toHaveBeenCalled(); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }, + ); + + test.each(["action", "mutation", "query"] as const)( + "a failure filter reuses an existing completion handler for failed %s work", + async (kind) => { + const id = await enqueue(kind, { key: kind, fail: true }); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect( + (await events(kind)).filter((e) => e.kind === "callback"), + ).toMatchObject([ + { + workId: id, + context: { key: kind }, + result: { kind: "failed", error: "work failed" }, + }, + ]); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + if (kind === "mutation") { + expect((await events(kind)).some((e) => e.kind === "work")).toBe(false); + } + }, + ); + + test("a failure filter runs once after retry exhaustion", async () => { + await enqueue("action", { key: "retry", fail: true }, { retry: retries }); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect((await events("retry")).map((e) => e.kind)).toEqual([ + "attempt", + "attempt", + "attempt", + "callback", + ]); + }); + + test("a failure filter is skipped if a retry succeeds", async () => { + await enqueue("action", { key: "retry", failures: 2 }, { retry: retries }); + await drain(); + expect(callback).not.toHaveBeenCalled(); + expect((await events("retry")).map((e) => e.kind)).toEqual([ + "attempt", + "attempt", + "attempt", + ]); + }); + + test("NonRetryableError skips remaining attempts and invokes the failure callback", async () => { + await enqueue( + "action", + { key: "terminal", fail: true, nonRetryable: true }, + { retry: retries }, + ); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect((await events("terminal")).map((e) => e.kind)).toEqual([ + "attempt", + "callback", + ]); + expect((await events("terminal"))[1].result).toEqual({ + kind: "failed", + error: "work failed", + }); + }); + + test.each(["all", "failed", "canceled"] as const)( + "%s preserves cancellation semantics", + async (mode) => { + const id = await enqueue( + "action", + { key: "cancel", fail: true }, + { mode, runAt: Date.now() + 60_000 }, + ); + await t.mutation((ctx) => pool.cancel(ctx, id)); + await drain(); + const recorded = await events("cancel"); + if (mode === "failed") { + expect(callback).not.toHaveBeenCalled(); + expect(recorded).toEqual([]); + } else { + expect(callback).toHaveBeenCalledTimes(1); + expect(recorded).toMatchObject([ + { workId: id, kind: "callback", result: { kind: "canceled" } }, + ]); + } + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }, + ); + + test.each(["action", "mutation", "query"] as const)( + "a cancellation filter skips successful and failed %s work", + async (kind) => { + const ids = [ + await enqueue(kind, { key: "success" }, { mode: "canceled" }), + await enqueue( + kind, + { key: "failure", fail: true }, + { mode: "canceled", retry: retries }, + ), + ]; + await drain(); + expect(callback).not.toHaveBeenCalled(); + expect(await t.query((ctx) => pool.statusBatch(ctx, ids))).toEqual([ + { state: "finished" }, + { state: "finished" }, + ]); + if (kind === "action") { + expect(await events("failure")).toHaveLength(3); + } + }, + ); + + test("a cancellation filter runs once when a job is canceled during retry backoff", async () => { + const id = await enqueue( + "action", + { key: "backoff", fail: true }, + { + mode: "canceled", + retry: { ...retries, initialBackoffMs: 60_000 }, + }, + ); + await vi.waitFor( + async () => { + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "pending", + previousAttempts: 1, + }); + }, + { timeout: 10_000 }, + ); + expect(callback).not.toHaveBeenCalled(); + await t.mutation((ctx) => pool.cancel(ctx, id)); + await t.mutation((ctx) => pool.cancel(ctx, id)); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect((await events("backoff")).map((e) => e.kind)).toEqual([ + "attempt", + "callback", + ]); + expect((await events("backoff"))[1]).toMatchObject({ + workId: id, + result: { kind: "canceled" }, + }); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }); + + test.each([ + { name: "success", fail: false, retry: false, expected: "success" }, + { name: "failure", fail: true, retry: false, expected: "failed" }, + { + name: "prevented retry", + fail: true, + retry: retries, + expected: "canceled", + }, + { + name: "non-retryable failure", + fail: true, + nonRetryable: true, + retry: retries, + expected: "failed", + }, + { + name: "exhausted retries", + fail: true, + retry: { ...retries, maxAttempts: 1 }, + expected: "failed", + }, + ] as const)( + "canceling running work dispatches by its final result ($name)", + async ({ name, fail, retry, expected, ...args }) => { + let release!: () => void; + actionGate = new Promise((resolve) => { + release = resolve; + }); + try { + const id = await t.mutation((ctx) => + pool.enqueueAction( + ctx, + refs.action, + { key: name, fail, ...args }, + { + onComplete: refs.complete, + onCompleteExcludeKinds: ["success"], + context: { key: name }, + retry, + }, + ), + ); + await vi.waitFor( + async () => { + expect((await events(name)).map((e) => e.kind)).toEqual([ + "attempt", + ]); + }, + { timeout: 10_000 }, + ); + await t.mutation((ctx) => pool.cancel(ctx, id)); + expect(await t.query((ctx) => pool.status(ctx, id))).toMatchObject({ + state: "running", + }); + expect(callback).not.toHaveBeenCalled(); + + release(); + await drain(); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + const recorded = await events(name); + expect(recorded.filter((e) => e.kind === "attempt")).toHaveLength(1); + expect(recorded.filter((e) => e.kind === "callback")).toMatchObject( + expected === "success" ? [] : [{ result: { kind: expected } }], + ); + if (expected === "success") { + expect(callback).not.toHaveBeenCalled(); + } else { + expect(callback).toHaveBeenCalledTimes(1); + expect(recorded[1]).toMatchObject({ + workId: id, + result: { kind: expected }, + }); + } + } finally { + release(); + } + }, + ); + + test.each(["action", "mutation", "query"] as const)( + "cancelAll invokes the cancellation callback for each pending %s in a batch", + async (kind) => { + const args = [{ key: "first" }, { key: "second" }]; + const options = { + onComplete: refs.complete, + onCompleteExcludeKinds: ["success"] as const, + context: { key: "batch-cancel" }, + runAt: Date.now() + 60_000, + }; + const ids = await t.mutation((ctx) => { + switch (kind) { + case "action": + return pool.enqueueActionBatch(ctx, refs.action, args, options); + case "mutation": + return pool.enqueueMutationBatch(ctx, refs.mutation, args, options); + case "query": + return pool.enqueueQueryBatch(ctx, refs.query, args, options); + } + }); + // convex-test gives consecutive inserts fractional creation timestamps. + // Move beyond those timestamps before cancelAll takes its cutoff. + await vi.advanceTimersByTimeAsync(1); + await t.mutation((ctx) => pool.cancelAll(ctx)); + await drain(); + expect(callback).toHaveBeenCalledTimes(2); + const recorded = await events("batch-cancel"); + expect(recorded.map((e) => e.workId).sort()).toEqual([...ids].sort()); + expect(recorded.every((e) => e.result?.kind === "canceled")).toBe(true); + expect(await t.query((ctx) => pool.statusBatch(ctx, ids))).toEqual([ + { state: "finished" }, + { state: "finished" }, + ]); + expect(await events("first")).toEqual([]); + expect(await events("second")).toEqual([]); + }, + ); + + test.each([false, true])( + "a cancellation filter restores large context (large arguments: %s)", + async (largeArgs) => { + const context = { key: "large-cancel", padding: "x".repeat(12_000) }; + const id = await enqueue( + "action", + { + key: context.key, + ...(largeArgs ? { payload: context.padding } : {}), + }, + { mode: "canceled", context, runAt: Date.now() + 60_000 }, + ); + await t.mutation((ctx) => pool.cancel(ctx, id)); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect(await events(context.key)).toMatchObject([ + { workId: id, context, result: { kind: "canceled" } }, + ]); + }, + ); + + test("a failing cancellation callback rolls back without starting the work", async () => { + const id = await enqueue( + "mutation", + { key: "broken-cancel" }, + { + mode: "canceled", + context: { key: "broken-cancel", throw: true }, + runAt: Date.now() + 60_000, + }, + ); + await t.mutation((ctx) => pool.cancel(ctx, id)); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect(await events("broken-cancel")).toEqual([]); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }); + + test.each(["action", "mutation", "query"] as const)( + "onComplete still receives successful %s results", + async (kind) => { + const id = await enqueue(kind, { key: kind }, { mode: "all" }); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect( + (await events(kind)).filter((e) => e.kind === "callback"), + ).toMatchObject([ + { + workId: id, + result: { kind: "success", returnValue: `${kind} result` }, + }, + ]); + }, + ); + + test.each([false, true])( + "a failure filter restores large context (large arguments: %s)", + async (largeArgs) => { + const context = { key: "large", padding: "x".repeat(12_000) }; + const id = await enqueue( + "action", + { + key: "large", + fail: true, + ...(largeArgs ? { payload: "x".repeat(12_000) } : {}), + }, + { context }, + ); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + const recorded = await events("large"); + expect(recorded.map((e) => e.kind)).toEqual(["attempt", "callback"]); + expect(recorded[1]).toMatchObject({ + context, + result: { kind: "failed", error: "work failed" }, + }); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }, + ); + + test.each(["failed", "canceled"] as const)( + "%s supports omitted context", + async (mode) => { + const id = await t.mutation((ctx) => + pool.enqueueMutation( + ctx, + refs.mutation, + { key: "default", fail: true }, + mode === "failed" + ? { + onComplete: refs.optionalComplete, + onCompleteExcludeKinds: ["success", "canceled"], + } + : { + onComplete: refs.optionalComplete, + onCompleteExcludeKinds: ["success", "failed"], + runAt: Date.now() + 60_000, + }, + ), + ); + if (mode === "canceled") { + await t.mutation((ctx) => pool.cancel(ctx, id)); + } + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect((await events("default"))[0]).toMatchObject({ + workId: id, + result: { kind: mode === "failed" ? "failed" : "canceled" }, + }); + expect((await events("default"))[0].context).toBeUndefined(); + }, + ); + + test.each(["action", "mutation", "query"] as const)( + "batch %s enqueue only calls back for failed items", + async (kind) => { + const args = [{ key: "success" }, { key: "failure", fail: true }]; + const opts = { + onComplete: refs.complete, + onCompleteExcludeKinds: ["success", "canceled"] as const, + context: { key: "batch" }, + }; + const ids = await t.mutation((ctx) => { + switch (kind) { + case "action": + return pool.enqueueActionBatch(ctx, refs.action, args, opts); + case "mutation": + return pool.enqueueMutationBatch(ctx, refs.mutation, args, opts); + case "query": + return pool.enqueueQueryBatch(ctx, refs.query, args, opts); + } + }); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect(await events("batch")).toMatchObject([ + { workId: ids[1], result: { kind: "failed" } }, + ]); + expect(await t.query((ctx) => pool.statusBatch(ctx, ids))).toEqual([ + { state: "finished" }, + { state: "finished" }, + ]); + }, + ); + + test("a failing failure callback rolls back without retrying the original work", async () => { + const id = await enqueue( + "action", + { key: "broken", fail: true }, + { retry: retries, context: { key: "broken", throw: true } }, + ); + await drain(); + expect(callback).toHaveBeenCalledTimes(1); + expect((await events("broken")).map((e) => e.kind)).toEqual([ + "attempt", + "attempt", + "attempt", + ]); + expect(await t.query((ctx) => pool.status(ctx, id))).toEqual({ + state: "finished", + }); + }); + + test.each([false, true])( + "rejects invalid result kinds at runtime (batch: %s)", + async (batch) => { + const opts = { + onComplete: refs.complete, + onCompleteExcludeKinds: ["running"], + context: { key: "invalid" }, + } as unknown as EnqueueOptions; + await expect( + t.mutation((ctx) => + batch + ? pool.enqueueMutationBatch( + ctx, + refs.mutation, + [{ key: "invalid" }], + opts, + ) + : pool.enqueueMutation( + ctx, + refs.mutation, + { key: "invalid" }, + opts, + ), + ), + ).rejects.toThrow(/Validator error/); + await drain(); + expect(await events("invalid")).toEqual([]); + expect(callback).not.toHaveBeenCalled(); + }, + ); + + test.each([false, true])( + "omits callback metadata when onComplete is disabled (batch: %s)", + async (batch) => { + for (const [index, options] of [{}, { onComplete: null }].entries()) { + const key = `disabled-${index}`; + await t.mutation((ctx) => + batch + ? pool.enqueueMutationBatch(ctx, refs.mutation, [{ key }], options) + : pool.enqueueMutation(ctx, refs.mutation, { key }, options), + ); + await drain(); + expect((await events(key)).map((event) => event.kind)).toEqual([ + "work", + ]); + } + expect(callback).not.toHaveBeenCalled(); + }, + ); +}); + type MutationCtx = GenericMutationCtx; const action = makeFunctionReference<"action", { value: number }, number>( "work:action", @@ -67,3 +819,54 @@ test("runAt and runAfter are mutually exclusive", () => { const options: EnqueueOptions = { runAt: 1, runAfter: 2 }; expectTypeOf(options).toMatchTypeOf(); }); + +test("exclude filters accept existing handlers on all enqueue methods", () => { + const options = { + onComplete, + onCompleteExcludeKinds: ["success"] as const, + context: { label: "job" }, + }; + const mutation = makeFunctionReference<"mutation", { value: number }, number>( + "work:mutation", + ); + const query = makeFunctionReference<"query", { value: number }, number>( + "work:query", + ); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueAction(ctx, action, { value: 1 }, options), + ).returns.toEqualTypeOf>(); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueMutation(ctx, mutation, { value: 1 }, options), + ).returns.toEqualTypeOf>(); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueQuery(ctx, query, { value: 1 }, options), + ).returns.toEqualTypeOf>(); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueActionBatch(ctx, action, [{ value: 1 }], options), + ).returns.toEqualTypeOf>(); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueMutationBatch(ctx, mutation, [{ value: 1 }], options), + ).returns.toEqualTypeOf>(); + expectTypeOf((pool: Workpool, ctx: MutationCtx) => + pool.enqueueQueryBatch(ctx, query, [{ value: 1 }], options), + ).returns.toEqualTypeOf>(); +}); + +test("exclude filters only accept terminal result kinds", () => { + // @ts-expect-error A running job has no terminal result. + const options: EnqueueOptions = { onCompleteExcludeKinds: ["running"] }; + expectTypeOf(options).toMatchTypeOf(); + expectTypeOf({ + onComplete: null, + onCompleteExcludeKinds: [], + }).toMatchTypeOf(); +}); + +test("generated enqueue types match the component validators", () => { + expectTypeOf< + FunctionArgs + >().toEqualTypeOf>(); + expectTypeOf< + FunctionArgs + >().toEqualTypeOf>(); +}); diff --git a/src/client/index.ts b/src/client/index.ts index 989a67df..662151b1 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -490,6 +490,13 @@ export type EnqueueOptions = { OnCompleteArgs > | null; + /** + * Do not invoke `onComplete` for these terminal result kinds. Omit it (or + * pass an empty array) to handle every result. Intermediate failed attempts + * that will be retried never invoke the callback. + */ + onCompleteExcludeKinds?: readonly RunResult["kind"][]; + /** * A context object to pass to the `onComplete` mutation. * Useful for passing data from the enqueue site to the onComplete site. @@ -577,6 +584,9 @@ async function enqueueArgs( ? { fnHandle: await createFunctionHandle(opts.onComplete), context: opts.context, + excludeKinds: opts.onCompleteExcludeKinds + ? [...opts.onCompleteExcludeKinds] + : undefined, } : undefined, runAt: getRunAt(opts), diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 2ea8267f..8facd992 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -69,7 +69,11 @@ export type ComponentApi = fnHandle: string; fnName: string; fnType: "action" | "mutation" | "query"; - onComplete?: { context?: any; fnHandle: string }; + onComplete?: { + context?: any; + excludeKinds?: Array<"success" | "failed" | "canceled">; + fnHandle: string; + }; retryBehavior?: { base: number; initialBackoffMs: number; @@ -93,7 +97,11 @@ export type ComponentApi = fnHandle: string; fnName: string; fnType: "action" | "mutation" | "query"; - onComplete?: { context?: any; fnHandle: string }; + onComplete?: { + context?: any; + excludeKinds?: Array<"success" | "failed" | "canceled">; + fnHandle: string; + }; retryBehavior?: { base: number; initialBackoffMs: number; diff --git a/src/component/complete.ts b/src/component/complete.ts index da03f26e..20ec9b81 100644 --- a/src/component/complete.ts +++ b/src/component/complete.ts @@ -126,7 +126,10 @@ export async function completeHandler( work.attempts < maxAttempts; if (!retry) { let scheduledId = undefined; - if (work.onComplete) { + if ( + work.onComplete && + !work.onComplete.excludeKinds?.includes(job.runResult.kind) + ) { try { // Retrieve large context if stored separately let context = work.onComplete.context; diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index d09aa10a..bf65459d 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -37,6 +37,80 @@ describe("lib", () => { }); describe("enqueue", () => { + it.each([false, true])( + "accepts legacy, empty, and populated callback exclusions (batch: %s)", + async (batch) => { + const callbacks = [ + { fnHandle: "callback", context: { key: "legacy" } }, + { fnHandle: "callback", excludeKinds: [] }, + { + fnHandle: "callback", + excludeKinds: ["failed", "canceled"] as const, + }, + ].map((callback) => ({ + ...callback, + excludeKinds: callback.excludeKinds && [...callback.excludeKinds], + })); + const items = callbacks.map((onComplete) => ({ + fnHandle: "workHandle", + fnName: "work", + fnArgs: {}, + fnType: "mutation" as const, + runAt: Date.now(), + onComplete, + })); + const ids = batch + ? await t.mutation(api.lib.enqueueBatch, { items, config: {} }) + : await Promise.all( + items.map((item) => + t.mutation(api.lib.enqueue, { ...item, config: {} }), + ), + ); + expect( + await t.run(async (ctx) => + Promise.all( + ids.map(async (id) => (await ctx.db.get("work", id))?.onComplete), + ), + ), + ).toEqual(callbacks); + }, + ); + + it.each([false, true])( + "rejects invalid status filters before writing any work (batch: %s)", + async (batch) => { + const item = { + fnHandle: "workHandle", + fnName: "work", + fnArgs: { payload: "x".repeat(10_000) }, + fnType: "mutation" as const, + runAt: Date.now(), + }; + const invalid = { + ...item, + onComplete: { + fnHandle: "callback", + // @ts-expect-error JavaScript callers can send an invalid kind. + excludeKinds: ["running"] as ("success" | "failed" | "canceled")[], + }, + }; + await expect( + batch + ? t.mutation(api.lib.enqueueBatch, { + items: [item, invalid], + config: {}, + }) + : t.mutation(api.lib.enqueue, { ...invalid, config: {} }), + ).rejects.toThrow(/Validator error/); + await t.run(async (ctx) => { + expect(await ctx.db.query("work").collect()).toEqual([]); + expect(await ctx.db.query("pendingStart").collect()).toEqual([]); + expect(await ctx.db.query("payload").collect()).toEqual([]); + expect(await ctx.db.query("globals").collect()).toEqual([]); + }); + }, + ); + it("should successfully enqueue a work item", async () => { const id = await t.mutation(api.lib.enqueue, { fnHandle: "testHandle", diff --git a/src/component/recovery.test.ts b/src/component/recovery.test.ts index 32f97005..f72452ec 100644 --- a/src/component/recovery.test.ts +++ b/src/component/recovery.test.ts @@ -1,4 +1,6 @@ import { convexTest } from "convex-test"; +import { createFunctionHandle, makeFunctionReference } from "convex/server"; +import { v } from "convex/values"; import type { DocumentByName, GenericDatabaseReader, @@ -18,15 +20,35 @@ import { } from "vitest"; import { internal } from "./_generated/api.js"; import type { Doc, Id } from "./_generated/dataModel.js"; -import type { MutationCtx } from "./_generated/server.js"; +import { internalMutation, type MutationCtx } from "./_generated/server.js"; import batchWorker from "@convex-dev/batch-worker/test"; import { recoveryHandler } from "./recovery.js"; import schema from "./schema.js"; import { modules } from "./setup.test.js"; +import { type OnCompleteArgs, vResult } from "./shared.js"; + +// "none" matches no kind, so it filters to all three: exclude everything. +const ALL_KINDS = ["success", "failed", "canceled"] as const; describe("recovery", () => { + const callback = vi.fn<(args: OnCompleteArgs) => void>(); + const callbackRef = makeFunctionReference<"mutation", OnCompleteArgs>( + "recoveryCallbacks:complete", + ); async function setupTest() { - const t = convexTest(schema, modules); + const t = convexTest(schema, { + ...modules, + "./recoveryCallbacks.ts": async () => ({ + complete: internalMutation({ + args: { + workId: v.id("work"), + context: v.any(), + result: vResult, + }, + handler: async (_ctx, args) => callback(args), + }), + }), + }); batchWorker.register(t); return t; } @@ -64,6 +86,7 @@ describe("recovery", () => { beforeEach(async () => { vi.useFakeTimers(); + callback.mockClear(); t = await setupTest(); // Set up globals for logging @@ -79,6 +102,75 @@ describe("recovery", () => { vi.useRealTimers(); }); + describe.each(["all", "failed", "canceled", "success", "none"] as const)( + "%s callbacks", + (mode) => { + it.each([ + ["missing", "Scheduled job not found"], + ["failed", "Function execution failed"], + ["canceled", "Canceled via scheduler"], + ] as const)( + "dispatches the appropriate callback for a %s scheduled job", + async (state, error) => { + const job = await t.run(async (ctx) => { + const handle = await createFunctionHandle(callbackRef); + const workId = await makeDummyWork(ctx, { + onComplete: { + fnHandle: handle, + context: { key: state }, + excludeKinds: + mode === "all" + ? undefined + : ALL_KINDS.filter((k) => k !== mode), + }, + }); + const scheduledId = await makeDummyScheduledFunction(ctx, workId); + // Prevent the placeholder worker from running when callbacks are drained. + await ctx.scheduler.cancel(scheduledId); + return { workId, scheduledId, attempt: 0, started: Date.now() }; + }); + await t.run(async (ctx) => { + const scheduled = await ctx.db.system.get( + "_scheduled_functions", + job.scheduledId, + ); + assert(scheduled); + // Only emulate the scheduler state; recovery and callback dispatch run + // normally, including scheduling and executing the callback mutation. + ctx.db.system.get = patchedSystemGet(ctx.db, { + [job.scheduledId]: + state === "missing" + ? null + : { + ...scheduled, + state: + state === "failed" + ? { kind: state, error } + : { kind: state }, + }, + }); + await recoveryHandler(ctx, { jobs: [job] }); + }); + // A repeated recovery scan must not dispatch the callback again. + await t.mutation(internal.recovery.recover, { jobs: [job] }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + if (mode !== "all" && mode !== "failed") { + expect(callback).not.toHaveBeenCalled(); + } else { + expect(callback).toHaveBeenCalledExactlyOnceWith({ + workId: job.workId, + context: { key: state }, + result: { kind: "failed", error }, + }); + } + expect( + await t.run((ctx) => ctx.db.get("work", job.workId)), + ).toBeNull(); + }, + ); + }, + ); + describe("recover", () => { it("should skip jobs that already have a pendingCompletion", async () => { // Create work and scheduled function diff --git a/src/component/shared.ts b/src/component/shared.ts index 371db8bb..6c778a76 100644 --- a/src/component/shared.ts +++ b/src/component/shared.ts @@ -112,6 +112,11 @@ export type RunResult = export const vOnCompleteFnContext = v.object({ fnHandle: v.string(), // mutation context: v.optional(v.any()), + excludeKinds: v.optional( + v.array( + v.union(v.literal("success"), v.literal("failed"), v.literal("canceled")), + ), + ), }); export type OnCompleteArgs = {