From 491d9ee428bbda1319b41358d60f891cf0ce0f25 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:58:48 +0530 Subject: [PATCH 1/2] feat(node): pooled resource scope --- sdks/node/src/errors.ts | 20 +- sdks/node/src/index.ts | 3 + sdks/node/src/queue.ts | 23 +- sdks/node/src/resources/index.ts | 3 + sdks/node/src/resources/pool.ts | 220 ++++++++++++++++ sdks/node/src/resources/runtime.ts | 97 ++++++- sdks/node/src/resources/types.ts | 36 ++- sdks/node/test/resources/resources.test.ts | 278 ++++++++++++++++++++- 8 files changed, 664 insertions(+), 16 deletions(-) create mode 100644 sdks/node/src/resources/pool.ts diff --git a/sdks/node/src/errors.ts b/sdks/node/src/errors.ts index 9d844b74..cb93697f 100644 --- a/sdks/node/src/errors.ts +++ b/sdks/node/src/errors.ts @@ -116,14 +116,28 @@ export class ResourceNotFoundError extends ResourceError { } } -/** Thrown when a task-scoped resource is resolved at worker scope. */ +/** Thrown when a resource is resolved from a scope that outlives it. */ export class ResourceScopeError extends ResourceError { - constructor(readonly resourceName: string) { - super(`Resource "${resourceName}" is task-scoped and cannot be resolved at worker scope`); + constructor( + readonly resourceName: string, + scope: string = "task", + context: string = "worker", + ) { + super( + `Resource "${resourceName}" is ${scope}-scoped and cannot be resolved at ${context} scope`, + ); this.name = "ResourceScopeError"; } } +/** Thrown when a pooled resource cannot be checked out before its acquire timeout. */ +export class ResourceUnavailableError extends ResourceError { + constructor(message: string) { + super(message); + this.name = "ResourceUnavailableError"; + } +} + // ── Workflows ─────────────────────────────────────────────────────────────── /** Thrown on workflow definition, submission, or query errors. */ diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts index eb4a1233..f5c69086 100644 --- a/sdks/node/src/index.ts +++ b/sdks/node/src/index.ts @@ -14,6 +14,7 @@ export { ResourceError, ResourceNotFoundError, ResourceScopeError, + ResourceUnavailableError, ResultTimeoutError, SerializationError, TaskitoError, @@ -37,6 +38,8 @@ export { Queue, type QueueOptions } from "./queue"; export { type MockResource, mockResource, + type PoolOptions, + type PoolStats, type ResourceContext, type ResourceDefinition, type ResourceMetrics, diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts index ce2f0624..24e11459 100644 --- a/sdks/node/src/queue.ts +++ b/sdks/node/src/queue.ts @@ -8,6 +8,7 @@ import { LockNotAcquiredError, PredicateRejectedError, QueueError, + ResourceError, ResultTimeoutError, SerializationError, TaskitoError, @@ -25,6 +26,7 @@ import { import { encodeNotes } from "./notes"; import type { Predicate } from "./predicates"; import { + type PoolOptions, type ResourceContext, type ResourceMetrics, ResourceRuntime, @@ -248,18 +250,31 @@ export class Queue { /** * Register an injectable resource. Worker-scoped (default) values are built * once and shared across the worker's lifetime; task-scoped values are built - * per job invocation. Reach them from a handler via `useResource(name)` or the - * declarative `inject` option on {@link Queue.task}. + * per job invocation; pooled values are checked out of a bounded pool per job + * and returned when it finishes (tune via `pool`). Reach them from a handler + * via `useResource(name)` or the declarative `inject` option on + * {@link Queue.task}. */ resource( name: string, factory: (ctx: ResourceContext) => T | Promise, - options?: { scope?: ResourceScope; dispose?: (value: T) => void | Promise }, + options?: { + scope?: ResourceScope; + dispose?: (value: T) => void | Promise; + pool?: PoolOptions; + }, ): this { + const scope = options?.scope ?? "worker"; + if (options?.pool && scope !== "pooled") { + throw new ResourceError( + `Resource "${name}": pool options require scope "pooled" (got "${scope}")`, + ); + } this.resources.register(name, { factory, - scope: options?.scope ?? "worker", + scope, dispose: options?.dispose, + pool: options?.pool, }); return this; } diff --git a/sdks/node/src/resources/index.ts b/sdks/node/src/resources/index.ts index 3ccc5fc8..6f0a0a14 100644 --- a/sdks/node/src/resources/index.ts +++ b/sdks/node/src/resources/index.ts @@ -1,7 +1,10 @@ export { runWithResolver, useResource } from "./context"; export { type MockResource, mockResource } from "./mock"; +export { ResourcePool } from "./pool"; export { ResourceRuntime, type TaskScope } from "./runtime"; export type { + PoolOptions, + PoolStats, ResourceContext, ResourceDefinition, ResourceMetrics, diff --git a/sdks/node/src/resources/pool.ts b/sdks/node/src/resources/pool.ts new file mode 100644 index 00000000..ffceea62 --- /dev/null +++ b/sdks/node/src/resources/pool.ts @@ -0,0 +1,220 @@ +import { ResourceUnavailableError } from "../errors"; +import { createLogger } from "../utils"; +import type { PoolOptions, PoolStats } from "./types"; + +const log = createLogger("resources"); + +const DEFAULT_POOL_SIZE = 4; +const DEFAULT_ACQUIRE_TIMEOUT_MS = 10_000; + +/** Lifecycle callbacks so the runtime's created/disposed counters stay accurate. */ +export interface PoolHooks { + /** Called after the factory builds an instance. */ + onCreated?: () => void; + /** Called after an instance is actually disposed (eviction / shutdown). */ + onDisposed?: () => void; +} + +/** An instance waiting in the pool, stamped when it (re-)entered the idle list. */ +interface IdleEntry { + instance: unknown; + createdAt: number; +} + +/** A pending `acquire` waiting for a permit; granted FIFO by `release`. */ +interface Waiter { + grant: () => void; +} + +/** + * Promise-based counting semaphore with FIFO waiters and per-waiter timeout. + * `acquire` resolves `false` on timeout — never rejects — so the pool decides + * what a timeout means. Private to the pool; not part of the public API. + */ +class Semaphore { + private permits: number; + private readonly waiters: Waiter[] = []; + + constructor(permits: number) { + this.permits = permits; + } + + /** Take a permit, waiting up to `timeoutMs`. Resolves `false` on timeout. */ + acquire(timeoutMs: number): Promise { + if (this.permits > 0) { + this.permits -= 1; + return Promise.resolve(true); + } + return new Promise((resolve) => { + // On timeout the waiter leaves the queue, so a permit released later goes + // to the next waiter instead of a caller that already gave up. The timer + // is unref'd so a pending checkout never keeps the process alive. + const timer = Number.isFinite(timeoutMs) + ? setTimeout(() => { + const index = this.waiters.indexOf(waiter); + if (index !== -1) { + this.waiters.splice(index, 1); + } + resolve(false); + }, timeoutMs) + : undefined; + timer?.unref(); + const waiter: Waiter = { + grant: () => { + clearTimeout(timer); + resolve(true); + }, + }; + this.waiters.push(waiter); + }); + } + + /** Return a permit, waking the oldest waiter if any. */ + release(): void { + const next = this.waiters.shift(); + if (next) { + next.grant(); + } else { + this.permits += 1; + } + } +} + +/** + * Bounded checkout/return pool behind a `"pooled"`-scope resource. At most + * `poolSize` instances are checked out at once; released instances go back to + * an idle list for reuse until `maxLifetimeMs` evicts them. Implements the + * cross-SDK pooled-resource semantics. + */ +export class ResourcePool { + private readonly poolSize: number; + private readonly poolMin: number; + private readonly acquireTimeoutMs: number; + private readonly maxLifetimeMs: number; + private readonly semaphore: Semaphore; + private readonly idle: IdleEntry[] = []; + private activeCount = 0; + private totalAcquisitions = 0; + private totalTimeouts = 0; + private shutDown = false; + + constructor( + private readonly name: string, + private readonly factory: () => Promise, + private readonly dispose: ((value: unknown) => void | Promise) | undefined, + options: PoolOptions, + private readonly hooks: PoolHooks = {}, + ) { + this.poolSize = options.poolSize ?? DEFAULT_POOL_SIZE; + this.poolMin = options.poolMin ?? 0; + this.acquireTimeoutMs = options.acquireTimeoutMs ?? DEFAULT_ACQUIRE_TIMEOUT_MS; + this.maxLifetimeMs = options.maxLifetimeMs ?? Number.POSITIVE_INFINITY; + this.semaphore = new Semaphore(this.poolSize); + } + + /** + * Top the idle list up to `poolMin` instances. Best-effort: logs and stops on + * failure. Idempotent, so repeated worker leases don't overfill the pool. + */ + async prewarm(): Promise { + while (this.idle.length + this.activeCount < this.poolMin) { + try { + const instance = await this.buildInstance(); + this.idle.push({ instance, createdAt: Date.now() }); + } catch (error) { + log.warn(() => `failed to prewarm pooled resource "${this.name}"`, error); + break; + } + } + } + + /** + * Check out an instance, waiting up to `acquireTimeoutMs` for capacity. + * Reuses an idle instance when one is fresh enough, otherwise builds one. + */ + async acquire(): Promise { + const granted = await this.semaphore.acquire(this.acquireTimeoutMs); + if (!granted) { + this.totalTimeouts += 1; + throw new ResourceUnavailableError( + `Resource "${this.name}" pool timed out after ${this.acquireTimeoutMs}ms`, + ); + } + + // Reuse the oldest idle instance still within its lifetime; evict expired ones. + let entry = this.idle.shift(); + while (entry) { + if (Date.now() - entry.createdAt < this.maxLifetimeMs) { + this.recordAcquired(); + return entry.instance; + } + await this.disposeInstance(entry.instance); + entry = this.idle.shift(); + } + + // Nothing idle — build fresh. The permit is held across the build and only + // returned on failure, so a rejecting factory never leaks capacity. `active` + // is bumped only after the factory resolves, so it can never overcount. + let instance: unknown; + try { + instance = await this.buildInstance(); + } catch (error) { + this.semaphore.release(); + throw error; + } + this.recordAcquired(); + return instance; + } + + /** Return an instance to the pool (or dispose it if the pool is shut down). */ + async release(instance: unknown): Promise { + this.activeCount -= 1; + if (this.shutDown) { + await this.disposeInstance(instance); + } else { + this.idle.push({ instance, createdAt: Date.now() }); + } + this.semaphore.release(); + } + + /** Dispose every idle instance and route future returns straight to disposal. */ + async shutdown(): Promise { + this.shutDown = true; + const entries = this.idle.splice(0); + for (const entry of entries) { + await this.disposeInstance(entry.instance); + } + } + + /** Point-in-time pool counters. */ + stats(): PoolStats { + return { + size: this.poolSize, + active: this.activeCount, + idle: this.idle.length, + totalAcquisitions: this.totalAcquisitions, + totalTimeouts: this.totalTimeouts, + }; + } + + private async buildInstance(): Promise { + const instance = await this.factory(); + this.hooks.onCreated?.(); + return instance; + } + + private recordAcquired(): void { + this.totalAcquisitions += 1; + this.activeCount += 1; + } + + /** Dispose one instance; errors are logged, never thrown (best-effort cleanup). */ + private async disposeInstance(instance: unknown): Promise { + try { + await this.dispose?.(instance); + } catch (error) { + log.debug(() => `disposing pooled resource "${this.name}" failed`, error); + } + this.hooks.onDisposed?.(); + } +} diff --git a/sdks/node/src/resources/runtime.ts b/sdks/node/src/resources/runtime.ts index 2251a977..051a914e 100644 --- a/sdks/node/src/resources/runtime.ts +++ b/sdks/node/src/resources/runtime.ts @@ -1,9 +1,11 @@ import { ResourceNotFoundError, ResourceScopeError } from "../errors"; +import { ResourcePool } from "./pool"; import type { ResourceContext, ResourceDefinition, ResourceMetrics, ResourceResolver, + ResourceScope, } from "./types"; /** A disposal thunk plus the resource name, for error context. */ @@ -22,9 +24,10 @@ export interface TaskScope { /** * Registry + lifecycle for injectable resources. Worker-scoped values are built - * once and shared; task-scoped values are built per invocation. The worker wires - * this into task execution; tasks reach values via `useResource` or declarative - * `inject`. + * once and shared; task-scoped values are built per invocation; pooled values + * are checked out of a bounded pool per invocation and returned afterwards. The + * worker wires this into task execution; tasks reach values via `useResource` + * or declarative `inject`. * * Every resolver always returns a promise — guard and factory failures surface as * rejections, never synchronous throws — so a failed build is awaited and retried, @@ -35,6 +38,8 @@ export class ResourceRuntime { private readonly workerCache = new Map>(); private readonly workerTeardown: Teardown[] = []; private readonly counters = new Map(); + /** Checkout pools behind pooled-scope resources, built lazily per resource. */ + private readonly pools = new Map(); /** Active worker leases sharing this runtime; teardown disposes only at zero. */ private workerLeases = 0; @@ -44,6 +49,12 @@ export class ResourceRuntime { // Drop any built worker instance so "replace" takes effect; the old // instance is still disposed by its queued teardown. this.workerCache.delete(name); + // Likewise retire the old pool; its idle instances are disposed best-effort. + const stalePool = this.pools.get(name); + if (stalePool) { + this.pools.delete(name); + void stalePool.shutdown(); + } } /** True when nothing is registered — lets the worker skip resource wiring. */ @@ -81,7 +92,7 @@ export class ResourceRuntime { return Promise.reject(unregistered(name)); } if (def.scope !== "worker") { - return Promise.reject(new ResourceScopeError(name)); + return Promise.reject(new ResourceScopeError(name, def.scope)); } const ctx: ResourceContext = { scope: "worker", @@ -104,6 +115,43 @@ export class ResourceRuntime { return built; } + /** + * Resolve a dependency that must be worker-scoped. Pooled instances outlive + * any single task, so their factories may only depend on worker singletons. + */ + private resolveWorkerDependency(name: string, context: ResourceScope): Promise { + const def = this.defs.get(name); + if (!def) { + return Promise.reject(unregistered(name)); + } + if (def.scope !== "worker") { + return Promise.reject(new ResourceScopeError(name, def.scope, context)); + } + return this.resolveWorker(name); + } + + /** Get-or-create the checkout pool behind a pooled-scope resource. */ + private poolFor(name: string, def: ResourceDefinition): ResourcePool { + let pool = this.pools.get(name); + if (!pool) { + const ctx: ResourceContext = { + scope: "pooled", + use: (dep: string) => this.resolveWorkerDependency(dep, "pooled") as Promise, + }; + const counter = this.counter(name); + pool = new ResourcePool(name, () => startFactory(def, ctx), def.dispose, def.pool ?? {}, { + onCreated: () => { + counter.created += 1; + }, + onDisposed: () => { + counter.disposed += 1; + }, + }); + this.pools.set(name, pool); + } + return pool; + } + /** Begin a per-invocation task scope. */ createTaskScope(): TaskScope { const taskCache = new Map>(); @@ -117,6 +165,33 @@ export class ResourceRuntime { if (def.scope === "worker") { return this.resolveWorker(name); // shared singleton, even when first reached here } + if (def.scope === "pooled") { + const pending = taskCache.get(name); + if (pending) { + return pending; // one checkout per task, shared by every resolve + } + const pool = this.poolFor(name, def); + // Cache the pending checkout BEFORE awaiting (see the task branch below) + // so concurrent resolves within the task share a single checkout. + const checkout = pool.acquire().catch((error) => { + taskCache.delete(name); // failed checkout is retryable, not cached + throw error; + }); + taskCache.set(name, checkout); + taskTeardown.push({ + name, + run: async () => { + let value: unknown; + try { + value = await checkout; + } catch { + return; // checkout failed — nothing to return to the pool + } + await pool.release(value); // return, don't dispose: the instance stays pooled + }, + }); + return checkout; + } const cached = taskCache.get(name); if (cached) { return cached; @@ -147,6 +222,13 @@ export class ResourceRuntime { /** Register a worker that shares this runtime's worker-scoped resources. */ acquireWorker(): void { this.workerLeases += 1; + // Best-effort prewarm of pooled resources that asked for warm instances. + // `prewarm` never rejects — build failures are logged inside the pool. + for (const [name, def] of this.defs) { + if (def.scope === "pooled" && (def.pool?.poolMin ?? 0) > 0) { + void this.poolFor(name, def).prewarm(); + } + } } /** @@ -164,6 +246,13 @@ export class ResourceRuntime { } const pending = this.workerTeardown.splice(0); this.workerCache.clear(); + // Shut pools down first — pooled instances may depend on worker resources. + // `shutdown` never throws; per-instance dispose failures are logged. + const pools = [...this.pools.values()]; + this.pools.clear(); + for (const pool of pools) { + await pool.shutdown(); + } await runTeardown(pending); } diff --git a/sdks/node/src/resources/types.ts b/sdks/node/src/resources/types.ts index 446d3c3d..49d68da6 100644 --- a/sdks/node/src/resources/types.ts +++ b/sdks/node/src/resources/types.ts @@ -1,10 +1,10 @@ /** Lifetime of a registered resource. */ -export type ResourceScope = "worker" | "task"; +export type ResourceScope = "worker" | "task" | "pooled"; /** - * Passed to a resource factory so it can depend on other resources. A - * worker-scoped factory may only depend on other worker-scoped resources; a - * task-scoped factory may depend on either. + * Passed to a resource factory so it can depend on other resources. Worker- and + * pooled-scoped factories may only depend on worker-scoped resources (their + * instances outlive any single task); a task-scoped factory may depend on any. */ export interface ResourceContext { /** Scope of the resource currently being built. */ @@ -21,6 +21,34 @@ export interface ResourceDefinition { scope: ResourceScope; /** Tear-down hook, run LIFO when the scope ends (worker stop / job finish). */ dispose?: (value: T) => void | Promise; + /** Pool tuning; only meaningful for `"pooled"` scope. */ + pool?: PoolOptions; +} + +/** Tuning for the bounded checkout pool behind a `"pooled"`-scope resource. */ +export interface PoolOptions { + /** Max instances checked out concurrently. Default 4. */ + poolSize?: number; + /** Instances built eagerly when the worker starts. Default 0 (lazy). */ + poolMin?: number; + /** How long a checkout waits for a free slot before failing. Default 10000. */ + acquireTimeoutMs?: number; + /** Max age of an idle instance before it is disposed and rebuilt. Default unlimited. */ + maxLifetimeMs?: number; +} + +/** Point-in-time counters for one resource pool. */ +export interface PoolStats { + /** Configured capacity (`poolSize`). */ + size: number; + /** Instances currently checked out. */ + active: number; + /** Instances sitting idle, ready for reuse. */ + idle: number; + /** Successful checkouts so far. */ + totalAcquisitions: number; + /** Checkouts that timed out waiting for capacity. */ + totalTimeouts: number; } /** Resolves a resource by name within the current scope. @internal */ diff --git a/sdks/node/test/resources/resources.test.ts b/sdks/node/test/resources/resources.test.ts index c9f011fd..19deb810 100644 --- a/sdks/node/test/resources/resources.test.ts +++ b/sdks/node/test/resources/resources.test.ts @@ -2,7 +2,14 @@ import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; -import { Queue, TaskitoError, useResource, type Worker } from "../../src/index"; +import { + Queue, + ResourceUnavailableError, + TaskitoError, + useResource, + type Worker, +} from "../../src/index"; +import { ResourcePool } from "../../src/resources/pool"; import { ResourceRuntime } from "../../src/resources/runtime"; let worker: Worker | undefined; @@ -166,6 +173,197 @@ describe("ResourceRuntime", () => { }); }); +describe("ResourcePool", () => { + it("returns a released instance to the next acquire", async () => { + let builds = 0; + const pool = new ResourcePool( + "db", + async () => { + builds += 1; + return { id: builds }; + }, + undefined, + {}, + ); + const first = await pool.acquire(); + await pool.release(first); + const second = await pool.acquire(); + expect(second).toBe(first); // same instance came back + expect(builds).toBe(1); + }); + + it("times out with ResourceUnavailableError when the pool is exhausted", async () => { + const pool = new ResourcePool("db", async () => ({}), undefined, { + poolSize: 1, + acquireTimeoutMs: 30, + }); + await pool.acquire(); // takes the only slot + await expect(pool.acquire()).rejects.toThrow(ResourceUnavailableError); + await expect(pool.acquire()).rejects.toThrow(/Resource "db" pool timed out/); + expect(pool.stats().totalTimeouts).toBe(2); + }); + + it("does not leak capacity when the factory rejects", async () => { + let attempts = 0; + const pool = new ResourcePool( + "db", + async () => { + attempts += 1; + if (attempts === 1) { + throw new Error("connect failed"); + } + return "ok"; + }, + undefined, + { poolSize: 1, acquireTimeoutMs: 50 }, + ); + await expect(pool.acquire()).rejects.toThrow("connect failed"); + expect(pool.stats().active).toBe(0); // never counted as checked out + expect(await pool.acquire()).toBe("ok"); // permit was returned, not leaked + }); + + it("prewarm builds poolMin instances eagerly", async () => { + let builds = 0; + const pool = new ResourcePool( + "db", + async () => { + builds += 1; + return builds; + }, + undefined, + { poolMin: 2 }, + ); + await pool.prewarm(); + expect(builds).toBe(2); + expect(pool.stats().idle).toBe(2); + }); + + it("evicts an idle instance past maxLifetime and builds a fresh one", async () => { + let builds = 0; + const disposed: number[] = []; + const pool = new ResourcePool( + "db", + async () => { + builds += 1; + return builds; + }, + (value) => void disposed.push(value as number), + { maxLifetimeMs: 20 }, + ); + const first = await pool.acquire(); + await pool.release(first); + await new Promise((resolve) => setTimeout(resolve, 40)); // let it expire + const second = await pool.acquire(); + expect(second).toBe(2); // freshly built + expect(disposed).toEqual([1]); // expired instance was disposed + }); + + it("stats reflect active, idle, acquisitions, and timeouts", async () => { + const pool = new ResourcePool("db", async () => ({}), undefined, { + poolSize: 2, + acquireTimeoutMs: 20, + }); + const a = await pool.acquire(); + const b = await pool.acquire(); + await expect(pool.acquire()).rejects.toThrow(ResourceUnavailableError); + expect(pool.stats()).toEqual({ + size: 2, + active: 2, + idle: 0, + totalAcquisitions: 2, + totalTimeouts: 1, + }); + await pool.release(a); + expect(pool.stats().active).toBe(1); + expect(pool.stats().idle).toBe(1); + await pool.release(b); + expect(pool.stats().active).toBe(0); + expect(pool.stats().idle).toBe(2); + }); +}); + +describe("pooled resources in the runtime", () => { + it("checks out one pooled instance per task and returns it on teardown", async () => { + const rt = new ResourceRuntime(); + let builds = 0; + rt.register("conn", { + scope: "pooled", + factory: () => { + builds += 1; + return { id: builds }; + }, + pool: { poolSize: 1 }, + }); + const a = rt.createTaskScope(); + const first = await a.resolver("conn"); + expect(await a.resolver("conn")).toBe(first); // one checkout per task + await a.teardown(); // released back to the pool, not disposed + const b = rt.createTaskScope(); + expect(await b.resolver("conn")).toBe(first); // reused across tasks + expect(builds).toBe(1); + expect(rt.metrics().conn).toEqual({ created: 1, disposed: 0, active: 1 }); + }); + + it("rejects a pooled factory depending on a task-scoped resource", async () => { + const rt = new ResourceRuntime(); + rt.register("perJob", { scope: "task", factory: () => "x" }); + rt.register("conn", { + scope: "pooled", + factory: (ctx) => ctx.use("perJob"), + }); + const scope = rt.createTaskScope(); + await expect(scope.resolver("conn")).rejects.toThrow( + /task-scoped and cannot be resolved at pooled scope/, + ); + }); + + it("lets a pooled factory depend on a worker-scoped resource", async () => { + const rt = new ResourceRuntime(); + rt.register("base", { scope: "worker", factory: () => 10 }); + rt.register("conn", { + scope: "pooled", + factory: async (ctx) => (await ctx.use("base")) * 2, + }); + const scope = rt.createTaskScope(); + expect(await scope.resolver("conn")).toBe(20); + }); + + it("prewarms pooled instances when a worker acquires the runtime", async () => { + const rt = new ResourceRuntime(); + let builds = 0; + rt.register("conn", { + scope: "pooled", + factory: () => { + builds += 1; + return builds; + }, + pool: { poolMin: 2 }, + }); + rt.acquireWorker(); + expect(await waitFor(() => builds === 2)).toBe(true); + await rt.teardownWorker(); + }); + + it("disposes idle pooled instances at worker teardown", async () => { + const rt = new ResourceRuntime(); + let disposals = 0; + rt.register("conn", { + scope: "pooled", + factory: () => ({}), + dispose: () => { + disposals += 1; + }, + }); + rt.acquireWorker(); + const scope = rt.createTaskScope(); + await scope.resolver("conn"); + await scope.teardown(); + await rt.teardownWorker(); + expect(disposals).toBe(1); + expect(rt.metrics().conn).toEqual({ created: 1, disposed: 1, active: 0 }); + }); +}); + describe("resource injection in a worker", () => { it("resolves a worker-scoped resource via useResource(), shared across jobs", async () => { const queue = newQueue(); @@ -250,6 +448,84 @@ describe("resource injection in a worker", () => { expect(builds).toBe(2); // one per invocation }); + it("reuses one pooled instance across jobs through a worker", async () => { + const queue = newQueue(); + let builds = 0; + const seen: number[] = []; + queue.resource( + "conn", + () => { + builds += 1; + return { id: builds }; + }, + { scope: "pooled", pool: { poolSize: 1 } }, + ); + queue.task("hit", async () => { + const conn = await useResource<{ id: number }>("conn"); + seen.push(conn.id); + }); + + queue.enqueue("hit"); + queue.enqueue("hit"); + worker = queue.runWorker(); + + expect(await waitFor(() => seen.length === 2)).toBe(true); + expect(seen).toEqual([1, 1]); // same instance both times + expect(builds).toBe(1); + }); + + it("disposes pooled instances when the worker stops", async () => { + const queue = newQueue(); + let disposals = 0; + let done = false; + queue.resource("conn", () => ({}), { + scope: "pooled", + dispose: () => { + disposals += 1; + }, + pool: { poolSize: 1 }, + }); + queue.task("touch", async () => { + await useResource("conn"); + done = true; + }); + + queue.enqueue("touch"); + worker = queue.runWorker(); + expect(await waitFor(() => done)).toBe(true); + + worker.stop(); + worker = undefined; + expect(await waitFor(() => disposals === 1)).toBe(true); + }); + + it("fails a job whose pooled factory uses a task-scoped resource", async () => { + const queue = newQueue(); + queue.resource("perJob", () => "x", { scope: "task" }); + queue.resource("conn", (ctx) => ctx.use("perJob"), { scope: "pooled" }); + let failure = ""; + queue.task("guarded", async () => { + try { + await useResource("conn"); + } catch (error) { + failure = (error as Error).message; + } + }); + + queue.enqueue("guarded"); + worker = queue.runWorker(); + + expect(await waitFor(() => failure !== "")).toBe(true); + expect(failure).toMatch(/task-scoped and cannot be resolved at pooled scope/); + }); + + it("rejects pool options on a non-pooled scope", () => { + const queue = newQueue(); + expect(() => queue.resource("bad", () => 1, { pool: { poolSize: 2 } })).toThrow( + /pool options require scope "pooled"/, + ); + }); + it("throws when useResource is called outside a task", () => { expect(() => useResource("anything")).toThrow(TaskitoError); }); From aa8cc2ad3b31fa470bf6893a0aa8b7d6c3f59be3 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:58:59 +0530 Subject: [PATCH 2/2] docs(node): document pooled resource scope --- .../guides/resources/dependency-injection.mdx | 35 +++++++++++++++++-- .../docs/node/guides/resources/index.mdx | 2 +- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/docs/content/docs/node/guides/resources/dependency-injection.mdx b/docs/content/docs/node/guides/resources/dependency-injection.mdx index cc845d7d..80499680 100644 --- a/docs/content/docs/node/guides/resources/dependency-injection.mdx +++ b/docs/content/docs/node/guides/resources/dependency-injection.mdx @@ -35,6 +35,7 @@ A resource's scope decides its lifetime: |---|---|---|---| | `worker` (default) | Lazily, on first use | The worker — a shared singleton | Connection pools, HTTP clients, SDK clients | | `task` | Once per job invocation | That job — disposed when it finishes | Per-job transactions, request-scoped clients | +| `pooled` | Lazily, up to `poolSize` at once | Checked out per job, returned when it finishes | Expensive clients that must stay bounded but be reused | ```ts queue.resource("tx", async () => db.begin(), { @@ -47,6 +48,34 @@ Worker-scoped resources are built once even under concurrency — concurrent initialization is de-duplicated. A factory that throws is **not** cached, so the next job retries it. +## Pooled resources + +A `pooled` resource sits between the other two scopes: instances are reused like +worker resources but bound to a single job at a time like task resources, with a +hard cap on how many exist at once. Each job checks out one instance — reusing +an idle one, building a new one only when none is free — and returns it to the +pool when the job finishes. A job that cannot get an instance within +`acquireTimeoutMs` fails with `ResourceUnavailableError`: + +```ts +queue.resource("ftp", () => connectFtp(), { + scope: "pooled", + dispose: (conn) => conn.close(), + pool: { poolSize: 4, poolMin: 1, acquireTimeoutMs: 10_000, maxLifetimeMs: 300_000 }, +}); +``` + +| Option | Default | Meaning | +|---|---|---| +| `poolSize` | `4` | Max instances checked out concurrently. Jobs wait when exhausted. | +| `poolMin` | `0` | Instances pre-built when the worker starts. `0` means lazy. | +| `acquireTimeoutMs` | `10000` | How long a checkout waits before failing the job. | +| `maxLifetimeMs` | unlimited | Idle instances older than this are disposed and rebuilt. | + +Pooled instances outlive any single job, so a pooled factory may only depend on +worker-scoped resources. `dispose` runs when an instance is evicted or when the +worker stops — not when a job returns it to the pool. + ## Injecting resources Two equivalent ways to reach a resource from a handler. @@ -104,9 +133,9 @@ queue.resource("db", async (ctx) => { }); ``` -A worker-scoped factory may only depend on other worker-scoped resources; -reaching for a task-scoped one throws (it has no job to bind to). Task-scoped -factories may depend on either. +Worker-scoped and pooled factories may only depend on worker-scoped resources; +reaching for anything shorter-lived throws (it has no job to bind to). +Task-scoped factories may depend on any scope. ## Teardown diff --git a/docs/content/docs/node/guides/resources/index.mdx b/docs/content/docs/node/guides/resources/index.mdx index 89cb202f..41176e46 100644 --- a/docs/content/docs/node/guides/resources/index.mdx +++ b/docs/content/docs/node/guides/resources/index.mdx @@ -23,7 +23,7 @@ queue.task("sync", async (id: string) => { | Page | What it covers | |---|---| -| [Dependency injection](/node/guides/resources/dependency-injection) | `queue.resource()`, worker/task scopes, `useResource`, declarative `inject`, teardown | +| [Dependency injection](/node/guides/resources/dependency-injection) | `queue.resource()`, worker/task/pooled scopes, `useResource`, declarative `inject`, teardown | | [Enqueue interception](/node/guides/resources/interception) | The `onEnqueue` hook — validate, redact, and rewrite jobs before serialization |