From cc26b0759816d6c0cf495afce3fa21dc59788637 Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:22:16 +0530
Subject: [PATCH 1/6] feat(node): producer-side batch accumulator

Buffers enqueues for one task and flushes them through enqueueMany on a size
or time trigger, matching Java's Batcher and Python's BatchAccumulator.
---
 sdks/node/src/batching.ts | 191 ++++++++++++++++++++++++++++++++++++++
 sdks/node/src/index.ts    |   1 +
 sdks/node/src/queue.ts    |  13 +++
 3 files changed, 205 insertions(+)
 create mode 100644 sdks/node/src/batching.ts

diff --git a/sdks/node/src/batching.ts b/sdks/node/src/batching.ts
new file mode 100644
index 00000000..df00781a
--- /dev/null
+++ b/sdks/node/src/batching.ts
@@ -0,0 +1,191 @@
+// Producer-side batching: buffer enqueues for one task and send them as a
+// single `enqueueMany` round-trip. The counterpart of Python's
+// `BatchAccumulator` and Java's `Batcher<T>`.
+
+import { QueueError } from "./errors";
+import type { Queue } from "./queue";
+import type { EnqueueOptions, TaskMap } from "./types";
+import { createLogger } from "./utils";
+
+const DEFAULT_MAX_SIZE = 100;
+const DEFAULT_MAX_WAIT_MS = 500;
+const log = createLogger("batching");
+
+/** Construction options for a {@link Batcher}. */
+export interface BatcherOptions {
+  /** Flush as soon as this many entries have accumulated. Default 100. */
+  maxSize?: number;
+  /**
+   * Flush this long after the first buffered entry arrived, even if `maxSize`
+   * isn't reached — the worst-case latency any single entry can see. Default 500.
+   */
+  maxWaitMs?: number;
+  /**
+   * Called when a timer-driven flush throws. A timed flush has no caller to
+   * raise to, so without this the failure is only logged. The entries stay
+   * buffered either way and are retried on the next window.
+   */
+  onError?: (error: unknown) => void;
+}
+
+/** One buffered enqueue: the same `{ args, options }` shape `enqueueMany` takes. */
+interface BufferedEntry<Args extends unknown[]> {
+  args?: Args;
+  options?: EnqueueOptions;
+}
+
+/**
+ * Buffers enqueues for one task and flushes them through
+ * {@link Queue.enqueueMany} once the buffer reaches `maxSize` or `maxWaitMs`
+ * elapses since the first buffered entry — producer-side batching, to cut
+ * storage round-trips when many small jobs share a task.
+ *
+ * ```ts
+ * using batcher = queue.batcher("sendEmail", { maxSize: 100, maxWaitMs: 500 });
+ * for (const email of emails) {
+ *   batcher.add([email]);
+ * }
+ * // block exit flushes the remainder
+ * ```
+ *
+ * The flush timer is `unref`'d, so a partially filled buffer never keeps the
+ * process alive — and is lost if the process exits without {@link Batcher.close}
+ * (or `using`). This matches Java's daemon scheduler and Python's atexit-only
+ * flush.
+ *
+ * Distinct from worker-side dequeue batching (`batchSize` on the worker), which
+ * controls how many jobs one poll claims.
+ */
+export class Batcher<
+  TTasks extends TaskMap = TaskMap,
+  Name extends keyof TTasks & string = keyof TTasks & string,
+> {
+  private buffer: BufferedEntry<Parameters<TTasks[Name]>>[] = [];
+  private readonly maxSize: number;
+  private readonly maxWaitMs: number;
+  private readonly onError: ((error: unknown) => void) | undefined;
+  private timer?: ReturnType<typeof setTimeout>;
+  private isClosed = false;
+
+  constructor(
+    private readonly queue: Queue<TTasks>,
+    readonly name: Name,
+    options: BatcherOptions = {},
+  ) {
+    const maxSize = options.maxSize ?? DEFAULT_MAX_SIZE;
+    if (!Number.isInteger(maxSize) || maxSize < 1) {
+      throw new RangeError(`Batcher maxSize must be a positive integer, got ${maxSize}`);
+    }
+    const maxWaitMs = options.maxWaitMs ?? DEFAULT_MAX_WAIT_MS;
+    if (!Number.isFinite(maxWaitMs) || maxWaitMs <= 0) {
+      throw new RangeError(`Batcher maxWaitMs must be a positive finite number, got ${maxWaitMs}`);
+    }
+    this.maxSize = maxSize;
+    this.maxWaitMs = maxWaitMs;
+    this.onError = options.onError;
+  }
+
+  /** How many entries are buffered right now. */
+  get size(): number {
+    return this.buffer.length;
+  }
+
+  /** Whether {@link Batcher.close} has run — `add` throws once closed. */
+  get closed(): boolean {
+    return this.isClosed;
+  }
+
+  /**
+   * Buffer one enqueue, typed exactly like {@link Queue.enqueue}. Returns the
+   * job ids if this call filled the buffer to `maxSize` (flushing immediately),
+   * otherwise an empty array — the timed flush is still pending.
+   */
+  add(args?: Parameters<TTasks[Name]>, options?: EnqueueOptions): string[] {
+    if (this.isClosed) {
+      throw new QueueError(`batcher for task "${this.name}" is closed`);
+    }
+    this.buffer.push({ args, options });
+    if (this.buffer.length >= this.maxSize) {
+      return this.flush();
+    }
+    this.arm();
+    return [];
+  }
+
+  /**
+   * Enqueue whatever is buffered now, cancelling any pending timed flush.
+   * Returns the new job ids, or an empty array if the buffer was empty. Stays
+   * callable after {@link Batcher.close} so a failed final flush can be retried.
+   */
+  flush(): string[] {
+    this.disarm();
+    if (this.buffer.length === 0) {
+      return [];
+    }
+    // Take the entries out before enqueueing: `job.enqueued` handlers run
+    // synchronously inside `enqueueMany` and may re-enter `add`, so the buffer
+    // must already be free of what's in flight.
+    const entries = this.buffer.splice(0, this.buffer.length);
+    try {
+      return this.queue.enqueueMany(this.name, entries);
+    } catch (error) {
+      // A failed flush keeps its payloads rather than silently dropping them —
+      // ahead of anything buffered meanwhile, so submission order holds.
+      this.buffer = entries.concat(this.buffer);
+      throw error;
+    }
+  }
+
+  /**
+   * Flush the remainder and stop the timer. Returns the final job ids. Idempotent;
+   * rethrows if that last flush fails, leaving the entries buffered for a manual
+   * {@link Batcher.flush} retry.
+   */
+  close(): string[] {
+    if (this.isClosed) {
+      return [];
+    }
+    this.isClosed = true;
+    return this.flush();
+  }
+
+  [Symbol.dispose](): void {
+    this.close();
+  }
+
+  /** Start the timed flush, unless one is already scheduled. */
+  private arm(): void {
+    if (this.timer) {
+      return;
+    }
+    this.timer = setTimeout(() => {
+      this.timer = undefined;
+      this.flushOnTimer();
+    }, this.maxWaitMs);
+    this.timer.unref();
+  }
+
+  private disarm(): void {
+    if (this.timer) {
+      clearTimeout(this.timer);
+      this.timer = undefined;
+    }
+  }
+
+  private flushOnTimer(): void {
+    try {
+      this.flush();
+    } catch (error) {
+      if (this.onError) {
+        this.onError(error);
+      } else {
+        log.warn(() => `batched flush for task "${this.name}" failed`, error);
+      }
+      // The entries are still buffered — re-arm so a transient failure doesn't
+      // strand them until the next `add` or an explicit flush.
+      if (!this.isClosed && this.buffer.length > 0) {
+        this.arm();
+      }
+    }
+  }
+}
diff --git a/sdks/node/src/index.ts b/sdks/node/src/index.ts
index 53be848a..efcf2fce 100644
--- a/sdks/node/src/index.ts
+++ b/sdks/node/src/index.ts
@@ -10,6 +10,7 @@ export {
   serveAutoscaler,
   WorkerProcessManager,
 } from "./autoscale";
