@studnicky/batch

Batch concurrent execution for processing items in controlled batches.

Install

bash

pnpm add @studnicky/batch

Requires @studnicky:registry=https://npm.pkg.github.com in .npmrc.

Usage

Batch.create(maxConcurrent) returns a batch processor. Its process method is an async generator that yields results batch-by-batch, enabling incremental processing and backpressure handling. Pass any async operation:

import { Batch } from '../src/index.js';
import { BasicProcessingFixture } from './fixtures/BasicProcessingFixture.js';

class NumberItem {
  static double(n: number): Promise<number> {
    const doubled = n * 2;
    return Promise.resolve(doubled);
  }
}

const allResults: number[] = [];
let batchIndex = 0;

for await (const batch of Batch.create<number>(2).process(BasicProcessingFixture.Items, NumberItem.double)) {
  console.log(`Batch ${batchIndex}:`, batch);
  allResults.push(...batch);
  batchIndex++;
}

console.log('All results:', allResults);

Partial-failure support

processSettled uses Promise.allSettled internally so a single rejection does not abort the batch or subsequent batches. Each yield produces a PromiseSettledResult[] covering both fulfilled values and rejection reasons:

import { Batch } from '../src/index.js';
import { SettledProcessingFixture } from './fixtures/SettledProcessingFixture.js';

// Custom typed mapper — the extension seam for Batch.
// Static method on the produced type is the canonical shape.
class Result {
  constructor(
    readonly id: number,
    readonly value: string
  ) {}

  static process(item: ItemEntity.Type): Promise<Result> {
    if (item.shouldFail) {
      return Promise.reject(new Error(`Item ${item.id} failed`));
    }
    return Promise.resolve(new Result(item.id, `processed-${item.id}`));
  }
}

