diff --git a/README.md b/README.md index f30d5179..f9a108ff 100644 --- a/README.md +++ b/README.md @@ -160,12 +160,31 @@ await pool.enqueueAction(ctx, internal.email.send, args, { ``` `onFailure` runs after retries are exhausted, or immediately after a -`NonRetryableError`. It is skipped for successful jobs and jobs canceled through -the Workpool API. Canceling a scheduled function directly in the dashboard is -treated as a failure and can trigger retries and `onFailure`. Like `onComplete`, -it runs in a separate transaction from the work. A callback error does not retry -the original work. Use `onComplete` when you also need success or cancellation -notifications; specifying both options is an error. +`NonRetryableError`. It is skipped for successful and canceled results. +Canceling a scheduled function directly in the dashboard is treated as a failure +and can trigger retries and `onFailure`. Like `onComplete`, it runs in a +separate transaction from the work. A callback error does not retry the original +work. + +Use `onCancel` to handle a `{ kind: "canceled" }` result when cancellation +prevents work from starting or being retried. It accepts the same completion +handler and context, and can be combined with `onFailure`. For example, to send +both outcomes to the existing handler: + +```ts +await pool.enqueueAction(ctx, internal.email.send, args, { + onFailure: internal.email.emailSent, + onCancel: internal.email.emailSent, + context: { emailType: args.emailType, userId: args.userId }, + retry: false, // don't retry this action, as we can't guarantee idempotency. +}); +``` + +The two callbacks can also use different handlers. Both run in separate +transactions from the work, and neither can be combined with `onComplete`. +`onCancel` follows the final result, so requesting cancellation does not +necessarily invoke it: in-progress work can still finish with success or +failure. ### Idempotency @@ -333,8 +352,11 @@ options include: - `onFailure`: A mutation to run after a terminal failure, using the same arguments as `onComplete`. Not called on success or cancellation. Cannot be combined with `onComplete`. -- `context`: Any data you want to pass to the `onComplete` or `onFailure` - mutation. +- `onCancel`: A mutation to run after a canceled result, using the same + arguments as `onComplete`. Can be combined with `onFailure`, but not + `onComplete`. +- `context`: Any data you want to pass to the `onComplete`, `onFailure`, or + `onCancel` mutation. - `runAt` and `runAfter`: Similar to `ctx.scheduler.run*`, allows you to schedule the work to run later. By default it's immediate. @@ -431,7 +453,11 @@ 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` and `onCancel` runs if provided. ## Monitoring the workpool diff --git a/src/client/index.test.ts b/src/client/index.test.ts index b4a519e2..fb0b5732 100644 --- a/src/client/index.test.ts +++ b/src/client/index.test.ts @@ -81,6 +81,10 @@ const callback = vi.fn( return null; }, ); +const optionalCallback = vi.fn((...args: Parameters) => + callback(...args), +); +let actionGate: Promise | undefined; const fixtures = { attempt: internalMutationGeneric({ args: { key: v.string() }, @@ -99,6 +103,7 @@ const fixtures = { 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"); @@ -137,7 +142,7 @@ const fixtures = { optionalComplete: internalMutationGeneric({ args: vOnCompleteArgs(), returns: v.null(), - handler: callback, + handler: optionalCallback, }), }; const refs = ( @@ -153,7 +158,7 @@ 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 = "onComplete" | "onFailure"; +type Mode = "onComplete" | "onFailure" | "onCancel"; describe("completion callbacks through the client and scheduler", () => { let t: ReturnType; @@ -165,6 +170,8 @@ describe("completion callbacks through the client and scheduler", () => { beforeEach(() => { vi.useFakeTimers(); callback.mockClear(); + optionalCallback.mockClear(); + actionGate = undefined; t = setup(); }); afterEach(async () => { @@ -197,7 +204,9 @@ describe("completion callbacks through the client and scheduler", () => { const opts = mode === "onFailure" ? { onFailure: refs.complete, context, ...rest } - : { onComplete: refs.complete, context, ...rest }; + : mode === "onCancel" + ? { onCancel: refs.complete, context, ...rest } + : { onComplete: refs.complete, context, ...rest }; return t.mutation((ctx) => { switch (kind) { case "action": @@ -287,7 +296,7 @@ describe("completion callbacks through the client and scheduler", () => { }); }); - test.each(["onComplete", "onFailure"] as const)( + test.each(["onComplete", "onFailure", "onCancel"] as const)( "%s preserves cancellation semantics", async (mode) => { const id = await enqueue( @@ -313,6 +322,229 @@ describe("completion callbacks through the client and scheduler", () => { }, ); + test.each(["action", "mutation", "query"] as const)( + "onCancel skips successful and failed %s work", + async (kind) => { + const ids = [ + await enqueue(kind, { key: "success" }, { mode: "onCancel" }), + await enqueue( + kind, + { key: "failure", fail: true }, + { mode: "onCancel", 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("onCancel runs once when a job is canceled during retry backoff", async () => { + const id = await enqueue( + "action", + { key: "backoff", fail: true }, + { + mode: "onCancel", + 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 }, + { + onFailure: refs.optionalComplete, + onCancel: refs.complete, + 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 } }], + ); + expect(optionalCallback).toHaveBeenCalledTimes( + expected === "failed" ? 1 : 0, + ); + 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 onCancel for each pending %s in a batch", + async (kind) => { + const args = [{ key: "first" }, { key: "second" }]; + const options = { + onFailure: refs.optionalComplete, + onCancel: refs.complete, + 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(optionalCallback).not.toHaveBeenCalled(); + 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])( + "onCancel 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: "onCancel", 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 onCancel callback rolls back without starting the work", async () => { + const id = await enqueue( + "mutation", + { key: "broken-cancel" }, + { + mode: "onCancel", + 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) => { @@ -357,23 +589,31 @@ describe("completion callbacks through the client and scheduler", () => { }, ); - test("onFailure supports omitted context", async () => { - const id = await t.mutation((ctx) => - pool.enqueueMutation( - ctx, - refs.mutation, - { key: "default", fail: true }, - { onFailure: refs.optionalComplete }, - ), - ); - await drain(); - expect(callback).toHaveBeenCalledTimes(1); - expect((await events("default"))[0]).toMatchObject({ - workId: id, - result: { kind: "failed" }, - }); - expect((await events("default"))[0].context).toBeUndefined(); - }); + test.each(["onFailure", "onCancel"] as const)( + "%s supports omitted context", + async (mode) => { + const id = await t.mutation((ctx) => + pool.enqueueMutation( + ctx, + refs.mutation, + { key: "default", fail: true }, + mode === "onFailure" + ? { onFailure: refs.optionalComplete } + : { onCancel: refs.optionalComplete, runAt: Date.now() + 60_000 }, + ), + ); + if (mode === "onCancel") { + 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 === "onFailure" ? "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", @@ -424,33 +664,61 @@ describe("completion callbacks through the client and scheduler", () => { "rejects mixed completion options at runtime (batch: %s)", async (batch) => { // JavaScript callers can bypass TypeScript's mutually exclusive options. - const opts = { - onComplete: refs.complete, - onFailure: refs.complete, - 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("Cannot define both onComplete and onFailure"); + for (const outcomeCallbacks of [ + { onFailure: refs.complete }, + { onCancel: refs.complete }, + { onFailure: refs.complete, onCancel: refs.complete }, + ]) { + const opts = { + onComplete: refs.complete, + ...outcomeCallbacks, + 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("Cannot define both onComplete and onFailure"); + } await drain(); expect(await events("invalid")).toEqual([]); expect(callback).not.toHaveBeenCalled(); }, ); + + test.each([false, true])( + "omits callback metadata when outcome handlers are disabled (batch: %s)", + async (batch) => { + for (const [index, options] of [ + {}, + { onFailure: null, onCancel: 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; @@ -508,35 +776,46 @@ test("runAt and runAfter are mutually exclusive", () => { expectTypeOf(options).toMatchTypeOf(); }); -test("onFailure accepts existing onComplete handlers on all enqueue methods", () => { - const options = { onFailure: onComplete, 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.each(["onFailure", "onCancel", "both"] as const)( + "%s accepts existing onComplete handlers on all enqueue methods", + (mode) => { + const callbacks = + mode === "onFailure" + ? { onFailure: onComplete } + : mode === "onCancel" + ? { onCancel: onComplete } + : { onFailure: onComplete, onCancel: onComplete }; + const options = { ...callbacks, 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("onFailure preserves context typing", () => { +test("outcome callbacks preserve context typing", () => { expectTypeOf((pool: Workpool, ctx: MutationCtx) => { void pool.enqueueAction( ctx, @@ -548,18 +827,36 @@ test("onFailure preserves context typing", () => { context: { label: 1 }, }, ); + void pool.enqueueAction( + ctx, + action, + { value: 1 }, + { + onCancel: onComplete, + // @ts-expect-error The reused completion handler requires a string label. + context: { label: 1 }, + }, + ); }).returns.toBeVoid(); }); -test("onComplete and onFailure are mutually exclusive", () => { +test("onComplete and outcome callbacks are mutually exclusive", () => { // @ts-expect-error The completion options cannot be combined. const options: EnqueueOptions = { onComplete, onFailure: onComplete }; expectTypeOf(options).toMatchTypeOf(); + // @ts-expect-error The completion options cannot be combined. + const cancelOptions: EnqueueOptions = { onComplete, onCancel: onComplete }; + expectTypeOf(cancelOptions).toMatchTypeOf(); }); test("callback options can each be disabled with null", () => { expectTypeOf({ onComplete: null }).toMatchTypeOf(); expectTypeOf({ onFailure: null }).toMatchTypeOf(); + expectTypeOf({ onCancel: null }).toMatchTypeOf(); + expectTypeOf({ + onFailure: null, + onCancel: null, + }).toMatchTypeOf(); }); test("generated enqueue types match the component validators", () => { diff --git a/src/client/index.ts b/src/client/index.ts index 44f397c9..3685a8a9 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -462,7 +462,7 @@ export type EnqueueOptions = { */ name?: string; /** - * A context object to pass to the `onComplete` or `onFailure` mutation. + * A context object to pass to the `onComplete`, `onFailure`, or `onCancel` mutation. * Useful for passing data from the enqueue site to the onComplete site. */ context?: Context; @@ -470,7 +470,7 @@ export type EnqueueOptions = { | { /** * A mutation to run after the function succeeds, fails, or is canceled. - * Cannot be used together with `onFailure`. + * Cannot be used together with `onFailure` or `onCancel`. * The context type is for your use, feel free to provide a validator for it. * e.g. * ```ts @@ -498,22 +498,36 @@ export type EnqueueOptions = { OnCompleteArgs > | null; onFailure?: never; + onCancel?: never; } | { /** * A mutation to run when the function fails with no retries remaining, - * or throws a NonRetryableError. Not called on success or cancellation - * through the Workpool API. Recovery treats direct scheduler cancellation - * as a failure, which can trigger retries and this callback. + * or throws a NonRetryableError. Not called for successful or canceled results. + * Direct scheduler cancellation is treated as a failure, which can trigger + * retries and this callback. * Uses the same arguments and separate transaction as `onComplete`, * so an existing `onComplete` handler can be passed here unchanged. - * Cannot be used together with `onComplete`. + * Can be used together with `onCancel`, but not `onComplete`. */ onFailure?: FunctionReference< "mutation", FunctionVisibility, OnCompleteArgs > | null; + /** + * A mutation to run when work finishes with a canceled result, using the + * same arguments and separate transaction as `onComplete`. + * Cancellation prevents starting or retrying work; it does not stop an + * in-progress attempt. That attempt may still finish with success or failure. + * Direct scheduler cancellation is treated as a failure, not cancellation. + * Can be used together with `onFailure`, but not `onComplete`. + */ + onCancel?: FunctionReference< + "mutation", + FunctionVisibility, + OnCompleteArgs + > | null; onComplete?: never; } ) & @@ -588,8 +602,10 @@ async function enqueueArgs( Partial & { retryBehavior?: RetryBehavior }) | undefined, ) { - if (opts?.onComplete && opts.onFailure) { - throw new Error("Cannot define both onComplete and onFailure handlers."); + if (opts?.onComplete && (opts.onFailure || opts.onCancel)) { + throw new Error( + "Cannot define both onComplete and onFailure/onCancel handlers.", + ); } const [fnHandle, fnName] = typeof fn === "string" && fn.startsWith("function://") @@ -603,10 +619,15 @@ async function enqueueArgs( fnHandle: await createFunctionHandle(opts.onComplete), context: opts.context, } - : opts?.onFailure + : opts?.onFailure || opts?.onCancel ? { onStatusHandle: { - failed: await createFunctionHandle(opts.onFailure), + failed: opts.onFailure + ? await createFunctionHandle(opts.onFailure) + : undefined, + canceled: opts.onCancel + ? await createFunctionHandle(opts.onCancel) + : undefined, }, context: opts.context, } diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 76f1c7ef..7db2522e 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -71,7 +71,10 @@ export type ComponentApi = fnType: "action" | "mutation" | "query"; onComplete?: | { context?: any; fnHandle: string } - | { context?: any; onStatusHandle: { failed: string } }; + | { + context?: any; + onStatusHandle: { canceled?: string; failed?: string }; + }; retryBehavior?: { base: number; initialBackoffMs: number; @@ -97,7 +100,10 @@ export type ComponentApi = fnType: "action" | "mutation" | "query"; onComplete?: | { context?: any; fnHandle: string } - | { context?: any; onStatusHandle: { failed: string } }; + | { + context?: any; + onStatusHandle: { canceled?: string; failed?: string }; + }; retryBehavior?: { base: number; initialBackoffMs: number; diff --git a/src/component/complete.ts b/src/component/complete.ts index a4fe2759..1057ccad 100644 --- a/src/component/complete.ts +++ b/src/component/complete.ts @@ -131,8 +131,8 @@ export async function completeHandler( work.onComplete && ("fnHandle" in work.onComplete ? work.onComplete.fnHandle - : job.runResult.kind === "failed" - ? work.onComplete.onStatusHandle.failed + : job.runResult.kind !== "success" + ? work.onComplete.onStatusHandle[job.runResult.kind] : undefined); if (work.onComplete && fnHandle) { try { diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index 459b4391..c50d2dc0 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -38,7 +38,7 @@ describe("lib", () => { describe("enqueue", () => { it.each([false, true])( - "accepts legacy and failure callback handles (batch: %s)", + "accepts legacy and outcome callback handles (batch: %s)", async (batch) => { const callbacks = [ { fnHandle: "completeHandle", context: { key: "complete" } }, @@ -46,6 +46,17 @@ describe("lib", () => { onStatusHandle: { failed: "failureHandle" }, context: { key: "failure" }, }, + { + onStatusHandle: { canceled: "cancelHandle" }, + context: { key: "cancel" }, + }, + { + onStatusHandle: { + failed: "failureHandle", + canceled: "cancelHandle", + }, + context: { key: "both" }, + }, ]; const items = callbacks.map((onComplete) => ({ fnHandle: "workHandle", @@ -74,27 +85,73 @@ describe("lib", () => { "rejects mixed callback handles at the component boundary (batch: %s)", async (batch) => { // Bypass the client guard to exercise the component's own validator. + for (const onStatusHandle of [ + { failed: "failureHandle" }, + { canceled: "cancelHandle" }, + { failed: "failureHandle", canceled: "cancelHandle" }, + ]) { + const item = { + fnHandle: "workHandle", + fnName: "work", + fnArgs: {}, + fnType: "mutation" as const, + runAt: Date.now(), + onComplete: { + fnHandle: "completeHandle", + onStatusHandle, + }, + }; + await expect( + batch + ? t.mutation(api.lib.enqueueBatch, { items: [item], config: {} }) + : t.mutation(api.lib.enqueue, { ...item, 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("globals").collect()).toEqual([]); + }); + }, + ); + + it.each([false, true])( + "rejects empty outcome callbacks without enqueueing work (batch: %s)", + async (batch) => { const item = { fnHandle: "workHandle", fnName: "work", - fnArgs: {}, + fnArgs: { payload: "x".repeat(10_000) }, fnType: "mutation" as const, runAt: Date.now(), - onComplete: { - fnHandle: "completeHandle", - onStatusHandle: { failed: "failureHandle" }, - }, }; - await expect( - batch - ? t.mutation(api.lib.enqueueBatch, { items: [item], config: {} }) - : t.mutation(api.lib.enqueue, { ...item, 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("globals").collect()).toEqual([]); - }); + for (const onStatusHandle of [{}, { failed: "", canceled: "" }]) { + const invalid = { ...item, onComplete: { onStatusHandle } }; + await expect( + batch + ? t.mutation(api.lib.enqueueBatch, { + items: [ + { + ...item, + onComplete: { + onStatusHandle: { failed: "failureHandle" }, + }, + }, + invalid, + ], + config: {}, + }) + : t.mutation(api.lib.enqueue, { ...invalid, config: {} }), + ).rejects.toThrow( + "onStatusHandle must contain at least one callback handle", + ); + 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([]); + }); + } }, ); diff --git a/src/component/lib.ts b/src/component/lib.ts index a6005b63..67a0da33 100644 --- a/src/component/lib.ts +++ b/src/component/lib.ts @@ -43,6 +43,21 @@ const itemArgs = { onComplete: v.optional(vOnCompleteFnContext), retryBehavior: v.optional(retryBehavior), }; + +function validateOnComplete( + onComplete: ObjectType["onComplete"], +) { + if ( + onComplete && + "onStatusHandle" in onComplete && + !Object.values(onComplete.onStatusHandle).some(Boolean) + ) { + throw new Error( + "onStatusHandle must contain at least one callback handle.", + ); + } +} + const enqueueArgs = { ...itemArgs, config: vConfig.partial(), @@ -51,6 +66,7 @@ export const enqueue = mutation({ args: enqueueArgs, returns: v.id("work"), handler: async (ctx, { config, ...itemArgs }) => { + validateOnComplete(itemArgs.onComplete); const globals = await getOrUpdateGlobals(ctx, config); const console = createLogger(globals.logLevel); await kickMainLoop(ctx, "enqueue"); @@ -127,6 +143,10 @@ export const enqueueBatch = mutation({ }, returns: v.array(v.id("work")), handler: async (ctx, { config, items }) => { + // Validate the entire batch before enqueueing any of its work. + for (const item of items) { + validateOnComplete(item.onComplete); + } const globals = await getOrUpdateGlobals(ctx, config); const console = createLogger(globals.logLevel); await kickMainLoop(ctx, "enqueue"); diff --git a/src/component/recovery.test.ts b/src/component/recovery.test.ts index fb033be3..747bab33 100644 --- a/src/component/recovery.test.ts +++ b/src/component/recovery.test.ts @@ -99,62 +99,74 @@ describe("recovery", () => { vi.useRealTimers(); }); - describe.each(["onComplete", "onFailure"] as const)( + describe.each(["onComplete", "onFailure", "onCancel"] as const)( "%s callbacks", (mode) => { it.each([ ["missing", "Scheduled job not found"], ["failed", "Function execution failed"], ["canceled", "Canceled via scheduler"], - ] as const)("runs once 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: - mode === "onComplete" - ? { fnHandle: handle, context: { key: state } } - : { - onStatusHandle: { failed: handle }, - context: { key: state }, - }, + ] 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: + mode === "onComplete" + ? { fnHandle: handle, context: { key: state } } + : { + onStatusHandle: + mode === "onFailure" + ? { failed: handle } + : { canceled: handle }, + context: { key: state }, + }, + }); + 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() }; }); - 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 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] }); }); - 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); - 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(); - }); + // 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 === "onCancel") { + 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(); + }, + ); }, ); diff --git a/src/component/shared.ts b/src/component/shared.ts index 834a0f65..257c21cc 100644 --- a/src/component/shared.ts +++ b/src/component/shared.ts @@ -117,7 +117,11 @@ export const vOnCompleteFnContext = v.union( context: v.optional(v.any()), }), v.object({ - onStatusHandle: v.object({ failed: v.string() }), // mutations + // Enqueue mutations require at least one nonempty callback handle. + onStatusHandle: v.object({ + failed: v.optional(v.string()), // mutation + canceled: v.optional(v.string()), // mutation + }), context: v.optional(v.any()), }), );