+export { Batcher, type BatcherOptions } from "./batching";
 export { currentJob, type JobContext } from "./context";
 export {
   AuthStore,
diff --git a/sdks/node/src/queue.ts b/sdks/node/src/queue.ts
index 449a226b..a44a8d8c 100644
--- a/sdks/node/src/queue.ts
+++ b/sdks/node/src/queue.ts
@@ -1,5 +1,6 @@
 import { mkdirSync } from "node:fs";
 import { dirname } from "node:path";
+import { Batcher, type BatcherOptions } from "./batching";
 import {
   MiddlewareDisableStore,
   middlewareKey,
@@ -646,6 +647,18 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     return jobIds;
   }
 
+  /**
+   * A producer-side accumulator for `name`: buffers enqueues and flushes them
+   * through {@link Queue.enqueueMany} on a size or time trigger. Close it (or
+   * declare it with `using`) so the remainder isn't lost at shutdown.
+   */
+  batcher<Name extends keyof TTasks & string>(
+    name: Name,
+    options?: BatcherOptions,
+  ): Batcher<TTasks, Name> {
+    return new Batcher(this, name, options);
+  }
+
   // ── Topic pub/sub ─────────────────────────────────────────────────
 
   /**

From 5678e7d771f0df28d73e55c152d648bfb3b72ac4 Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:22:20 +0530
Subject: [PATCH 2/6] test(node): cover Batcher flush, failure, and re-entrancy

---
 sdks/node/test/core/batcher.test.ts | 191 ++++++++++++++++++++++++++++
 1 file changed, 191 insertions(+)
 create mode 100644 sdks/node/test/core/batcher.test.ts

diff --git a/sdks/node/test/core/batcher.test.ts b/sdks/node/test/core/batcher.test.ts
new file mode 100644
index 00000000..7cf9df06
--- /dev/null
+++ b/sdks/node/test/core/batcher.test.ts
@@ -0,0 +1,191 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { expect, it, vi } from "vitest";
+import { Batcher, Queue, QueueError } from "../../src/index";
+
+function newQueue() {
+  return new Queue({ dbPath: join(mkdtempSync(join(tmpdir(), "taskito-batcher-")), "q.db") }).task(
+    "collect",
+    (n: number) => n,
+  );
+}
+
+async function waitFor(predicate: () => Promise<boolean>, timeoutMs = 4000): Promise<boolean> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (await predicate()) {
+      return true;
+    }
+    await new Promise((r) => setTimeout(r, 20));
+  }
+  return false;
+}
+
+it("flushes once the buffer reaches maxSize", async () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 3, maxWaitMs: 60_000 });
+
+  expect(batcher.add([1])).toEqual([]);
+  expect(batcher.add([2])).toEqual([]);
+  expect(batcher.size).toBe(2);
+  expect((await queue.stats()).pending).toBe(0); // nothing enqueued yet
+
+  const ids = batcher.add([3]);
+  expect(ids).toHaveLength(3);
+  expect(batcher.size).toBe(0);
+  expect((await queue.stats()).pending).toBe(3);
+});
+
+it("flushes after maxWaitMs without reaching maxSize", async () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 100, maxWaitMs: 50 });
+
+  batcher.add([1]);
+  batcher.add([2]);
+  expect(await waitFor(async () => (await queue.stats()).pending === 2)).toBe(true);
+  expect(batcher.size).toBe(0);
+  batcher.close();
+});
+
+it("close flushes the remainder and blocks further adds", async () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 100, maxWaitMs: 60_000 });
+
+  batcher.add([1]);
+  batcher.add([2]);
+  expect(batcher.close()).toHaveLength(2);
+  expect((await queue.stats()).pending).toBe(2);
+  expect(batcher.closed).toBe(true);
+  expect(() => batcher.add([3])).toThrow(QueueError);
+
+  expect(batcher.close()).toEqual([]); // idempotent
+});
+
+it("flushes via Symbol.dispose", async () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 100, maxWaitMs: 60_000 });
+  batcher.add([1]);
+  expect((await queue.stats()).pending).toBe(0);
+
+  batcher[Symbol.dispose]();
+  expect((await queue.stats()).pending).toBe(1);
+  expect(batcher.closed).toBe(true);
+});
+
+it("passes per-entry args and options through to enqueueMany", () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 2, maxWaitMs: 60_000 });
+
+  batcher.add([1], { queue: "q0" });
+  const ids = batcher.add([2], { queue: "q1" });
+  expect(ids.map((id) => queue.getJob(id)?.queue)).toEqual(["q0", "q1"]);
+});
+
+it("batched jobs run like any other", async () => {
+  const seen: number[] = [];
+  const queue = newQueue().task("record", (n: number) => {
+    seen.push(n);
+  });
+
+  const batcher = queue.batcher("record", { maxSize: 3, maxWaitMs: 60_000 });
+  batcher.add([10]);
+  batcher.add([20]);
+  batcher.add([30]);
+
+  const worker = queue.runWorker();
+  try {
+    expect(await waitFor(async () => seen.length === 3)).toBe(true);
+    expect([...seen].sort((a, b) => a - b)).toEqual([10, 20, 30]);
+  } finally {
+    worker.stop();
+  }
+});
+
+it("keeps entries buffered when a flush throws", () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 2, maxWaitMs: 60_000 });
+  const enqueueMany = vi
+    .spyOn(queue, "enqueueMany")
+    .mockImplementationOnce(() => {
+      throw new Error("storage down");
+    })
+    .mockImplementationOnce(() => ["job-1", "job-2"]);
+
+  batcher.add([1]);
+  expect(() => batcher.add([2])).toThrow("storage down");
+  expect(batcher.size).toBe(2); // nothing dropped
+
+  expect(batcher.flush()).toEqual(["job-1", "job-2"]);
+  expect(batcher.size).toBe(0);
+  enqueueMany.mockRestore();
+});
+
+it("reports a failed timed flush to onError and retries it", async () => {
+  const queue = newQueue();
+  const errors: unknown[] = [];
+  const batcher = queue.batcher("collect", {
+    maxSize: 100,
+    maxWaitMs: 30,
+    onError: (error) => errors.push(error),
+  });
+  const enqueueMany = vi.spyOn(queue, "enqueueMany").mockImplementationOnce(() => {
+    throw new Error("storage down");
+  });
+
+  batcher.add([1]);
+  expect(await waitFor(async () => errors.length > 0)).toBe(true);
+  expect((errors[0] as Error).message).toBe("storage down");
+
+  // The retry uses the real enqueueMany, so the entry lands rather than stranding.
+  expect(await waitFor(async () => (await queue.stats()).pending === 1)).toBe(true);
+  expect(batcher.size).toBe(0);
+  enqueueMany.mockRestore();
+  batcher.close();
+});
+
+it("rejects invalid tunables", () => {
+  const queue = newQueue();
+  expect(() => queue.batcher("collect", { maxSize: 0 })).toThrow(RangeError);
+  expect(() => queue.batcher("collect", { maxSize: 1.5 })).toThrow(RangeError);
+  expect(() => queue.batcher("collect", { maxWaitMs: 0 })).toThrow(RangeError);
+  expect(() => queue.batcher("collect", { maxWaitMs: Number.NaN })).toThrow(RangeError);
+});
+
+it("types add() from the registered task", () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 100 });
+
+  // @ts-expect-error wrong argument types for the batched task
+  batcher.add(["not a number"]);
+  expect(batcher.size).toBe(1);
+  batcher.close();
+});
+
+it("can be constructed directly against a queue", async () => {
+  const queue = newQueue();
+  const batcher = new Batcher(queue, "collect", { maxSize: 1 });
+  expect(batcher.name).toBe("collect");
+  expect(batcher.add([1])).toHaveLength(1);
+  expect((await queue.stats()).pending).toBe(1);
+});
+
+it("stays consistent when a job.enqueued handler re-enters add", async () => {
+  const queue = newQueue();
+  const batcher = queue.batcher("collect", { maxSize: 2, maxWaitMs: 60_000 });
+  let reentered = false;
+  queue.on("job.enqueued", () => {
+    if (!reentered) {
+      reentered = true;
+      batcher.add([99]); // fires while the first flush is still in enqueueMany
+    }
+  });
+
+  batcher.add([1]);
+  expect(batcher.add([2])).toHaveLength(2);
+  expect(batcher.size).toBe(1); // only the re-entrant entry is left buffered
+  expect((await queue.stats()).pending).toBe(2); // and it wasn't enqueued twice
+
+  expect(batcher.close()).toHaveLength(1);
+  expect((await queue.stats()).pending).toBe(3);
+});

From 568a32118376e2e8c3e96ff5dba1b2057556b236 Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:22:20 +0530
Subject: [PATCH 3/6] docs(node): batching API reference and core guide

Also documents enqueueMany, which Node shipped undocumented.
---
 .../docs/node/api-reference/batching.mdx      | 126 +++++++++++++++
 .../content/docs/node/api-reference/index.mdx |   1 +
 .../content/docs/node/api-reference/meta.json |   2 +-
 .../docs/node/guides/core/batching.mdx        | 147 ++++++++++++++++++
 docs/content/docs/node/guides/core/index.mdx  |   5 +
 docs/content/docs/node/guides/core/meta.json  |   2 +-
 .../content/docs/shared/guides/core/tasks.mdx |   3 +-
 7 files changed, 283 insertions(+), 3 deletions(-)
 create mode 100644 docs/content/docs/node/api-reference/batching.mdx
 create mode 100644 docs/content/docs/node/guides/core/batching.mdx

diff --git a/docs/content/docs/node/api-reference/batching.mdx b/docs/content/docs/node/api-reference/batching.mdx
new file mode 100644
index 00000000..83eaedf0
--- /dev/null
+++ b/docs/content/docs/node/api-reference/batching.mdx
@@ -0,0 +1,126 @@
+---
+title: Batching
+description: "enqueueMany and Batcher — send many enqueues in one storage round-trip."
+---
+
+Two levels of producer-side batching: `enqueueMany` sends a list you already
+have, and `Batcher` accumulates one-at-a-time enqueues and sends them for you.
+
+<Callout type="info">
+  This is producer-side batching. It is unrelated to the worker's `batchSize`,
+  which controls how many jobs one scheduler poll claims — see
+  [Worker](/node/api-reference/worker).
+</Callout>
+
+## `enqueueMany`
+
+```ts
+queue.enqueueMany(
+  name: string,
+  jobs: ReadonlyArray<{ args?: unknown[]; options?: EnqueueOptions }>,
+): string[];
+```
+
+Inserts every job in one storage call and returns the ids in input order. Each
+entry carries its own typed `args` and its own
+[`EnqueueOptions`](/node/guides/core/enqueue-options), so a batch can mix
+queues, priorities, and delays.
+
+```ts
+const ids = queue.enqueueMany("sendEmail", [
+  { args: [{ to: "a@example.com" }] },
+  { args: [{ to: "b@example.com" }], options: { priority: 5 } },
+]);
+```
+
+An entry with a `uniqueKey` dedups exactly like `enqueue`: a key that already
+has a pending or running job yields that job's id instead of a new row, so the
+returned array always has one id per input entry. `job.enqueued` fires once per
+entry, deduped ones included.
+
+The batch is admitted or rejected as a whole: if a target queue has a
+`maxPending` cap and the batch would exceed it, the call throws `QueueFullError`
+and nothing is inserted. An enqueue gate that `skip`s a single entry throws
+`EnqueueSkippedError` — a batch is one all-or-nothing native call, so dropping
+one row isn't expressible. `defer` is fine, since each entry carries its own
+options.
+
+## `Batcher`
+
+```ts
+queue.batcher(name: string, options?: BatcherOptions): Batcher;
+new Batcher(queue, name: string, options?: BatcherOptions);
+```
+
+Buffers enqueues for one task and flushes them through `enqueueMany` once the
+buffer reaches `maxSize` or `maxWaitMs` elapses since the first buffered entry.
+Use it when work arrives one item at a time — a request handler, a stream — and
+you want one round-trip per batch instead of per item. The Node counterpart of
+Python's `BatchAccumulator` and Java's `Batcher<T>`.
+
+| Option | Default | Description |
+|---|---|---|
+| `maxSize` | `100` | Flush as soon as this many entries have accumulated. Must be a positive integer. |
+| `maxWaitMs` | `500` | Flush this long after the first buffered entry, even if `maxSize` isn't reached — the worst-case latency of any one entry. Must be positive. |
+| `onError` | — | Called when a *timed* flush throws. Without it the failure is only logged. |
+
+```ts
+using batcher = queue.batcher("sendEmail", { maxSize: 100, maxWaitMs: 500 });
+
+for await (const email of stream) {
+  batcher.add([email]);
+}
+// block exit flushes whatever is left
+```
+
+### `add`
+
+```ts
+add(args?: unknown[], options?: EnqueueOptions): string[];
+```
+
+Buffers one enqueue, typed exactly like `enqueue`. Returns the job ids if this
+call filled the buffer to `maxSize` (flushing immediately), otherwise an empty
+array — the timed flush is still pending. Throws `QueueError` once the batcher
+is closed.
+
+### `flush`
+
+```ts
+flush(): string[];
+```
+
+Enqueues whatever is buffered right now, cancelling the pending timed flush.
+Returns the new job ids, or an empty array if the buffer was empty.
+
+### `close`
+
+```ts
+close(): string[];
+```
+
+Flushes the remainder and stops the timer. Idempotent, and also exposed as
+`Symbol.dispose` so `using` closes it at block exit.
+
+### `size` / `closed`
+
+`size` is how many entries are buffered right now; `closed` reports whether
+`close` has run.
+
+### Failure handling
+
+A failed flush keeps its entries — they go back at the head of the buffer, ahead
+of anything added meanwhile — rather than dropping them:
+
+- A flush triggered by `add` (or an explicit `flush`) throws to the caller, and
+  the entries stay buffered for a retry.
+- A timed flush has no caller, so it reports to `onError` (or logs) and re-arms
+  the timer, retrying on the next window.
+- `close` rethrows a failed final flush; the entries remain in `size` and
+  `flush` still works, so the shutdown path can retry.
+
+<Callout type="warning">
+  The flush timer is `unref`'d — a partially filled buffer never keeps the
+  process alive, and is lost if the process exits without `close` (or `using`).
+  Close your batchers on shutdown.
+</Callout>
diff --git a/docs/content/docs/node/api-reference/index.mdx b/docs/content/docs/node/api-reference/index.mdx
index 3ac3ecba..b2e6a24a 100644
--- a/docs/content/docs/node/api-reference/index.mdx
+++ b/docs/content/docs/node/api-reference/index.mdx
@@ -15,6 +15,7 @@ The public API exported from the `taskito` barrel. Contrib helpers live under
 | [Context](/node/api-reference/context) | `currentJob()` — signal, progress, ids. |
 | [Serializers](/node/api-reference/serializers) | `JsonSerializer`, `MsgpackSerializer`, `CborSerializer`, the `Serializer` interface. |
 | [Workflows](/node/api-reference/workflows) | The workflow builder + run handle. |
+| [Batching](/node/api-reference/batching) | `enqueueMany` and `Batcher` — many enqueues in one round-trip. |
 | [Testing](/node/api-reference/testing) | `mockResource` — stub an injected resource in tests. |
 | [Errors](/node/api-reference/errors) | The `TaskitoError` hierarchy. |
 | [CLI](/node/api-reference/cli) | The `taskito` command. |
diff --git a/docs/content/docs/node/api-reference/meta.json b/docs/content/docs/node/api-reference/meta.json
index 9a42889d..83c18fe8 100644
--- a/docs/content/docs/node/api-reference/meta.json
+++ b/docs/content/docs/node/api-reference/meta.json
@@ -1 +1 @@
-{ "title": "API Reference", "root": true, "pages": ["index", "queue", "task", "worker", "result", "context", "serializers", "workflows", "testing", "errors", "cli"] }
+{ "title": "API Reference", "root": true, "pages": ["index", "queue", "task", "worker", "result", "context", "serializers", "workflows", "batching", "testing", "errors", "cli"] }
diff --git a/docs/content/docs/node/guides/core/batching.mdx b/docs/content/docs/node/guides/core/batching.mdx
new file mode 100644
index 00000000..c0914c91
--- /dev/null
+++ b/docs/content/docs/node/guides/core/batching.mdx
@@ -0,0 +1,147 @@
+---
+title: Batching
+description: "Two unrelated batching mechanisms: producer-side Batcher/enqueueMany and worker-side batchSize."
+---
+
+"Batch" means two different things in the Node SDK, and they don't compose with
+each other:
+
+- **Producer batching** — `Batcher` and `enqueueMany` group many enqueues into
+  fewer storage writes, before any job exists.
+- **Dequeue batching** — `batchSize` on `runWorker` controls how many
+  already-enqueued jobs the scheduler claims per poll.
+
+This guide covers both and where each one applies.
+
+## Bulk enqueue with `enqueueMany`
+
+`enqueueMany` inserts many jobs for one task in a single storage call and
+returns their ids in input order. Every entry carries its own `args` and its own
+[enqueue options](/node/guides/core/enqueue-options), so one batch can mix
+queues, priorities, and delays:
+
+```ts
+const ids = queue.enqueueMany("resize", [
+  { args: [imageA] },
+  { args: [imageB], options: { priority: 5 } },
+  { args: [imageC], options: { queue: "bulk", delayMs: 30_000 } },
+]);
+```
+
+An entry with a `uniqueKey` dedups exactly like `enqueue`: a key that already
+has a pending or running job resolves to that job's id instead of inserting a
+row, so the returned array always has one id per input entry.
+
+The batch is admitted or rejected as a whole — if a target queue's `maxPending`
+cap would be exceeded, the call throws `QueueFullError` and nothing is inserted.
+
+## Producer batching with `Batcher`
+
+Use `enqueueMany` when you already hold the list. When work arrives one item at
+a time — a request handler, a stream, a log tail — `Batcher` buffers the
+enqueues for you and flushes them as one `enqueueMany` call when either
+threshold fires, whichever comes first:
+
+```ts
+const batcher = queue.batcher("ingestLog", { maxSize: 100, maxWaitMs: 500 });
+
+for await (const line of incoming) {
+  const ids = batcher.add([line]);
+  // ids is empty unless this call crossed maxSize (100 lines) —
+  // in that case it holds the new jobs' ids, in input order.
+}
+
+batcher.close(); // flushes whatever remains in the buffer
+```
+
+`add(args, options?)` is typed exactly like `enqueue` and takes the same
+per-entry options, so a `Batcher` is no less expressive than calling
+`enqueueMany` yourself. It returns flushed job ids only when *that call*
+triggered a flush; otherwise the entry just joined the buffer and an empty array
+comes back. Call `flush()` to force one on demand:
+
+```ts
+const ids = batcher.flush(); // whatever's buffered right now, or []
+```
+
+`Batcher` implements `Symbol.dispose`, so a `using` declaration closes it at
+block exit:
+
+```ts
+using batcher = queue.batcher("ingestLog", { maxSize: 100, maxWaitMs: 500 });
+for (const line of lines) {
+  batcher.add([line]);
+}
+// close() runs here — the remainder is flushed
+```
+
+### Don't lose the tail
+
+The delay timer is `unref`'d, so a partially filled buffer never keeps the
+process alive — and is dropped if the process exits without closing the batcher.
+Close it on your shutdown path:
+
+```ts
+process.once("SIGTERM", () => {
+  batcher.close();
+});
+```
+
+### When a flush fails
+
+Nothing is dropped when a flush fails — the entries go back at the head of the
+buffer, ahead of anything added meanwhile:
+
+- A flush triggered by `add` or `flush` throws to the caller; the entries stay
+  buffered and `batcher.size` still counts them.
+- A timed flush has no caller to throw to, so it reports to the `onError` option
+  (or logs a warning) and re-arms the timer to retry on the next window.
+
+```ts
+const batcher = queue.batcher("ingestLog", {
+  maxWaitMs: 500,
+  onError: (error) => metrics.increment("batch_flush_failed", { error: String(error) }),
+});
+```
+
+## Worker dequeue batching (`batchSize`)
+
+`batchSize` on `runWorker` controls how many **already-enqueued** jobs the
+scheduler claims from storage in one poll — a throughput knob for the claim
+query, not a payload grouping mechanism:
+
+```ts
+const worker = queue.runWorker({
+  concurrency: 8,
+  batchSize: 16, // claim up to 16 due jobs per poll (default 1)
+});
+```
+
+Each claimed job is still dispatched to the handler individually, one job in,
+one result out — raising `batchSize` amortizes the polling round-trip under high
+throughput, it does not turn several jobs into one handler call.
+
+<Callout type="warn">
+  `Batcher`/`enqueueMany` batching and `batchSize` are unrelated despite the
+  shared name: one is a producer-side accumulator that runs *before* jobs are
+  created, the other is a scheduler-side claim size that applies *after* jobs
+  already exist in storage. Using one has no effect on the other.
+</Callout>
+
+<Cards>
+  <Card
+    title="Enqueue options"
+    href="/node/guides/core/enqueue-options"
+    description="Priority, delay, idempotency, metadata, and routing."
+  />
+  <Card
+    title="Workers"
+    href="/node/guides/core/workers"
+    description="runWorker options, events, and graceful shutdown."
+  />
+  <Card
+    title="Batching API reference"
+    href="/node/api-reference/batching"
+    description="Full enqueueMany and Batcher signatures."
+  />
+</Cards>
diff --git a/docs/content/docs/node/guides/core/index.mdx b/docs/content/docs/node/guides/core/index.mdx
index 4509f0b5..60ddd944 100644
--- a/docs/content/docs/node/guides/core/index.mdx
+++ b/docs/content/docs/node/guides/core/index.mdx
@@ -29,6 +29,11 @@ description: "The building blocks of every Node.js taskito application."
     href="/node/guides/core/enqueue-options"
     description="Per-job delay, priority, uniqueness, and metadata"
   />
+  <Card
+    title="Batching"
+    href="/node/guides/core/batching"
+    description="Buffer enqueues with Batcher, bulk-insert with enqueueMany"
+  />
   <Card
     title="Predicates"
     href="/node/guides/core/predicates"
diff --git a/docs/content/docs/node/guides/core/meta.json b/docs/content/docs/node/guides/core/meta.json
index 1d1a3655..e761d084 100644
--- a/docs/content/docs/node/guides/core/meta.json
+++ b/docs/content/docs/node/guides/core/meta.json
@@ -1 +1 @@
-{ "title": "Core", "pages": ["tasks", "queues", "workers", "execution-model", "enqueue-options", "predicates", "scheduling", "cancellation", "streaming", "pubsub"] }
+{ "title": "Core", "pages": ["tasks", "queues", "workers", "execution-model", "enqueue-options", "batching", "predicates", "scheduling", "cancellation", "streaming", "pubsub"] }
diff --git a/docs/content/docs/shared/guides/core/tasks.mdx b/docs/content/docs/shared/guides/core/tasks.mdx
index e1038e36..b911b826 100644
--- a/docs/content/docs/shared/guides/core/tasks.mdx
+++ b/docs/content/docs/shared/guides/core/tasks.mdx
@@ -861,7 +861,8 @@ for `enqueue_many()`, per-item overrides, and per-item result tracking.
 
 Each entry has its own `args` and `options`. See
 <SdkLink to="guides/core/enqueue-options">Enqueue options</SdkLink> for the
-full batch-enqueue contract.
+full batch-enqueue contract, and <SdkLink to="guides/core/batching">Batching</SdkLink>
+for the buffering `Batcher` helper built on top of it.
 
 </SdkOnly>
 

From 43ae17077224d5b7205ab5726d82e2c485b24c6e Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:22:20 +0530
Subject: [PATCH 4/6] docs: changelog entry for the Node batcher

---
 CHANGELOG.md                              | 6 ++++++
 docs/content/docs/resources/changelog.mdx | 6 ++++++
 2 files changed, 12 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 13054e38..7eadd8d2 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -40,6 +40,12 @@ underlying Rust crates are released together, in lock-step.
   (`registerPredicate`) and referenced as `gate(task, "name")` for config-driven gating,
   `queue.predicateStats()` reports what they decided, and `predicate.skipped` /
   `predicate.deferred` join `predicate.rejected` on the event bus.
+- **Node producer-side batch accumulator.** `queue.batcher(task, { maxSize, maxWaitMs })` buffers
+  one-at-a-time enqueues and flushes them as a single `enqueueMany` call on a size or time
+  trigger — the counterpart of Python's `BatchAccumulator` and Java's `Batcher<T>`. Entries keep
+  their own `args` and `EnqueueOptions`, a failed flush keeps them buffered (and a timed one
+  retries, reporting to `onError`), and `Symbol.dispose` means `using` flushes the tail at block
+  exit.
 - **Node bare-metal autoscaler.** `serveAutoscaler(queue, { app })` and `taskito autoscale <app>`
   spawn and drain worker processes to track queue depth on hosts without Kubernetes, closing the
   last scaling gap against Python's `AutoscaleController`. Same HPA-shaped formula — depth and
diff --git a/docs/content/docs/resources/changelog.mdx b/docs/content/docs/resources/changelog.mdx
index 7c04ad8e..f49b2ed6 100644
--- a/docs/content/docs/resources/changelog.mdx
+++ b/docs/content/docs/resources/changelog.mdx
@@ -47,6 +47,12 @@ underlying Rust crates are released together, in lock-step.
   (`registerPredicate`) and referenced as `gate(task, "name")` for config-driven gating,
   `queue.predicateStats()` reports what they decided, and `predicate.skipped` /
   `predicate.deferred` join `predicate.rejected` on the event bus.
+- **Node producer-side batch accumulator.** `queue.batcher(task, { maxSize, maxWaitMs })` buffers
+  one-at-a-time enqueues and flushes them as a single `enqueueMany` call on a size or time
+  trigger — the counterpart of Python's `BatchAccumulator` and Java's `Batcher<T>`. Entries keep
+  their own `args` and `EnqueueOptions`, a failed flush keeps them buffered (and a timed one
+  retries, reporting to `onError`), and `Symbol.dispose` means `using` flushes the tail at block
+  exit.
 - **Node bare-metal autoscaler.** `serveAutoscaler(queue, { app })` and `taskito autoscale <app>`
   spawn and drain worker processes to track queue depth on hosts without Kubernetes, closing the
   last scaling gap against Python's `AutoscaleController`. Same HPA-shaped formula — depth and

From 90c4c5f7dab66c4ced03bf160b503f63d0fda4a7 Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:34:53 +0530
Subject: [PATCH 5/6] fix(node): reject batcher maxWaitMs past the timer range

Node clamps a setTimeout delay above 2147483647ms to 1ms, turning a long wait
into an immediate flush.
---
 docs/content/docs/node/api-reference/batching.mdx | 2 +-
 sdks/node/src/batching.ts                         | 9 +++++++--
 sdks/node/test/core/batcher.test.ts               | 4 ++++
 3 files changed, 12 insertions(+), 3 deletions(-)

diff --git a/docs/content/docs/node/api-reference/batching.mdx b/docs/content/docs/node/api-reference/batching.mdx
index 83eaedf0..82d2023b 100644
--- a/docs/content/docs/node/api-reference/batching.mdx
+++ b/docs/content/docs/node/api-reference/batching.mdx
@@ -61,7 +61,7 @@ Python's `BatchAccumulator` and Java's `Batcher<T>`.
 | Option | Default | Description |
 |---|---|---|
 | `maxSize` | `100` | Flush as soon as this many entries have accumulated. Must be a positive integer. |
-| `maxWaitMs` | `500` | Flush this long after the first buffered entry, even if `maxSize` isn't reached — the worst-case latency of any one entry. Must be positive. |
+| `maxWaitMs` | `500` | Flush this long after the first buffered entry, even if `maxSize` isn't reached — the worst-case latency of any one entry. Must be between 1 and 2147483647 (Node clamps longer timer delays to 1ms). |
 | `onError` | — | Called when a *timed* flush throws. Without it the failure is only logged. |
 
 ```ts
diff --git a/sdks/node/src/batching.ts b/sdks/node/src/batching.ts
index df00781a..4ce82fc1 100644
--- a/sdks/node/src/batching.ts
+++ b/sdks/node/src/batching.ts
@@ -9,6 +9,9 @@ import { createLogger } from "./utils";
 
 const DEFAULT_MAX_SIZE = 100;
 const DEFAULT_MAX_WAIT_MS = 500;
+// Node clamps any `setTimeout` delay above this to 1ms, which would turn a long
+// wait into an immediate flush — reject it at construction instead.
+const MAX_TIMER_DELAY_MS = 2_147_483_647;
 const log = createLogger("batching");
 
 /** Construction options for a {@link Batcher}. */
@@ -77,8 +80,10 @@ export class Batcher<
       throw new RangeError(`Batcher maxSize must be a positive integer, got ${maxSize}`);
     }
     const maxWaitMs = options.maxWaitMs ?? DEFAULT_MAX_WAIT_MS;
