diff --git a/sdks/node/src/errors.ts b/sdks/node/src/errors.ts index 7e442945..9ede48ad 100644 --- a/sdks/node/src/errors.ts +++ b/sdks/node/src/errors.ts @@ -150,3 +150,11 @@ export class PredicateRejectedError extends TaskitoError { this.name = "PredicateRejectedError"; } } + +/** Thrown when an enqueue interceptor rejects, misbehaves, or redirects illegally. */ +export class InterceptionError extends TaskitoError { + constructor(message: string) { + super(message); + this.name = "InterceptionError"; + } +} diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index 58920bd9..d3a38a25 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -2,6 +2,7 @@ export { currentJob, type JobContext } from "./context"; export { type DashboardAuth, type DashboardOptions, serveDashboard } from "./dashboard"; export { CryptoError, + InterceptionError, JobCancelledError, JobFailedError, LockLostError, @@ -19,6 +20,7 @@ export { WorkflowError, } from "./errors"; export type { EventHandler, EventName, OutcomeEvent } from "./events"; +export { Interception, type Interceptor } from "./interception"; export { Lock, type LockInfo, type LockOptions } from "./locks"; export type { EnqueueContext, Middleware, TaskContext } from "./middleware"; export { diff --git a/sdks/node/src/interception.ts b/sdks/node/src/interception.ts new file mode 100644 index 00000000..a668727b --- /dev/null +++ b/sdks/node/src/interception.ts @@ -0,0 +1,43 @@ +/** + * Producer-side enqueue interception. Interceptors registered via + * `Queue.intercept` run at the very start of every enqueue — before per-task + * defaults, middleware, gates, and serialization — and decide what happens + * to the call: + * + * - `pass` — enqueue unchanged. + * - `convert` — replace the args (task name unchanged), e.g. redact a field. + * - `redirect` — enqueue a different task with new args instead. + * - `reject` — block the enqueue; `Queue.enqueue` throws `InterceptionError`. + * + * Interceptors chain in registration order, each seeing the previous one's + * output. Distinct from `Middleware.onEnqueue` (mutate-in-place hooks) and + * from the Python SDK's argument-type interception. + */ +export type Interception = + | { readonly type: "pass" } + | { readonly type: "convert"; readonly args: unknown[] } + | { readonly type: "redirect"; readonly taskName: string; readonly args: unknown[] } + | { readonly type: "reject"; readonly reason: string }; + +/** Decides the outcome of an enqueue. Runs synchronously before serialization. */ +export type Interceptor = (taskName: string, args: unknown[]) => Interception; + +/** Factories for the four {@link Interception} outcomes. */ +export const Interception = { + /** Enqueue the original args unchanged. */ + pass(): Interception { + return { type: "pass" }; + }, + /** Replace the args; the task name stays the same. */ + convert(args: unknown[]): Interception { + return { type: "convert", args }; + }, + /** Enqueue a different task with new args instead of the original. */ + redirect(taskName: string, args: unknown[]): Interception { + return { type: "redirect", taskName, args }; + }, + /** Block the enqueue; the reason is surfaced on the thrown error. */ + reject(reason: string): Interception { + return { type: "reject", reason }; + }, +}; diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index c6ca0875..ce2f0624 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -1,6 +1,7 @@ import { mkdirSync } from "node:fs"; import { dirname } from "node:path"; import { + InterceptionError, JobCancelledError, JobFailedError, LockLostError, @@ -12,6 +13,7 @@ import { TaskitoError, } from "./errors"; import { Emitter, type EventHandler, type EventName } from "./events"; +import type { Interceptor } from "./interception"; import { Lock, type LockOptions } from "./locks"; import type { EnqueueContext, Middleware } from "./middleware"; import { @@ -100,6 +102,7 @@ export class Queue { private readonly tasks = new Map(); private readonly queueLimits = new Map(); private readonly middleware: Middleware[] = []; + private readonly interceptors: Interceptor[] = []; private readonly gates = new Map(); private readonly emitter = new Emitter(); private readonly resources = new ResourceRuntime(); @@ -281,6 +284,19 @@ export class Queue { * `onEnqueue`). If it returns `false`, the enqueue throws * {@link PredicateRejectedError}. Multiple gates on a task all must pass. */ + /** + * Register an enqueue interceptor. Interceptors run at the start of every + * enqueue — before defaults, middleware, and gates — chained in + * registration order, each seeing the previous one's task name and args. + * Returning `Interception.reject(...)` (or a null-ish value) makes the + * enqueue throw {@link InterceptionError}; `redirect` is not supported for + * batch enqueue or for tasks with per-task codecs. + */ + intercept(interceptor: Interceptor): this { + this.interceptors.push(interceptor); + return this; + } + gate( name: Name, predicate: (ctx: { taskName: Name; args: Parameters }) => boolean, @@ -309,8 +325,8 @@ export class Queue { args?: Parameters, options?: EnqueueOptions, ): string { - const { payload, options: nativeOpts } = this.prepareEnqueue(name, args, options); - return this.native.enqueue(name, payload, nativeOpts); + const { taskName, payload, options: nativeOpts } = this.prepareEnqueue(name, args, options); + return this.native.enqueue(taskName, payload, nativeOpts); } /** @@ -322,43 +338,100 @@ export class Queue { name: Name, jobs: ReadonlyArray<{ args?: Parameters; options?: EnqueueOptions }>, ): string[] { - const prepared = jobs.map((job) => this.prepareEnqueue(name, job.args, job.options)); + const prepared = jobs.map((job) => { + const { payload, options } = this.prepareEnqueue(name, job.args, job.options, { + batch: true, + }); + return { payload, options }; + }); return this.native.enqueueMany(name, prepared); } /** - * Merge per-task defaults, run the `onEnqueue` interception hooks, then - * serialize the args and encode the options — the shared path for - * {@link Queue.enqueue} and {@link Queue.enqueueMany}. + * Run enqueue interceptors, merge per-task defaults, run the middleware + * `onEnqueue` hooks, then serialize the args and encode the options — the + * shared path for {@link Queue.enqueue} and {@link Queue.enqueueMany}. + * Everything downstream of the interceptors keys off the (possibly + * redirected) final task name. */ private prepareEnqueue( name: Name, args: Parameters | undefined, options: EnqueueOptions | undefined, - ): { payload: Buffer; options: NativeEnqueueOptions } { - const defaults = this.tasks.get(name)?.options; + mode?: { batch: boolean }, + ): { taskName: string; payload: Buffer; options: NativeEnqueueOptions } { + const { taskName, args: finalArgs } = this.runInterceptors(name, [...(args ?? [])], mode); + const defaults = this.tasks.get(taskName)?.options; const merged: EnqueueOptions = { ...options, maxRetries: options?.maxRetries ?? defaults?.maxRetries, timeoutMs: options?.timeoutMs ?? defaults?.timeoutMs, }; - // Interception seam: let middleware validate/redact/rewrite before serializing. - const ctx: EnqueueContext = { taskName: name, args: [...(args ?? [])], options: merged }; + // Middleware seam: let onEnqueue hooks validate/redact/rewrite before serializing. + const ctx: EnqueueContext = { taskName, args: finalArgs, options: merged }; for (const mw of this.middleware) { mw.onEnqueue?.(ctx); } // Gate: predicates see the (possibly rewritten) args and may reject the enqueue. - for (const gate of this.gates.get(name) ?? []) { - if (!gate({ taskName: name, args: ctx.args })) { - throw new PredicateRejectedError(name); + for (const gate of this.gates.get(taskName) ?? []) { + if (!gate({ taskName, args: ctx.args })) { + throw new PredicateRejectedError(taskName); } } return { - payload: this.encodeTaskPayload(name, ctx.args), + taskName, + payload: this.encodeTaskPayload(taskName, ctx.args), options: toNativeEnqueueOptions(ctx.options), }; } + /** + * Chain the registered interceptors over `(taskName, args)`: a null-ish + * outcome or `reject` throws; `redirect` swaps both + * the task name and args, and is rejected for batch enqueue (the batch is + * stored under one task name) and for tasks with per-task codecs (the + * redirect target's codec chain cannot be resolved from a bare name — + * a cross-SDK behavioral contract). + */ + private runInterceptors( + name: string, + args: unknown[], + mode?: { batch: boolean }, + ): { taskName: string; args: unknown[] } { + let taskName = name; + let currentArgs = args; + for (const interceptor of this.interceptors) { + const outcome = interceptor(taskName, currentArgs); + if (!outcome) { + throw new InterceptionError(`interceptor returned null for task "${taskName}"`); + } + switch (outcome.type) { + case "pass": + break; + case "convert": + currentArgs = outcome.args; + break; + case "redirect": + if (mode?.batch) { + throw new InterceptionError( + `interceptor Redirect is not supported for batch enqueue of task "${taskName}"`, + ); + } + taskName = outcome.taskName; + currentArgs = outcome.args; + break; + case "reject": + throw new InterceptionError(`enqueue of "${taskName}" rejected: ${outcome.reason}`); + } + } + if (taskName !== name && (this.tasks.get(name)?.options?.codecs?.length ?? 0) > 0) { + throw new InterceptionError( + `interceptor Redirect is not supported for a task with payload codecs ("${name}")`, + ); + } + return { taskName, args: currentArgs }; + } + /** * Serialize a task payload and apply the task's named codecs in order. * Payload only — results stay on the queue serializer. diff --git a/sdks/node/test/core/interceptors.test.ts b/sdks/node/test/core/interceptors.test.ts new file mode 100644 index 00000000..641f051b --- /dev/null +++ b/sdks/node/test/core/interceptors.test.ts @@ -0,0 +1,147 @@ +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { GzipCodec, Interception, InterceptionError, Queue, type Worker } from "../../src/index"; + +let worker: Worker | undefined; + +afterEach(() => { + worker?.stop(); + worker = undefined; +}); + +function tempDb(): string { + return join(mkdtempSync(join(tmpdir(), "taskito-intercept-")), "queue.db"); +} + +async function waitFor(read: () => T | undefined, timeoutMs = 10_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const value = read(); + if (value !== undefined) { + return value; + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } + throw new Error("timed out waiting for job result"); +} + +describe("enqueue interceptors", () => { + it("convert rewrites the args the worker sees", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.echo", (x: number) => x) + .intercept((_taskName, args) => Interception.convert([(args[0] as number) * 10])); + + const id = queue.enqueue("ic.echo", [5]); + worker = queue.runWorker(); + await expect(waitFor(() => queue.getResult(id))).resolves.toBe(50); + }); + + it("redirect enqueues the other task; the original handler never runs", async () => { + const ranA: unknown[] = []; + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.a", (x: number) => { + ranA.push(x); + return x; + }) + .task("ic.b", (x: number) => x + 100) + .intercept((taskName, args) => + taskName === "ic.a" ? Interception.redirect("ic.b", args) : Interception.pass(), + ); + + const id = queue.enqueue("ic.a", [1]); + expect(queue.getJob(id)?.taskName).toBe("ic.b"); + worker = queue.runWorker(); + await expect(waitFor(() => queue.getResult(id))).resolves.toBe(101); + expect(ranA).toEqual([]); + }); + + it("reject throws InterceptionError and enqueues nothing", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.noop", () => null) + .intercept(() => Interception.reject("blocked by policy")); + + expect(() => queue.enqueue("ic.noop", [])).toThrow(InterceptionError); + expect(() => queue.enqueue("ic.noop", [])).toThrow(/blocked by policy/); + expect((await queue.stats()).pending).toBe(0); + }); + + it("a null-ish interceptor result throws InterceptionError", () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.noop", () => null) + // biome-ignore lint/suspicious/noExplicitAny: deliberately misbehaving interceptor + .intercept((() => undefined) as any); + + expect(() => queue.enqueue("ic.noop", [])).toThrow(/returned null/); + }); + + it("interceptors chain in registration order", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.echo", (x: number) => x) + .intercept((_name, args) => Interception.convert([(args[0] as number) + 1])) + .intercept((_name, args) => Interception.convert([(args[0] as number) * 2])); + + const id = queue.enqueue("ic.echo", [3]); // (3 + 1) * 2 + worker = queue.runWorker(); + await expect(waitFor(() => queue.getResult(id))).resolves.toBe(8); + }); + + it("batch enqueue converts per item", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.echo", (x: number) => x) + .intercept((_name, args) => Interception.convert([(args[0] as number) * 10])); + + const ids = queue.enqueueMany("ic.echo", [{ args: [1] }, { args: [2] }]); + worker = queue.runWorker(); + await expect(waitFor(() => queue.getResult(ids[0] as string))).resolves.toBe(10); + await expect(waitFor(() => queue.getResult(ids[1] as string))).resolves.toBe(20); + }); + + it("batch reject fails the whole batch — zero jobs stored", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.echo", (x: number) => x) + .intercept((_name, args) => + (args[0] as number) < 0 ? Interception.reject("negative") : Interception.pass(), + ); + + expect(() => + queue.enqueueMany("ic.echo", [{ args: [1] }, { args: [-2] }, { args: [3] }]), + ).toThrow(InterceptionError); + expect((await queue.stats()).pending).toBe(0); + }); + + it("batch redirect is unsupported", () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.a", (x: number) => x) + .task("ic.b", (x: number) => x) + .intercept((_name, args) => Interception.redirect("ic.b", args)); + + expect(() => queue.enqueueMany("ic.a", [{ args: [1] }])).toThrow( + /Redirect is not supported for batch/, + ); + }); + + it("redirect of a task with per-task codecs is rejected", () => { + const queue = new Queue({ dbPath: tempDb(), codecs: { gzip: new GzipCodec() } }) + .task("ic.codec", (x: number) => x, { codecs: ["gzip"] }) + .task("ic.b", (x: number) => x) + .intercept((_name, args) => Interception.redirect("ic.b", args)); + + expect(() => queue.enqueue("ic.codec", [1])).toThrow( + /Redirect is not supported for a task with payload codecs/, + ); + }); + + it("interceptors run before gates: a gate sees the converted args", async () => { + const queue = new Queue({ dbPath: tempDb() }) + .task("ic.echo", (x: number) => x) + .gate("ic.echo", ({ args }) => (args[0] as number) > 0) + .intercept((_name, args) => Interception.convert([Math.abs(args[0] as number)])); + + // Raw -5 would be gated; the interceptor converts to 5 first. + const id = queue.enqueue("ic.echo", [-5]); + worker = queue.runWorker(); + await expect(waitFor(() => queue.getResult(id))).resolves.toBe(5); + }); +});