feat(node): pooled resource scope

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
 

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 |
 
 <Callout type="info">

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. */

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,

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<TTasks extends TaskMap = TaskMap> {
   /**
    * 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<T>(
     name: string,
     factory: (ctx: ResourceContext) => T | Promise<T>,
-    options?: { scope?: ResourceScope; dispose?: (value: T) => void | Promise<void> },
+    options?: {
+      scope?: ResourceScope;
+      dispose?: (value: T) => void | Promise<void>;
+      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<T>(name, {
       factory,
-      scope: options?.scope ?? "worker",
+      scope,
       dispose: options?.dispose,
+      pool: options?.pool,
     });
     return this;
   }

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,

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<boolean> {
+    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<unknown>,
+    private readonly dispose: ((value: unknown) => void | Promise<void>) | 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<void> {
+    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<unknown> {
+    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<void> {
+    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<void> {
+    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<unknown> {
+    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<void> {
+    try {
+      await this.dispose?.(instance);
+    } catch (error) {
+      log.debug(() => `disposing pooled resource "${this.name}" failed`, error);
+    }
+    this.hooks.onDisposed?.();
+  }
+}