-    if (!Number.isFinite(maxWaitMs) || maxWaitMs <= 0) {
-      throw new RangeError(`Batcher maxWaitMs must be a positive finite number, got ${maxWaitMs}`);
+    if (!Number.isFinite(maxWaitMs) || maxWaitMs <= 0 || maxWaitMs > MAX_TIMER_DELAY_MS) {
+      throw new RangeError(
+        `Batcher maxWaitMs must be between 1 and ${MAX_TIMER_DELAY_MS}, got ${maxWaitMs}`,
+      );
     }
     this.maxSize = maxSize;
     this.maxWaitMs = maxWaitMs;
diff --git a/sdks/node/test/core/batcher.test.ts b/sdks/node/test/core/batcher.test.ts
index 7cf9df06..8fab6e58 100644
--- a/sdks/node/test/core/batcher.test.ts
+++ b/sdks/node/test/core/batcher.test.ts
@@ -150,6 +150,10 @@ it("rejects invalid tunables", () => {
   expect(() => queue.batcher("collect", { maxSize: 1.5 })).toThrow(RangeError);
   expect(() => queue.batcher("collect", { maxWaitMs: 0 })).toThrow(RangeError);
   expect(() => queue.batcher("collect", { maxWaitMs: Number.NaN })).toThrow(RangeError);
+  // Node clamps a delay past the 32-bit timer range to 1ms — an immediate flush
+  // instead of the ~25-day wait that was asked for.
+  expect(() => queue.batcher("collect", { maxWaitMs: 2_147_483_648 })).toThrow(RangeError);
+  expect(() => queue.batcher("collect", { maxWaitMs: 2_147_483_647 })).not.toThrow();
 });
 
 it("types add() from the registered task", () => {

From 89393e8a50a470d902c842f81560d2730ee474bb Mon Sep 17 00:00:00 2001
From: stromanni <dalstromannika4@gmail.com>
Date: Sat, 25 Jul 2026 17:35:23 +0530
Subject: [PATCH 6/6] fix(node): contain a throwing batcher onError

A handler that threw escaped the timer callback as an uncaught exception and
skipped the re-arm, stranding the buffered entries.
---
 .../docs/node/api-reference/batching.mdx      |  2 +-
 sdks/node/src/batching.ts                     | 25 ++++++++++++-------
 sdks/node/test/core/batcher.test.ts           | 24 ++++++++++++++++++
 3 files changed, 41 insertions(+), 10 deletions(-)

diff --git a/docs/content/docs/node/api-reference/batching.mdx b/docs/content/docs/node/api-reference/batching.mdx
index 82d2023b..765a953d 100644
--- a/docs/content/docs/node/api-reference/batching.mdx
+++ b/docs/content/docs/node/api-reference/batching.mdx
@@ -62,7 +62,7 @@ Python's `BatchAccumulator` and Java's `Batcher<T>`.
 |---|---|---|
 | `maxSize` | `100` | Flush as soon as this many entries have accumulated. Must be a positive integer. |
 | `maxWaitMs` | `500` | Flush this long after the first buffered entry, even if `maxSize` isn't reached — the worst-case latency of any one entry. Must be between 1 and 2147483647 (Node clamps longer timer delays to 1ms). |
-| `onError` | — | Called when a *timed* flush throws. Without it the failure is only logged. |
+| `onError` | — | Called when a *timed* flush throws. Without it the failure is only logged. A handler that throws is contained and logged. |
 
 ```ts
 using batcher = queue.batcher("sendEmail", { maxSize: 100, maxWaitMs: 500 });
diff --git a/sdks/node/src/batching.ts b/sdks/node/src/batching.ts
index 4ce82fc1..227c25be 100644
--- a/sdks/node/src/batching.ts
+++ b/sdks/node/src/batching.ts
@@ -181,15 +181,22 @@ export class Batcher<
     try {
       this.flush();
     } catch (error) {
-      if (this.onError) {
-        this.onError(error);
-      } else {
-        log.warn(() => `batched flush for task "${this.name}" failed`, error);
-      }
-      // The entries are still buffered — re-arm so a transient failure doesn't
-      // strand them until the next `add` or an explicit flush.
-      if (!this.isClosed && this.buffer.length > 0) {
-        this.arm();
+      try {
+        if (this.onError) {
+          this.onError(error);
+        } else {
+          log.warn(() => `batched flush for task "${this.name}" failed`, error);
+        }
+      } catch (reportingError) {
+        // A throwing handler must not escape the timer callback — that would be
+        // an uncaught exception and would skip the re-arm below.
+        log.warn(() => `batcher onError for task "${this.name}" threw`, reportingError);
+      } finally {
+        // The entries are still buffered — re-arm so a transient failure doesn't
+        // strand them until the next `add` or an explicit flush.
+        if (!this.isClosed && this.buffer.length > 0) {
+          this.arm();
+        }
       }
     }
   }
diff --git a/sdks/node/test/core/batcher.test.ts b/sdks/node/test/core/batcher.test.ts
index 8fab6e58..869130ea 100644
--- a/sdks/node/test/core/batcher.test.ts
+++ b/sdks/node/test/core/batcher.test.ts
@@ -156,6 +156,30 @@ it("rejects invalid tunables", () => {
   expect(() => queue.batcher("collect", { maxWaitMs: 2_147_483_647 })).not.toThrow();
 });
 
+it("contains a throwing onError and still retries the flush", async () => {
+  const queue = newQueue();
+  let calls = 0;
+  const batcher = queue.batcher("collect", {
+    maxSize: 100,
+    maxWaitMs: 30,
+    onError: () => {
+      calls += 1;
+      throw new Error("reporting blew up");
+    },
+  });
+  const enqueueMany = vi.spyOn(queue, "enqueueMany").mockImplementationOnce(() => {
+    throw new Error("storage down");
+  });
+
+  batcher.add([1]);
+  expect(await waitFor(async () => calls > 0)).toBe(true);
+  // The handler throwing must not skip the re-arm, so the entry still lands.
+  expect(await waitFor(async () => (await queue.stats()).pending === 1)).toBe(true);
+  expect(batcher.size).toBe(0);
+  enqueueMany.mockRestore();
+  batcher.close();
+});
+
 it("types add() from the registered task", () => {
   const queue = newQueue();
   const batcher = queue.batcher("collect", { maxSize: 100 });