feat(node): producer-side batch accumulator

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

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 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. A handler that throws is contained and 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>

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

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"] }

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>

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"

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"] }

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

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>