From 4f37f6c7d2dc5e490f4faba7f720a5a870f35298 Mon Sep 17 00:00:00 2001 From: Ian Macartney <366683+ianmacartney@users.noreply.github.com> Date: Wed, 9 Sep 2026 22:44:04 -0700 Subject: [PATCH 1/4] Filter completion callbacks by terminal result status --- README.md | 37 +- src/client/index.test.ts | 807 +++++++++++++++++++++++++- src/client/index.ts | 10 + src/component/_generated/component.ts | 12 +- src/component/complete.ts | 6 +- src/component/lib.test.ts | 71 +++ src/component/recovery.test.ts | 91 ++- src/component/shared.ts | 5 + 8 files changed, 1028 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 05fec59e..0bc47033 100644 --- a/README.md +++ b/README.md @@ -147,6 +147,30 @@ export const emailSent = pool.defineOnComplete({ }); ``` +To handle only some outcomes, add `onCompleteStatuses`. 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, + onCompleteStatuses: ["failed", "canceled"], + context: { emailType: args.emailType, userId: args.userId }, + retry: false, +}); +``` + +The available statuses match `result.kind`: `"success"`, `"failed"`, and +`"canceled"`. Omitting the list handles every outcome; an empty list disables +the callback. Repeated statuses do not cause repeated calls. The handler keeps +the same arguments and transaction behavior as an unfiltered `onComplete`. + +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 only if `"failed"` is selected. Cancellation through the workpool +prevents starting or retrying work and yields `"canceled"`. Direct scheduler +cancellation (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 +334,8 @@ 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. +- `onCompleteStatuses`: Optional list of result kinds to handle: `"success"`, + `"failed"`, and `"canceled"`. Defaults to all; `[]` disables 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 +357,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 if selected by +`onCompleteStatuses`, and skip any remaining retries. ```ts import { NonRetryableError } from "@convex-dev/workpool"; @@ -407,7 +433,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 only if that final +result is selected by `onCompleteStatuses` (or the list is omitted). ## Monitoring the workpool diff --git a/src/client/index.test.ts b/src/client/index.test.ts index c6bda3c3..24099ac4 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -1,18 +1,768 @@ 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 statusFilters = [ + 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(statusFilters.map((statuses) => ({ statuses })))( + "status filter $statuses", + ({ statuses }) => { + 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, + onCompleteStatuses: statuses, + 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 = + statuses === undefined || + (statuses as readonly string[]).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, + onCompleteStatuses: mode === "all" ? undefined : [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, + onCompleteStatuses: ["failed", "canceled"], + 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, + onCompleteStatuses: ["failed", "canceled"] 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, + onCompleteStatuses: ["failed"], + } + : { + onComplete: refs.optionalComplete, + onCompleteStatuses: ["canceled"], + 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, + onCompleteStatuses: ["failed"] 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 status filters at runtime (batch: %s)", + async (batch) => { + const opts = { + onComplete: refs.complete, + onCompleteStatuses: ["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 +817,54 @@ test("runAt and runAfter are mutually exclusive", () => { const options: EnqueueOptions = { runAt: 1, runAfter: 2 }; expectTypeOf(options).toMatchTypeOf(); }); + +test("status filters accept existing handlers on all enqueue methods", () => { + const options = { + onComplete, + onCompleteStatuses: ["success", "failed", "canceled"] 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("status filters only accept terminal result kinds", () => { + // @ts-expect-error A running job has no terminal result. + const options: EnqueueOptions = { onCompleteStatuses: ["running"] }; + expectTypeOf(options).toMatchTypeOf(); + expectTypeOf({ + onComplete: null, + onCompleteStatuses: [], + }).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..c63bc265 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -490,6 +490,13 @@ export type EnqueueOptions = { OnCompleteArgs > | null; + /** + * Only invoke `onComplete` for these terminal result kinds. Omit to handle + * every result, or pass an empty array to disable the callback. Intermediate + * failed attempts that will be retried never invoke the callback. + */ + onCompleteStatuses?: 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, + statuses: opts.onCompleteStatuses + ? [...opts.onCompleteStatuses] + : undefined, } : undefined, runAt: getRunAt(opts), diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 2ea8267f..4c66ec03 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; + fnHandle: string; + statuses?: Array<"success" | "failed" | "canceled">; + }; 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; + fnHandle: string; + statuses?: Array<"success" | "failed" | "canceled">; + }; retryBehavior?: { base: number; initialBackoffMs: number; diff --git a/src/component/complete.ts b/src/component/complete.ts index da03f26e..22b62848 100644 --- a/src/component/complete.ts +++ b/src/component/complete.ts @@ -126,7 +126,11 @@ export async function completeHandler( work.attempts < maxAttempts; if (!retry) { let scheduledId = undefined; - if (work.onComplete) { + if ( + work.onComplete && + (work.onComplete.statuses === undefined || + work.onComplete.statuses.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..d97d098f 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -37,6 +37,77 @@ describe("lib", () => { }); describe("enqueue", () => { + it.each([false, true])( + "accepts legacy, empty, and selected callback statuses (batch: %s)", + async (batch) => { + const callbacks = [ + { fnHandle: "callback", context: { key: "legacy" } }, + { fnHandle: "callback", statuses: [] }, + { fnHandle: "callback", statuses: ["failed", "canceled"] as const }, + ].map((callback) => ({ + ...callback, + statuses: callback.statuses && [...callback.statuses], + })); + 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 status. + statuses: ["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..5bd319f7 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,32 @@ 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"; 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 +83,7 @@ describe("recovery", () => { beforeEach(async () => { vi.useFakeTimers(); + callback.mockClear(); t = await setupTest(); // Set up globals for logging @@ -79,6 +99,73 @@ 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 }, + statuses: + mode === "all" ? undefined : mode === "none" ? [] : [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..2d2e47ff 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()), + statuses: v.optional( + v.array( + v.union(v.literal("success"), v.literal("failed"), v.literal("canceled")), + ), + ), }); export type OnCompleteArgs = { From 97eee756dbee36f306fb275134d737fdffee7dbb Mon Sep 17 00:00:00 2001 From: Ian Macartney <366683+ianmacartney@users.noreply.github.com> Date: Thu, 10 Sep 2026 20:12:49 -0700 Subject: [PATCH 2/4] clarify readme --- README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0bc47033..05db9487 100644 --- a/README.md +++ b/README.md @@ -167,9 +167,11 @@ the same arguments and transaction behavior as an unfiltered `onComplete`. 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 only if `"failed"` is selected. Cancellation through the workpool -prevents starting or retrying work and yields `"canceled"`. Direct scheduler -cancellation (for example, from the dashboard) is treated as a failure and can -trigger retries instead. +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 cancellation (for example, from the dashboard) is treated as a failure +and can trigger retries instead. ### Idempotency From fcaa6b9cbab1c585fbcbf8c0067cf163ce0c1394 Mon Sep 17 00:00:00 2001 From: Ian Macartney <366683+ianmacartney@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:29:09 -0700 Subject: [PATCH 3/4] readme --- README.md | 27 ++++++++++++--------------- 1 file changed, 12 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 05db9487..2cfdc670 100644 --- a/README.md +++ b/README.md @@ -147,30 +147,26 @@ export const emailSent = pool.defineOnComplete({ }); ``` -To handle only some outcomes, add `onCompleteStatuses`. For example, the same +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, - onCompleteStatuses: ["failed", "canceled"], + onCompleteExcludeKinds: ["success"], context: { emailType: args.emailType, userId: args.userId }, - retry: false, }); ``` -The available statuses match `result.kind`: `"success"`, `"failed"`, and -`"canceled"`. Omitting the list handles every outcome; an empty list disables -the callback. Repeated statuses do not cause repeated calls. The handler keeps -the same arguments and transaction behavior as an unfiltered `onComplete`. +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 only if `"failed"` is selected. Cancellation through the workpool +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 cancellation (for example, from the dashboard) is treated as a failure +scheduler cancelation (for example, from the dashboard) is treated as a failure and can trigger retries instead. ### Idempotency @@ -336,8 +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. -- `onCompleteStatuses`: Optional list of result kinds to handle: `"success"`, - `"failed"`, and `"canceled"`. Defaults to all; `[]` disables the callback. +- `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. @@ -359,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 if selected by -`onCompleteStatuses`, 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"; @@ -439,8 +436,8 @@ 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 only if that final -result is selected by `onCompleteStatuses` (or the list is omitted). +retried, the final result is `canceled`. The callback runs unless that final +result is listed in `onCompleteExcludeKinds`. ## Monitoring the workpool From b395444f218459f3b7e90b83f222851f202b95ad Mon Sep 17 00:00:00 2001 From: Ian Macartney <366683+ianmacartney@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:29:52 -0700 Subject: [PATCH 4/4] do exclusion instead --- src/client/index.test.ts | 44 ++++++++++++++------------- src/client/index.ts | 12 ++++---- src/component/_generated/component.ts | 4 +-- src/component/complete.ts | 3 +- src/component/lib.test.ts | 15 +++++---- src/component/recovery.test.ts | 9 ++++-- src/component/shared.ts | 2 +- 7 files changed, 49 insertions(+), 40 deletions(-) diff --git a/src/client/index.test.ts b/src/client/index.test.ts index 24099ac4..e64d80f1 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -156,7 +156,8 @@ 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 statusFilters = [ +const ALL_KINDS = ["success", "failed", "canceled"] as const; +const excludeFilters = [ undefined, [], ["success"], @@ -195,9 +196,9 @@ describe("completion callbacks through the client and scheduler", () => { .collect(), ); - describe.each(statusFilters.map((statuses) => ({ statuses })))( - "status filter $statuses", - ({ statuses }) => { + 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) => { @@ -208,7 +209,7 @@ describe("completion callbacks through the client and scheduler", () => { { key: "filtered", fail: result === "failed" }, { onComplete: refs.complete, - onCompleteStatuses: statuses, + onCompleteExcludeKinds: excludeKinds, context: { key: "filtered" }, runAt: result === "canceled" ? Date.now() + 60_000 : undefined, }, @@ -217,9 +218,9 @@ describe("completion callbacks through the client and scheduler", () => { if (result === "canceled") await t.mutation((ctx) => pool.cancel(ctx, id)); await drain(); - const selected = - statuses === undefined || - (statuses as readonly string[]).includes(result); + 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", @@ -248,7 +249,8 @@ describe("completion callbacks through the client and scheduler", () => { const { mode = "failed", context = { key: args.key }, ...rest } = options; const opts = { onComplete: refs.complete, - onCompleteStatuses: mode === "all" ? undefined : [mode], + onCompleteExcludeKinds: + mode === "all" ? undefined : ALL_KINDS.filter((k) => k !== mode), context, ...rest, }; @@ -463,7 +465,7 @@ describe("completion callbacks through the client and scheduler", () => { { key: name, fail, ...args }, { onComplete: refs.complete, - onCompleteStatuses: ["failed", "canceled"], + onCompleteExcludeKinds: ["success"], context: { key: name }, retry, }, @@ -514,7 +516,7 @@ describe("completion callbacks through the client and scheduler", () => { const args = [{ key: "first" }, { key: "second" }]; const options = { onComplete: refs.complete, - onCompleteStatuses: ["failed", "canceled"] as const, + onCompleteExcludeKinds: ["success"] as const, context: { key: "batch-cancel" }, runAt: Date.now() + 60_000, }; @@ -641,11 +643,11 @@ describe("completion callbacks through the client and scheduler", () => { mode === "failed" ? { onComplete: refs.optionalComplete, - onCompleteStatuses: ["failed"], + onCompleteExcludeKinds: ["success", "canceled"], } : { onComplete: refs.optionalComplete, - onCompleteStatuses: ["canceled"], + onCompleteExcludeKinds: ["success", "failed"], runAt: Date.now() + 60_000, }, ), @@ -669,7 +671,7 @@ describe("completion callbacks through the client and scheduler", () => { const args = [{ key: "success" }, { key: "failure", fail: true }]; const opts = { onComplete: refs.complete, - onCompleteStatuses: ["failed"] as const, + onCompleteExcludeKinds: ["success", "canceled"] as const, context: { key: "batch" }, }; const ids = await t.mutation((ctx) => { @@ -713,11 +715,11 @@ describe("completion callbacks through the client and scheduler", () => { }); test.each([false, true])( - "rejects invalid status filters at runtime (batch: %s)", + "rejects invalid result kinds at runtime (batch: %s)", async (batch) => { const opts = { onComplete: refs.complete, - onCompleteStatuses: ["running"], + onCompleteExcludeKinds: ["running"], context: { key: "invalid" }, } as unknown as EnqueueOptions; await expect( @@ -818,10 +820,10 @@ test("runAt and runAfter are mutually exclusive", () => { expectTypeOf(options).toMatchTypeOf(); }); -test("status filters accept existing handlers on all enqueue methods", () => { +test("exclude filters accept existing handlers on all enqueue methods", () => { const options = { onComplete, - onCompleteStatuses: ["success", "failed", "canceled"] as const, + onCompleteExcludeKinds: ["success"] as const, context: { label: "job" }, }; const mutation = makeFunctionReference<"mutation", { value: number }, number>( @@ -850,13 +852,13 @@ test("status filters accept existing handlers on all enqueue methods", () => { ).returns.toEqualTypeOf>(); }); -test("status filters only accept terminal result kinds", () => { +test("exclude filters only accept terminal result kinds", () => { // @ts-expect-error A running job has no terminal result. - const options: EnqueueOptions = { onCompleteStatuses: ["running"] }; + const options: EnqueueOptions = { onCompleteExcludeKinds: ["running"] }; expectTypeOf(options).toMatchTypeOf(); expectTypeOf({ onComplete: null, - onCompleteStatuses: [], + onCompleteExcludeKinds: [], }).toMatchTypeOf(); }); diff --git a/src/client/index.ts b/src/client/index.ts index c63bc265..662151b1 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -491,11 +491,11 @@ export type EnqueueOptions = { > | null; /** - * Only invoke `onComplete` for these terminal result kinds. Omit to handle - * every result, or pass an empty array to disable the callback. Intermediate - * failed attempts that will be retried never invoke the callback. + * 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. */ - onCompleteStatuses?: readonly RunResult["kind"][]; + onCompleteExcludeKinds?: readonly RunResult["kind"][]; /** * A context object to pass to the `onComplete` mutation. @@ -584,8 +584,8 @@ async function enqueueArgs( ? { fnHandle: await createFunctionHandle(opts.onComplete), context: opts.context, - statuses: opts.onCompleteStatuses - ? [...opts.onCompleteStatuses] + excludeKinds: opts.onCompleteExcludeKinds + ? [...opts.onCompleteExcludeKinds] : undefined, } : undefined, diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 4c66ec03..8facd992 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -71,8 +71,8 @@ export type ComponentApi = fnType: "action" | "mutation" | "query"; onComplete?: { context?: any; + excludeKinds?: Array<"success" | "failed" | "canceled">; fnHandle: string; - statuses?: Array<"success" | "failed" | "canceled">; }; retryBehavior?: { base: number; @@ -99,8 +99,8 @@ export type ComponentApi = fnType: "action" | "mutation" | "query"; onComplete?: { context?: any; + excludeKinds?: Array<"success" | "failed" | "canceled">; fnHandle: string; - statuses?: Array<"success" | "failed" | "canceled">; }; retryBehavior?: { base: number; diff --git a/src/component/complete.ts b/src/component/complete.ts index 22b62848..20ec9b81 100644 --- a/src/component/complete.ts +++ b/src/component/complete.ts @@ -128,8 +128,7 @@ export async function completeHandler( let scheduledId = undefined; if ( work.onComplete && - (work.onComplete.statuses === undefined || - work.onComplete.statuses.includes(job.runResult.kind)) + !work.onComplete.excludeKinds?.includes(job.runResult.kind) ) { try { // Retrieve large context if stored separately diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index d97d098f..bf65459d 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -38,15 +38,18 @@ describe("lib", () => { describe("enqueue", () => { it.each([false, true])( - "accepts legacy, empty, and selected callback statuses (batch: %s)", + "accepts legacy, empty, and populated callback exclusions (batch: %s)", async (batch) => { const callbacks = [ { fnHandle: "callback", context: { key: "legacy" } }, - { fnHandle: "callback", statuses: [] }, - { fnHandle: "callback", statuses: ["failed", "canceled"] as const }, + { fnHandle: "callback", excludeKinds: [] }, + { + fnHandle: "callback", + excludeKinds: ["failed", "canceled"] as const, + }, ].map((callback) => ({ ...callback, - statuses: callback.statuses && [...callback.statuses], + excludeKinds: callback.excludeKinds && [...callback.excludeKinds], })); const items = callbacks.map((onComplete) => ({ fnHandle: "workHandle", @@ -87,8 +90,8 @@ describe("lib", () => { ...item, onComplete: { fnHandle: "callback", - // @ts-expect-error JavaScript callers can send an invalid status. - statuses: ["running"] as ("success" | "failed" | "canceled")[], + // @ts-expect-error JavaScript callers can send an invalid kind. + excludeKinds: ["running"] as ("success" | "failed" | "canceled")[], }, }; await expect( diff --git a/src/component/recovery.test.ts b/src/component/recovery.test.ts index 5bd319f7..f72452ec 100644 --- a/src/component/recovery.test.ts +++ b/src/component/recovery.test.ts @@ -27,6 +27,9 @@ 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>( @@ -115,8 +118,10 @@ describe("recovery", () => { onComplete: { fnHandle: handle, context: { key: state }, - statuses: - mode === "all" ? undefined : mode === "none" ? [] : [mode], + excludeKinds: + mode === "all" + ? undefined + : ALL_KINDS.filter((k) => k !== mode), }, }); const scheduledId = await makeDummyScheduledFunction(ctx, workId); diff --git a/src/component/shared.ts b/src/component/shared.ts index 2d2e47ff..6c778a76 100644 --- a/src/component/shared.ts +++ b/src/component/shared.ts @@ -112,7 +112,7 @@ export type RunResult = export const vOnCompleteFnContext = v.object({ fnHandle: v.string(), // mutation context: v.optional(v.any()), - statuses: v.optional( + excludeKinds: v.optional( v.array( v.union(v.literal("success"), v.literal("failed"), v.literal("canceled")), ),