class SettledProcessingExample {
  static async run(): Promise<void> {
    const allSettled: PromiseSettledResult<Result>[] = [];

    for await (const batch of Batch.create<Result>(2).processSettled(SettledProcessingFixture.Items, Result.process)) {
      console.log('Batch settled results:');
      for (const result of batch) {
        if (result.status === 'fulfilled') {
          console.log('  fulfilled:', result.value);
        } else {
          console.log('  rejected:', (result.reason as Error).message);
        }
      }
      allSettled.push(...batch);
    }

    console.log('Total settled:', allSettled.length);

Observability hooks

Subclass Batch and override its protected lifecycle hooks to observe each stage of a run. Every hook is a no-op by default, so an un-subclassed Batch does no observability:

Hook	When it fires	Args
`onBatchStart`	Once, before the first batch begins	`total: number`
`onItemStart`	When each item begins processing	`index: number`
`onItemSuccess`	When an item resolves	`index: number, result: TResult`
`onItemError`	When an item rejects	`index: number, error: unknown`
`onItemSettled`	After each item finishes (success or error), after `onItemSuccess`/`onItemError`	`index: number`
`onConcurrencySaturated`	At the start of each batch where all concurrency slots are occupied	(none)
`onBatchComplete`	Once, after all items are processed	`stats: { total, succeeded, failed }`

import type { BatchStatsEntity } from '../src/index.js';

import { Batch } from '../src/index.js';
import { ObservedBatchFixture } from './fixtures/ObservedBatchFixture.js';

class ObservedBatch extends Batch<string> {
  public readonly capturedItemStarts: number[] = [];
  public readonly capturedSuccesses: { 'index': number; 'value': string }[] = [];
  public readonly capturedErrors: { 'index': number; 'message': string }[] = [];
  public readonly capturedSettled: number[] = [];
  public capturedSaturations = 0;
  public capturedStats: BatchStatsEntity.Type | undefined;

  public constructor(maxConcurrent?: number) { super(maxConcurrent); }

  protected override onBatchStart(total: number): void {
    console.log(`[batch] start — ${total} items`);
  }

  protected override onBatchComplete(stats: BatchStatsEntity.Type): void {
    console.log(`[batch] complete — total=${stats.total} succeeded=${stats.succeeded} failed=${stats.failed}`);
    this.capturedStats = stats;
  }

  protected override onConcurrencySaturated(): void {
    console.log('[batch] concurrency saturated — all slots in use');
    this.capturedSaturations++;
  }

  protected override onItemError(index: number, error: unknown): void {
    const msg = error instanceof Error ? error.message : String(error);
    console.log(`[batch] item[${index}] error — ${msg}`);
    this.capturedErrors.push({ 'index': index, 'message': msg });
  }

  protected override onItemSettled(index: number): void {
    console.log(`[batch] item[${index}] settled`);
    this.capturedSettled.push(index);
  }

  protected override onItemStart(index: number): void {
    console.log(`[batch] item[${index}] start`);
    this.capturedItemStarts.push(index);
  }

  protected override onItemSuccess(index: number, result: string): void {
    console.log(`[batch] item[${index}] success → ${result}`);
    this.capturedSuccesses.push({ 'index': index, 'value': result });
  }
}

class ObservedBatchExample {
  static async run(): Promise<void> {
    const observed = new ObservedBatch(2);
    const allSettled: PromiseSettledResult<string>[] = [];

    for await (const batchResults of observed.processSettled(
      ObservedBatchFixture.Tasks,
      (task) => {
        if (task.id === 3) {
          return Promise.reject(new Error(`task ${task.id} (${task.label}) failed`));
        }
        return Promise.resolve(`processed-${task.label}`);
      }
    )) {
      allSettled.push(...batchResults);
    }

batch never calls any logger or metrics library. Overriding the protected lifecycle hooks is the only observability seam.

Subpath exports

Subpath	Contents
`@studnicky/batch`	`Batch`
`@studnicky/batch/batch`	`Batch` (direct subpath)
`@studnicky/batch/constants`	Default batch configuration constants

Try it

Batch is a subclass-first primitive: configure concurrency through Batch.create(maxConcurrent) and add observability by overriding its protected lifecycle hooks.

Usage

Run a batch of items with concurrency 2 and watch results arrive batch-by-batch.

Batch concurrent processing

/** basic-processing — double five numbers with concurrency 2, batch by batch. Run: npx tsx examples/basic-processing.ts */

import assert from 'node:assert/strict';

// #region usage
import { Batch } from '../src/index.js';
import { BasicProcessingFixture } from './fixtures/BasicProcessingFixture.js';

class NumberItem {
  static double(n: number): Promise<number> {
    const doubled = n * 2;
    return Promise.resolve(doubled);
  }
}

const allResults: number[] = [];
let batchIndex = 0;

for await (const batch of Batch.create<number>(2).process(BasicProcessingFixture.Items, NumberItem.double)) {
  console.log(`Batch ${batchIndex}:`, batch);
  allResults.push(...batch);
  batchIndex++;
}

console.log('All results:', allResults);
// #endregion usage

// All 5 items must have been processed.
assert.equal(allResults.length, 5, 'Expected 5 results');

// Results must match expected doubled values (order preserved within each batch).
assert.deepEqual(allResults.sort((a, b) => {return a - b;}), BasicProcessingFixture.ExpectedDoubled);

console.log('basic-processing: all assertions passed');

Output

Press Execute to run this example against the real library.

Hooks

Each overridden hook fires in order — onBatchStart, then per-item onItemStart/onItemSuccess (or onItemError)/onItemSettled, then onBatchComplete. Item 3 rejects intentionally so onItemError is visible.

Batch lifecycle hooks

/** observedBatch — trace batch progress by subclassing Batch and overriding lifecycle hooks. Run: npx tsx examples/observedBatch.ts */

import assert from 'node:assert/strict';

// #region usage
import type { BatchStatsEntity } from '../src/index.js';

import { Batch } from '../src/index.js';
import { ObservedBatchFixture } from './fixtures/ObservedBatchFixture.js';

class ObservedBatch extends Batch<string> {
  public readonly capturedItemStarts: number[] = [];
  public readonly capturedSuccesses: { 'index': number; 'value': string }[] = [];
  public readonly capturedErrors: { 'index': number; 'message': string }[] = [];
  public readonly capturedSettled: number[] = [];
  public capturedSaturations = 0;
  public capturedStats: BatchStatsEntity.Type | undefined;

  public constructor(maxConcurrent?: number) { super(maxConcurrent); }

  protected override onBatchStart(total: number): void {
    console.log(`[batch] start — ${total} items`);
  }

  protected override onBatchComplete(stats: BatchStatsEntity.Type): void {
    console.log(`[batch] complete — total=${stats.total} succeeded=${stats.succeeded} failed=${stats.failed}`);
    this.capturedStats = stats;
  }

  protected override onConcurrencySaturated(): void {
    console.log('[batch] concurrency saturated — all slots in use');
    this.capturedSaturations++;
  }

  protected override onItemError(index: number, error: unknown): void {
    const msg = error instanceof Error ? error.message : String(error);
    console.log(`[batch] item[${index}] error — ${msg}`);
    this.capturedErrors.push({ 'index': index, 'message': msg });
  }

  protected override onItemSettled(index: number): void {
    console.log(`[batch] item[${index}] settled`);
    this.capturedSettled.push(index);
  }

  protected override onItemStart(index: number): void {
    console.log(`[batch] item[${index}] start`);
    this.capturedItemStarts.push(index);
  }

  protected override onItemSuccess(index: number, result: string): void {
    console.log(`[batch] item[${index}] success → ${result}`);
    this.capturedSuccesses.push({ 'index': index, 'value': result });
  }
}

class ObservedBatchExample {
  static async run(): Promise<void> {
    const observed = new ObservedBatch(2);
    const allSettled: PromiseSettledResult<string>[] = [];

    for await (const batchResults of observed.processSettled(
      ObservedBatchFixture.Tasks,
      (task) => {
        if (task.id === 3) {
          return Promise.reject(new Error(`task ${task.id} (${task.label}) failed`));
        }
        return Promise.resolve(`processed-${task.label}`);
      }
    )) {
      allSettled.push(...batchResults);
    }
    // #endregion usage

    // ── post-run assertions ───────────────────────────────────────────────────────

    // onItemStart fired for all 5 items; indices cover 0–4
    assert.strictEqual(observed.capturedItemStarts.length, 5, 'onItemStart must fire for every item');
    assert.deepStrictEqual(
      observed.capturedItemStarts.slice().sort((a, b) => { return a - b; }),
      [0, 1, 2, 3, 4]
    );

    // onItemSuccess: 4 items succeed (ids 1,2,4,5 → indices 0,1,3,4)
    assert.strictEqual(observed.capturedSuccesses.length, 4, 'onItemSuccess must fire for 4 successes');

    const successIndices: number[] = [];
    for (const e of observed.capturedSuccesses) {
      successIndices.push(e.index);
    }
    successIndices.sort((a, b) => { return a - b; });

    assert.deepStrictEqual(successIndices, [0, 1, 3, 4]);

    // onItemError: 1 item fails (index 2)
    assert.strictEqual(observed.capturedErrors.length, 1, 'onItemError must fire exactly once');
    assert.strictEqual(observed.capturedErrors[0]!.index, 2);
    assert.ok(observed.capturedErrors[0]!.message.includes('gamma'));

    // onItemSettled: fires for all 5 items
    assert.strictEqual(observed.capturedSettled.length, 5, 'onItemSettled must fire for every item');

    // onConcurrencySaturated: 2 full batches of 2 out of batches [0,1],[2,3],[4]
    assert.strictEqual(observed.capturedSaturations, 2, 'onConcurrencySaturated must fire for each full batch');

    // onBatchComplete: once, after processSettled finishes all batches
    assert.ok(observed.capturedStats !== undefined, 'onBatchComplete must fire');
    assert.deepStrictEqual(observed.capturedStats, { 'failed': 1, 'succeeded': 4, 'total': 5 });

    // processSettled produces 5 settled results
    assert.strictEqual(allSettled.length, 5);

    console.log('observedBatch: all assertions passed');
  }
}

await ObservedBatchExample.run();

Output

Press Execute to run this example against the real library.

Source on GitHub

@studnicky/batch ​

Install ​

Usage ​

Partial-failure support ​

Observability hooks ​

Subpath exports ​

Try it ​

Usage ​

Hooks ​

@studnicky/batch

Install

Usage

Partial-failure support

Observability hooks

Subpath exports

Try it

Usage

Hooks