@studnicky/batch

Batch concurrent execution for processing items in controlled batches.

Install

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;
    const result = Promise.resolve(doubled);
    return result;
  }
}

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) {
      const result = Promise.reject(RuntimeError.create(`Item ${item.id} failed`));
      return result;
    }
    const result = Promise.resolve(new Result(item.id, `processed-${item.id}`));
    return result;
  }
}

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:');
      const batchLength = batch.length;
      for (let index = 0; index < batchLength; index += 1) {
        const result = batch[index]!;
        if (result.status === 'fulfilled') {
          console.log('  fulfilled:', result.value);
        } else {
          const error = result.reason instanceof Error ? result.reason : RuntimeError.create(String(result.reason));
          const message = error.message;
          console.log('  rejected:', 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/entities/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(maximumConcurrent?: number) { super(maximumConcurrent); }

  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: Error): void {
    const message = error.message;
    console.log(`[batch] item[${index}] error — ${message}`);
    this.capturedErrors.push({ 'index': index, 'message': message });
  }

  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 });
  }

  static processTask(task: typeof ObservedBatchFixture.Tasks[number]): Promise<string> {
    if (task.id === 3) {
      const result = Promise.reject(RuntimeError.create(`task ${task.id} (${task.label}) failed`));
      return result;
    }
    const result = Promise.resolve(`processed-${task.label}`);
    return 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,
      ObservedBatch.processTask
    )) {
      allSettled.push(...batchResults);
    }

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

Public API

Import Batch and BatchError from @studnicky/batch; import BatchStatsEntity from @studnicky/batch/entities. Batching constants are implementation details.

BatchError.retryable uses the canonical ErrorClassificationEntity.Type['retryable'] field. Import ErrorClassificationEntity directly from @studnicky/errors; @studnicky/batch does not proxy-export dependency functionality.

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.

Loading example…

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.

Loading example…

Entities

@studnicky/batch/entities exports batch statistics schemas.

import { BatchStatsEntity } from '@studnicky/batch/entities';

Exports

Symbol	Purpose	Import path
Batch	Processes items in bounded concurrent batches.	@studnicky/batch
BatchError	Represents batch processing failures.	@studnicky/batch

Source on GitHub