Substrate

Appearance

Pattern Composition ​

A pattern in Substrate is a named composition of two or more primitives — a repeatable runtime shape, still usable without a graph executor. Five pattern families have shipped as real packages: the Execution Kit (@studnicky/request-executor), the Boundary Kit (@studnicky/boundary-kit), the Coordination Kit — split into @studnicky/keyed-work-gate and @studnicky/bounded-dispatcher since a single kit would force the event-bus dependency onto callers who only want keyed serialization — and the Process Kit (@studnicky/process-kit). Every primitive each kit wraps is already fully composable on its own, so each kit's value is a fixed, correctly-ordered composition rather than new runtime behavior: getting the order backwards by hand is easy to do silently, and each kit exists to protect a consumer from re-deriving it.

This page documents the fixed composition order each kit uses internally, and how to reach the identical result by hand-composing the same primitives yourself without depending on the kit package at all.

See Lifecycle Hooks for the hook idiom every composed primitive uses, Composition Anti-Patterns for where composition tips into orchestration, and the Dagonizer Boundary guide for where substrate composition ends and workflow orchestration begins.

The Execution Kit's composition order ​

RequestExecutor#execute() composes its five primitives in one fixed order:

context scope → timing span → retry loop → the caller's fn(client, signal) call

This order is fixed, not configurable, because each layer's job only makes sense wrapping everything inside it:

  • Context scope outermost — ContextScope#execute() has to wrap the entire call, retries included, so that a value set on attempt 2 is still readable on attempt 3, and so scope.terminate() only runs once the whole operation (successful or not) is finished.
  • Timing span next — the span is meant to answer "how long did this whole request take, including every retry attempt," not "how long did the last attempt take." Bracketing outside the retry loop captures that.
  • Retry loop wraps the actual call — retry needs to be the closest layer to fn so that every attempt, not just the first, gets the same cancellation signal and runs inside the same context/timing span.
  • The composed cancellation signal is threaded through, not layered — Signal#compose() runs once, before any of the above, merging a caller-supplied AbortSignal and/or deadlineMs into a single AbortSignal that every retry attempt receives identically. A signal is a value passed down, not a nesting layer.

Using the kit ​

ts
import type { RequestContextType, ResponseContextType } from '@studnicky/fetch/interfaces';
import type { RetryConfigInterface, RetryContextType } from '@studnicky/retry';

import { FetchClient } from '@studnicky/fetch';
import { Retry } from '@studnicky/retry';
import assert from 'node:assert/strict';
import { createServer } from 'node:http';

import { RequestExecutor } from '../src/index.js';

class TelemetryFetchClient extends FetchClient {
  readonly requestPaths: string[] = [];

  static override create(config = {}): TelemetryFetchClient {
    return new this(config);
  }

  protected override onRequest(context: RequestContextType): Promise<RequestContextType> {
    console.log(`[fetch] ${context.metadata.method} ${context.metadata.path}`);
    this.requestPaths.push(context.metadata.path);
    return Promise.resolve(context);
  }

  protected override onResponse(context: ResponseContextType): Promise<ResponseContextType> {
    console.log(`[fetch] <- ${context.response.status}`);
    return Promise.resolve(context);
  }
}

class TelemetryRetry extends Retry {
  readonly scheduledRetries: number[] = [];

  constructor(config?: Partial<RetryConfigInterface>) {
    super(config ?? {});
  }

  protected override onRetryScheduled(context: RetryContextType): void {
    console.log(`[retry] attempt ${context.attemptNumber} scheduled retry`);
    this.scheduledRetries.push(context.attemptNumber);
  }
}

/**
 * Advanced extension: RequestExecutor has no hooks of its own — observability is
 * delegated entirely to the composed primitives. A subclass can still add
 * convenience behavior by reaching the composed instances through the getters.
 */
class ReportingRequestExecutor extends RequestExecutor {
  // `this.create(...)` (not `RequestExecutor.create(...)`) so the inherited factory's
  // `new this(...)` binds to ReportingRequestExecutor — same `new this()` polymorphism
  // FetchClient/Timing/Retry use for their own subclass factories.
  static tracked(fetchClient: TelemetryFetchClient, retry: TelemetryRetry): ReportingRequestExecutor {
    const result = this.create({ 'fetchClient': fetchClient, 'retry': retry }) as ReportingRequestExecutor;
    return result;
  }

  report(): { 'retries': number; 'totalRequests': number } {
    const stats = this.getRetry().getStats();

    return { 'retries': stats.totalRetries, 'totalRequests': stats.totalRequests };
  }
}

Composing the same primitives directly, without the kit ​

RequestExecutor is a thin facade, not a hidden runtime — the Layer Transparency Rule requires that a higher layer never trap a caller away from the primitives it composes. The kit's entire execute() body is under 40 lines of composition glue; a caller who wants the identical behavior without depending on @studnicky/request-executor at all can hand-compose the same five primitives directly:

ts
import { Context } from '@studnicky/context';
import { FetchClient } from '@studnicky/fetch';
import { Retry } from '@studnicky/retry';
import { Signal } from '@studnicky/signal';
import { Timing, TIMING_STATUS, TimingEvent } from '@studnicky/timing';
import assert from 'node:assert/strict';
import { createServer } from 'node:http';

/**
 * The same composition order RequestExecutor#execute() uses internally: a context scope
 * wraps the whole call, a timing span brackets the retry loop, the retry loop wraps the
 * caller's fn, and the composed cancellation signal threads through into whatever call
 * fn makes. Nothing here is hidden inside a facade class — every primitive is a plain
 * local variable the caller owns.
 */
class Directly {
  static async execute<T>(
    fetchClient: FetchClient,
    retry: Retry,
    signal: Signal,
    timing: Timing,
    context: Context,
    fn: (client: FetchClient, abortSignal: AbortSignal) => Promise<T>,
    options: { 'deadlineMs'?: number } = {}
  ): Promise<T> {
    const composedSignal = await signal.compose(
      options.deadlineMs !== undefined ? { 'deadlineMs': options.deadlineMs } : {}
    );

    const scope = context.initialize();

    try {
      const result = await scope.execute(async () => {
        timing.event(
          TimingEvent.create().component('directComposition').operation('execute').status(TIMING_STATUS.START).build()
        );

        try {
          const attemptResult = await retry.execute(() => { const result = fn(fetchClient, composedSignal); return result; });

          timing.event(
            TimingEvent.create().component('directComposition').operation('execute').status(TIMING_STATUS.COMPLETE).build()
          );

          return attemptResult;
        } catch (error) {
          timing.event(
            TimingEvent.create().component('directComposition').operation('execute').status(TIMING_STATUS.ERROR).build()
          );
          throw error;
        }
      });

      return result;
    } finally {
      scope.terminate();
    }
  }
}

Both versions are correct, produce identical results, and use exactly the same five primitives in exactly the same order — one names the order once inside a reusable class, the other writes it out inline. Reach for the kit when the same composition repeats across many call sites; hand-compose when a single call site needs a one-off variant (a different composition order, an extra primitive spliced in, or no Context at all).

The Boundary Kit ​

@studnicky/boundary-kit's BoundaryKit composes @studnicky/throttle, @studnicky/resilience's CircuitBreaker, and @studnicky/retry into a fixed-order "protect a flaky dependency" shape. BoundaryKit.create() accepts config for each of the three composed primitives and returns a kit whose execute(fn) runs fn through all three in the fixed order below — see the "Using the kit" example for the full call shape.

Composition order: throttle (bounds concurrency) → circuitBreaker (fast-fail) → retry (attempt+backoff) → fn — throttle.execute(() => circuitBreaker.execute(() => retry.execute(fn))). Throttle sits outermost because concurrency has to be bounded before a call is even attempted; the circuit breaker sits next so an already-open circuit short-circuits before retry burns any attempts against a dependency known to be down; retry sits innermost, closest to fn, so only genuinely transient errors get retried before a failure reaches the circuit breaker's failure count. Each composed primitive accepts either a pre-built instance (subclassed or not) or the config shape passed straight to that primitive's own create() — circuitBreaker has no zero-arg default of its own, so an omitted circuitBreaker key resolves against BoundaryKit's own default ({ failureThreshold: 5, resetTimeoutMs: 30_000 }).

Throttle#execute() resolves undefined (rather than rejecting) when the throttle discards a call via its detach-and-abandon abort behavior; BoundaryKit#execute() cannot return undefined as T, so it surfaces that discard as a rejected BoundaryKitAbortedError instead.

ts
import type { CircuitBreakerOptionsInterface } from '@studnicky/resilience';
import type { RetryConfigInterface, RetryContextType } from '@studnicky/retry';
import type { ThrottleConfigEntity } from '@studnicky/throttle';

import { CircuitBreaker } from '@studnicky/resilience';
import { Retry } from '@studnicky/retry';
import { Throttle } from '@studnicky/throttle';
import assert from 'node:assert/strict';

import { BoundaryKit } from '../src/index.js';

/**
 * Advanced usage: BoundaryKit has no hooks of its own — observability is delegated
 * entirely to the composed primitives. Subclass Throttle/CircuitBreaker/Retry directly
 * and pass the pre-built instances in; their own hooks keep firing exactly as they
 * would standalone, and the kit's getters return those exact subclass instances back.
 */
class TelemetryThrottle extends Throttle {
  readonly acquisitions: number[] = [];

  constructor(config?: Partial<ThrottleConfigEntity.Type>) {
    super(config);
  }

  protected override onAcquire(activeCount: number, queuedCount: number): void {
    console.log(`[throttle] slot acquired (active=${String(activeCount)}, queued=${String(queuedCount)})`);
    this.acquisitions.push(activeCount);
  }
}

class TelemetryCircuitBreaker extends CircuitBreaker {
  readonly rejections: number[] = [];

  constructor(options: CircuitBreakerOptionsInterface) {
    super(options);
  }

  protected override onReject(): void {
    console.log('[circuitBreaker] rejected — circuit is open');
    this.rejections.push(Date.now());
  }
}

class TelemetryRetry extends Retry {
  readonly scheduledRetries: number[] = [];

  constructor(config?: Partial<RetryConfigInterface>) {
    super(config ?? {});
  }

  protected override onRetryScheduled(context: RetryContextType): void {
    console.log(`[retry] attempt ${String(context.attemptNumber)} scheduled retry`);
    this.scheduledRetries.push(context.attemptNumber);
  }
}

class ObservedBoundaryKitExample {
  static async run(): Promise<void> {
    /**
     * Default construction: BoundaryKit.create() with no config resolves every composed
     * primitive to sensible defaults — including CircuitBreaker, which has no zero-arg
     * default of its own.
     */
    const defaultKit = BoundaryKit.create();

    let defaultAttempts = 0;

    const defaultResult = await defaultKit.execute(() => {
      defaultAttempts += 1;

      if (defaultAttempts < 2) {
        throw new Error('transient failure');
      }

      return 'default-ok';
    });

    console.log('Default kit result:', defaultResult, `(${String(defaultAttempts)} attempts)`);

    const throttle = new TelemetryThrottle({ 'concurrencyLimit': 3 });
    const circuitBreaker = new TelemetryCircuitBreaker({ 'failureThreshold': 2, 'resetTimeoutMs': 5000 });
    const retry = new TelemetryRetry({ 'maxRetries': 2 });

    const observedKit = BoundaryKit.create({
      'circuitBreaker': circuitBreaker,
      'retry': retry,
      'throttle': throttle
    });

    let flakyAttempts = 0;

    const observedResult = await observedKit.execute(() => {
      flakyAttempts += 1;

      if (flakyAttempts < 2) {
        throw new Error('transient failure');
      }

      return 'observed-ok';
    });

    console.log('Observed kit result:', observedResult);
    console.log('Throttle acquisitions:', throttle.acquisitions);
    console.log('Retry scheduled attempts:', retry.scheduledRetries);

BoundaryKit introduces no hook of its own — every observable stage is already covered by the composed primitive it delegates to, reachable through getThrottle(), getCircuitBreaker(), and getRetry(). Every getter returns the exact instance passed to create()/builder() — never a copy or wrapper — so a caller who subclassed Throttle/CircuitBreaker/Retry for their own hooks (onAcquire, onOpen, onGiveUp, and so on) keeps full access to those subclasses' own hooks. See @studnicky/retry, @studnicky/resilience, and @studnicky/throttle for each primitive's full hook table.

Hand-composing the Boundary Kit without the package ​

All three primitives are already fully composition-ready on their own, so the identical throttle → circuitBreaker → retry → fn order can be written out inline instead of depending on @studnicky/boundary-kit:

ts
import { CircuitBreaker, CircuitBreakerOpenError } from '@studnicky/resilience';
import { Retry } from '@studnicky/retry';
import assert from 'node:assert/strict';

import { Throttle } from '../src/index.js';

// One Retry/CircuitBreaker pair per protected dependency; one Throttle shared across every
// call made to that dependency, bounding how much concurrent load it ever sees.
const retry = Retry.create({ 'maxRetries': 2 });
const circuitBreaker = CircuitBreaker.create({ 'failureThreshold': 3, 'resetTimeoutMs': 60_000 });
const throttle = Throttle.create({ 'concurrencyLimit': 2 });

// The kit's entire value-add would have been this composition order — nothing here is
// hidden: retry, circuitBreaker, and throttle stay reachable as plain local variables,
// each with its own hooks, getStats()/state, and updateConfig() untouched.
class ThroughBoundary {
  static run<T>(fn: () => Promise<T>): Promise<T | undefined> {
    const result = throttle.execute(() => { const result = circuitBreaker.execute(() => { const result = retry.execute(fn); return result; }); return result; });
    return result;
  }
}

Reach for BoundaryKit when the same throttle-then-circuit-breaker-then-retry composition repeats across many call sites and BoundaryKitAbortedError's uniform error surface is useful; hand-compose when a single call site needs a one-off variant (a different composition order, an extra primitive spliced in, or direct access to Throttle#execute()'s raw undefined-on-discard result).

The Coordination Kit ​

The Coordination Kit splits into two independent packages rather than one kit, since a single kit would force the event-bus dependency onto callers who only want keyed serialization.

KeyedWorkGate: single-flight coalescing falling through to mutex-serialized access ​

@studnicky/keyed-work-gate's KeyedWorkGate composes @studnicky/mutex's Mutex and @studnicky/concurrency's Coalesce into two related recipes for a keyed resource: runSingleFlight collapses concurrent same-key callers onto one in-flight execution (via Coalesce), which itself runs under mutex-guarded exclusive access; runSerialized skips coalescing entirely and goes straight to Mutex for callers that need every call to actually execute, just never concurrently for the same key. KeyedWorkGate.create<TKey>() returns a gate whose runSingleFlight(key, fn) and runSerialized(key, fn) both take the key and the unit of work as arguments — see the example below for the full call shape.

Composition order for runSingleFlight: Coalesce.run() outermost, Mutex.runExclusive() innermost — a joining caller shares the leader's result without ever touching the mutex itself; only the leader acquires the lock. This order is not interchangeable with mutex-first: reversing it would defeat single-flight collapsing, because every caller would separately queue for the lock before coalescing ever got a chance to join them, so coalescing would only ever see one queued caller at a time.

Mutex's timeout (queue-wait ceiling) and Coalesce's timeout (shared in-flight wait ceiling) together bound how long a caller can wait on either side, so the composition below never hangs forever on a stuck upstream call.

ts
import { Coalesce } from '@studnicky/concurrency';
import { Mutex } from '@studnicky/mutex';
import assert from 'node:assert/strict';
import { setTimeout } from 'node:timers/promises';

import { KeyedWorkGate } from '../src/index.js';

class TelemetryMutex extends Mutex<string> {
  readonly acquisitions: string[] = [];

  static override create(config = {}): TelemetryMutex {
    return new this(config);
  }

  protected override afterAcquire(key: string, waitTimeMs: number): void {
    console.log(`[mutex] acquired '${key}' after ${waitTimeMs}ms wait`);
    this.acquisitions.push(key);
  }

  protected override onEnterKey(key: string, to: 'locked' | 'queued' | 'unlocked', from: 'locked' | 'queued' | 'unlocked'): void {
    console.log(`[mutex] '${key}' ${from} -> ${to}`);
  }
}

class TelemetryCoalesce extends Coalesce<unknown> {
  readonly leaders: string[] = [];
  readonly joiners: string[] = [];

  static override create(options = {}): TelemetryCoalesce {
    return new this(options);
  }

  protected override onCoalesceStart(key: string): void {
    console.log(`[coalesce] '${key}' leader executing`);
    this.leaders.push(key);
  }

  protected override onCoalesceJoin(key: string): void {
    console.log(`[coalesce] '${key}' caller joined in-flight execution`);
    this.joiners.push(key);
  }
}

/**
 * Advanced extension: KeyedWorkGate has no hooks of its own — observability is
 * delegated entirely to the composed primitives. A subclass can still add
 * convenience behavior by reaching the composed instances through the getters.
 */
class ReportingKeyedWorkGate extends KeyedWorkGate<string> {
  // `this.create(...)` (not `KeyedWorkGate.create(...)`) so the inherited factory's
  // `new this(...)` binds to ReportingKeyedWorkGate — same `new this()` polymorphism
  // RequestExecutor/Mutex/Coalesce use for their own subclass factories.
  static tracked(mutex: TelemetryMutex, coalesce: TelemetryCoalesce): ReportingKeyedWorkGate {
    const result = this.create({ 'coalesce': coalesce, 'mutex': mutex }) as ReportingKeyedWorkGate;
    return result;
  }

  report(): { 'coalesceJoins': number; 'coalesceLeaders': number; 'mutexAcquisitions': number } {
    const mutex = this.getMutex() as TelemetryMutex;
    const coalesce = this.getCoalesce() as TelemetryCoalesce;

    return {
      'coalesceJoins': coalesce.joiners.length,
      'coalesceLeaders': coalesce.leaders.length,
      'mutexAcquisitions': mutex.acquisitions.length
    };
  }
}

KeyedWorkGate introduces no hook of its own — every observable stage is already covered by the composed primitive it delegates to, reachable through getMutex() and getCoalesce(). Every getter returns the exact instance passed to create()/builder() — never a copy or wrapper. See @studnicky/mutex and @studnicky/concurrency for each primitive's full hook table.

Hand-composing KeyedWorkGate without the package ​

ts
import { Coalesce, CoalesceTimeoutError } from '@studnicky/concurrency';
import assert from 'node:assert/strict';

import { LockTimeoutError, Mutex } from '../src/index.js';

// One Mutex/Coalesce pair per keyed resource family. `timeout` on each bounds how long a
// caller waits — Mutex's queue wait and Coalesce's shared in-flight wait — so a stuck
// upstream call cannot pin a key indefinitely.
const mutex = Mutex.create<string>({ 'timeout': 200 });
const coalesce = Coalesce.create<unknown>({ 'timeout': 100 });

// Single-flight: concurrent same-key callers collapse onto one in-flight execution, which
// itself runs under mutex-guarded exclusive access (so a non-coalesced caller reaching the
// same key via Serialized still can't interleave with it).
class SingleFlight {
  static run<T>(key: string, fn: () => Promise<T>): Promise<T> {
    const result = coalesce.run(key, () => { const result = mutex.runExclusive(key, fn); return result; }) as Promise<T>;
    return result;
  }
}

// Serialized: no coalescing — every call actually runs `fn`, but exclusively per key.
class Serialized {
  static run<T>(key: string, fn: () => Promise<T>): Promise<T> {
    const result = mutex.runExclusive(key, fn);
    return result;
  }
}

Reach for KeyedWorkGate when both runSingleFlight and runSerialized recipes for the same key repeat across call sites; hand-compose when a single call site needs only one of the two recipes, or a different Coalesce/Mutex composition order for a deliberate reason.

BoundedDispatcher: bounded work dispatch with local event coordination ​

@studnicky/bounded-dispatcher's BoundedDispatcher composes @studnicky/concurrency's Semaphore, @studnicky/event-bus's EventBus, and @studnicky/scheduler's SchedulerProviderType into a bounded-concurrency dispatch shape: dispatch() acquires a Semaphore permit before running the caller's fn, publishes 'dispatch' lifecycle events ({ phase: 'start' } before fn runs, then { phase: 'success', result } or { phase: 'error', error } after it settles) onto a composed EventBus, and releases the permit after the terminal event is published, but always before dispatch()'s returned promise settles. scheduleDispatch() layers a scheduler-driven delayed dispatch on top of the same dispatch() function, returning the scheduler's own cancellable task handle. BoundedDispatcher.create({ permits }) returns a dispatcher whose dispatch(fn) runs fn under the bounded semaphore — see the example below for the full call shape.

permits is shorthand for Semaphore.create({ permits }). bus accepts either a pre-built EventBus instance or BusQueueOptionsEntity.Type config (e.g. { highWaterMark: 4 }) passed straight to EventBus.create(). scheduler accepts a pre-built SchedulerProviderType — defaults to RealTimeScheduler.create(), or pass a VirtualScheduler for deterministic test fixtures.

EventBus.create() forwards highWaterMark into every subscriber's internal BusQueue, so the composition below tunes bus-wide backpressure for a dispatcher's lifecycle events directly through configuration, with no subclass override of internal EventBus wiring needed.

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

import { BoundedDispatcher } from '../src/index.js';

/**
 * Advanced extension: BoundedDispatcher has no hooks of its own — dispatch-level
 * observability is the `'dispatch'` topic on the composed `EventBus`, reachable
 * through `getBus()`. A subclass can still add convenience behavior by reaching
 * the composed instances through the getters.
 */
class ReportingBoundedDispatcher extends BoundedDispatcher {
  readonly #completedCount = { 'value': 0 };
  readonly #failedCount = { 'value': 0 };

  // `this.create(...)` (not `BoundedDispatcher.create(...)`) so the inherited factory's
  // `new this(...)` binds to ReportingBoundedDispatcher — same `new this()` polymorphism
  // Semaphore/EventBus/RealTimeScheduler use for their own subclass factories.
  static tracked(permits: number): ReportingBoundedDispatcher {
    const result = this.create({ 'bus': { 'highWaterMark': 4 }, 'permits': permits }) as ReportingBoundedDispatcher;
    result.getBus().subscribe('dispatch', (payload) => { result.#record(payload); });
    return result;
  }

  report(): { 'completed': number; 'failed': number } {
    return { 'completed': this.#completedCount.value, 'failed': this.#failedCount.value };
  }

  #record(payload: BoundedDispatcherEventType): void {
    if (payload.phase === 'success') { this.#completedCount.value += 1; }
    if (payload.phase === 'error') { this.#failedCount.value += 1; }
  }
}

BoundedDispatcher introduces no hook of its own. Permit-level observability stays on Semaphore's existing hooks (onAcquire, onAcquireWait, onContended, onRelease, onReleaseDelegated); dispatch-level observability is the 'dispatch' topic on the composed EventBus, reachable through getBus(), getSemaphore(), and getScheduler(). See @studnicky/concurrency, @studnicky/event-bus, and @studnicky/scheduler for each primitive's full hook table.

Hand-composing BoundedDispatcher without the package ​

ts
import { EventBus } from '@studnicky/event-bus';
import { RealTimeScheduler } from '@studnicky/scheduler';
import assert from 'node:assert/strict';

import { Semaphore } from '../src/index.js';

// json-schema-uninexpressible: 'error' is typed `unknown` — TypeScript types caught values as
// `unknown` by design (a `catch` clause may receive any thrown value, not just an Error), which
// JSON Schema cannot model. 'result' is 'string' because every dispatched task in this demo
// returns a string; the generic Dispatcher.dispatch<T> is constrained to T extends string to
// match, rather than widening the published event payload to 'unknown' for convenience.
type DispatchTopicMapType = {
  'dispatch.completed': { 'key': string; 'result': string };
  'dispatch.failed': { 'error': unknown; 'key': string; };
  'dispatch.started': { 'key': string };
};

// One Semaphore bounds how many dispatched tasks run at once; one EventBus, tuned with a
// highWaterMark now that EventBus.create() forwards it into every subscriber's BusQueue,
// publishes lifecycle events for whoever wants to observe dispatch without coupling to the
// dispatcher itself; one scheduler drives delayed dispatch.
const semaphore = Semaphore.create({ 'permits': 2 });
const bus = EventBus.create<DispatchTopicMapType>({ 'highWaterMark': 4 });
const scheduler = RealTimeScheduler.create();

class Dispatcher {
  static async dispatch<T extends string>(key: string, fn: () => Promise<T>): Promise<T> {
    const release = await semaphore.acquire();
    try {
      await bus.publish('dispatch.started', { 'key': key });
      const result = await fn();
      await bus.publish('dispatch.completed', { 'key': key, 'result': result });
      return result;
    } catch (error) {
      await bus.publish('dispatch.failed', { 'error': error, 'key': key });
      throw error;
    } finally {
      await release();
    }
  }

  static scheduleDispatch<T extends string>(atMs: number, key: string, fn: () => Promise<T>): ScheduledTaskType {
    const task = scheduler.scheduleAt(atMs, () => { void Dispatcher.dispatch(key, fn); });
    return task;
  }
}

Reach for BoundedDispatcher when the acquire-before-publish-start, release-exactly-once, publish-before-permit-frees-next-waiter wiring across the three primitives repeats across call sites — that ordering is easy to get subtly wrong by hand; hand-compose when a single call site needs a one-off variant of the semaphore/bus/scheduler wiring.

The Process Kit ​

@studnicky/process-kit's ProcessKit composes @studnicky/fsm's StateMachine and EffectInterpreter, @studnicky/scheduler's SchedulerProviderType, and @studnicky/signal's Signal into a reducer-with-effects shape: a caller-supplied StateMachine subclass owns getInitialState()/reduce() as the only source of transition logic, an internally-built EffectInterpreter runs the side effects an event produced, a scheduler drives time-delayed transitions, and a composed Signal cancels the process. ProcessKit does not implement any reducer logic itself. ProcessKit.create({ machine }) wraps a caller-built StateMachine subclass; kit.start() then kit.dispatch(event) drive it — see the example below for the full call shape.

machine is the only required field. handlers (effect handlers), scheduler, and signal are all optional and defaulted internally (RealTimeScheduler.create(), Signal.create()).

This is the pattern family nearest the Dagonizer boundary — see Composition Anti-Patterns for the specific orchestration-shaped mistakes to avoid when using or hand-composing it. ProcessKit's own README enforces three of these by convention (no runtime guard — the discipline is architectural): no chained scheduleDispatch calls that branch on the resulting state to schedule the next step; no registry/lookup of many named ProcessKit instances dispatched into by name; no save/resume checkpoint pair backed by a store. The example below respects every one of those risk flags: one flat machine, no multi-instance registry, no checkpoint/resume.

Composition notes worth calling out explicitly, because the underlying primitives' real capabilities have a sharp edge here:

  • EffectInterpreter's effect handlers receive their own (effect, dispatch) => void capability, whose dispatch(event) enqueues an event at the front of the mailbox and is only ever processed within the same drain cycle that invoked the handler — genuine same-cycle self-advance, with no extra round trip through the caller.
  • ProcessKit#dispatch(event) is a different thing entirely: it is the public, external entry point, and it always goes through the interpreter's real send() — the only path available once execution is outside that drain cycle.
  • A time-delayed transition is a different shape from either: by the time a scheduler-scheduled callback fires, the drain cycle that scheduled it has already ended, so the effect-handler dispatch capability has no reach past it. ProcessKit#scheduleDispatch(atMs, event) schedules a callback that fires from the scheduler well after any drain cycle has ended, so it correctly calls dispatch()/send(), never the effect-handler capability. The example below uses both mechanisms side by side and names the distinction rather than blurring it.
  • VirtualScheduler gives the example a deterministic, fast-running clock — no real timers, no flaky delays.
  • Cancellation composes a caller's AbortController through Signal#compose() into an AbortSignal, whose abort listener cancels the pending scheduled task and drives a cancel event into the interpreter — no bespoke cancellation plumbing.
  • TransitionRejectedError (a reducer's deliberate rejection) and MachineTerminatedError (an event sent after the machine reached a terminal state) are exercised as distinct, instanceof-checkable outcomes.
ts
import type { FsmStepType } from '@studnicky/fsm';

import { VirtualTimeCounter } from '@studnicky/clock';
import { StateMachine, TransitionRejectedError } from '@studnicky/fsm';
import { VirtualScheduler } from '@studnicky/scheduler';
import { Signal } from '@studnicky/signal';
import assert from 'node:assert/strict';

import { ProcessKit } from '../src/index.js';

// --- Domain: a job that starts, self-acknowledges in the same cycle, waits for a scheduled
// advance, then settles. reduce() stays a pure function of (state, event) throughout. ---

type JobState =
  | { readonly 'variant': 'acknowledged' }
  | { readonly 'variant': 'cancelled' }
  | { readonly 'variant': 'completed' }
  | { readonly 'variant': 'idle' }
  | { readonly 'variant': 'waiting' };

type JobEvent =
  | { readonly 'type': 'acknowledge' }
  | { readonly 'type': 'advance' }
  | { readonly 'type': 'cancel' }
  | { readonly 'type': 'start' };

type JobEffect =
  | { readonly 'delayMs': number; readonly 'variant': 'scheduleAdvance' }
  | { readonly 'variant': 'requestAck' };

class JobProcess extends StateMachine<JobState, JobEvent, JobEffect> {
  static make(): JobProcess { return new JobProcess(); }

  getInitialState(): JobState { return { 'variant': 'idle' }; }

  reduce(state: JobState, event: JobEvent): FsmStepType<JobState, JobEffect> {
    if (state.variant === 'idle' && event.type === 'start') {
      return { 'effects': [{ 'variant': 'requestAck' }], 'state': { 'variant': 'waiting' } };
    }
    if (state.variant === 'waiting' && event.type === 'acknowledge') {
      return { 'effects': [{ 'delayMs': 50, 'variant': 'scheduleAdvance' }], 'state': { 'variant': 'acknowledged' } };
    }
    if (state.variant === 'acknowledged' && event.type === 'advance') {
      return { 'effects': [], 'state': { 'variant': 'completed' } };
    }
    if ((state.variant === 'waiting' || state.variant === 'acknowledged') && event.type === 'cancel') {
      return { 'effects': [], 'state': { 'variant': 'cancelled' } };
    }
    throw new TransitionRejectedError({
      'eventType': event.type,
      'reason': `no transition defined for state '${state.variant}'`,
      'stateVariant': state.variant
    });
  }

  // Once settled, further transitions are rejected outright — reduce() is never called.
  protected override isTerminated(state: JobState): boolean {
    return state.variant === 'completed' || state.variant === 'cancelled';
  }
}

// `VirtualScheduler` gives this example a deterministic, fast clock — no real timers.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });

class Kit {
  static make(): ProcessKit<JobState, JobEvent, JobEffect> {
    const kitRef = { 'current': null as unknown as ProcessKit<JobState, JobEvent, JobEffect> };

    const requestAckFn = (_effect: JobEffect, dispatch: (event: JobEvent) => void): void => {
      dispatch({ 'type': 'acknowledge' });
    };

    const scheduleAdvanceFn = (effect: JobEffect): void => {
      kitRef.current.scheduleDispatch(counter.nowMs() + (effect as Extract<JobEffect, { 'variant': 'scheduleAdvance' }>).delayMs, { 'type': 'advance' });
    };

    kitRef.current = ProcessKit.create<JobState, JobEvent, JobEffect>({
      'handlers': {
        'requestAck': requestAckFn,
        'scheduleAdvance': scheduleAdvanceFn
      },
      'machine': JobProcess.make(),
      'scheduler': scheduler
    });

    return kitRef.current;
  }
}

// Cancellation composed via the now-instantiable Signal: an AbortSignal drives a 'cancel'
// event into the composed ProcessKit's public dispatch().
class CancellationWiring {
  static wireCancellation(kit: ProcessKit<JobState, JobEvent, JobEffect>, abortSignal: AbortSignal): void {
    abortSignal.addEventListener('abort', () => {
      void kit.dispatch({ 'type': 'cancel' });
    }, { 'once': true });
  }
}

ProcessKit introduces no hook of its own — every observable stage is already covered by the primitive it delegates to, reachable through getMachine(), getInterpreter(), getScheduler(), and getSignal(). A caller who subclassed StateMachine for its 6 lifecycle hooks (onTransition, onEnterState, onExitState, onTransitionRejected, isTerminated, onTerminatedAccess) keeps full access to those hooks; EffectInterpreter's 9 hooks and the scheduler's own hooks remain reachable the same way. See @studnicky/fsm, @studnicky/scheduler, and @studnicky/signal for each primitive's full hook table.

Hand-composing the Process Kit without the package ​

ts
import { VirtualTimeCounter } from '@studnicky/clock';
import { VirtualScheduler } from '@studnicky/scheduler';
import { Signal } from '@studnicky/signal';
import assert from 'node:assert/strict';

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

import { EffectInterpreter, MachineTerminatedError, StateMachine, TransitionRejectedError } from '../src/index.js';

// --- Domain: a job that starts, self-acknowledges in the same cycle, waits for a scheduled
// advance, then settles. Two terminal outcomes (completed, cancelled) — reduce() stays a
// pure function of (state, event) throughout. ---

type JobState =
  | { readonly 'variant': 'acknowledged' }
  | { readonly 'variant': 'cancelled' }
  | { readonly 'variant': 'completed' }
  | { readonly 'variant': 'idle' }
  | { readonly 'variant': 'waiting' };

type JobEvent =
  | { readonly 'type': 'acknowledge' }
  | { readonly 'type': 'advance' }
  | { readonly 'type': 'cancel' }
  | { readonly 'type': 'start' };

type JobEffect =
  | { readonly 'delayMs': number; readonly 'variant': 'scheduleAdvance' }
  | { readonly 'variant': 'requestAck' };

class JobProcess extends StateMachine<JobState, JobEvent, JobEffect> {
  static make(): JobProcess { return new JobProcess(); }

  getInitialState(): JobState { return { 'variant': 'idle' }; }

  reduce(state: JobState, event: JobEvent): FsmStepType<JobState, JobEffect> {
    if (state.variant === 'idle' && event.type === 'start') {
      return { 'effects': [{ 'variant': 'requestAck' }], 'state': { 'variant': 'waiting' } };
    }
    if (state.variant === 'waiting' && event.type === 'acknowledge') {
      return { 'effects': [{ 'delayMs': 50, 'variant': 'scheduleAdvance' }], 'state': { 'variant': 'acknowledged' } };
    }
    if (state.variant === 'acknowledged' && event.type === 'advance') {
      return { 'effects': [], 'state': { 'variant': 'completed' } };
    }
    if ((state.variant === 'waiting' || state.variant === 'acknowledged') && event.type === 'cancel') {
      return { 'effects': [], 'state': { 'variant': 'cancelled' } };
    }
    // Deliberate rejection (e.g. cancelling before the job ever started) — distinct from a
    // reducer defect, so callers can `instanceof`-check TransitionRejectedError.
    throw new TransitionRejectedError({
      'eventType': event.type,
      'reason': `no transition defined for state '${state.variant}'`,
      'stateVariant': state.variant
    });
  }

  // Once settled, further transitions are rejected outright — reduce() is never called.
  protected override isTerminated(state: JobState): boolean {
    return state.variant === 'completed' || state.variant === 'cancelled';
  }
}

// `VirtualScheduler` gives this example a deterministic, fast clock — no real timers.
const counter = VirtualTimeCounter.create({ 'startMs': 0 });
const scheduler = VirtualScheduler.create({ 'counter': counter });

class Job {
  static make(): {
    'getScheduledTask': () => ScheduledTaskType | undefined;
    'interpreter': EffectInterpreter<JobState, JobEvent, JobEffect>;
  } {
    const scheduledTaskRef: { 'current': ScheduledTaskType | undefined } = { 'current': undefined };
    const interpreterRef = { 'current': null as unknown as EffectInterpreter<JobState, JobEvent, JobEffect> };

    const requestAckFn = (_effect: JobEffect, dispatch: (event: JobEvent) => void): void => {
      dispatch({ 'type': 'acknowledge' });
    };

    const scheduleAdvanceFn = (effect: JobEffect): void => {
      scheduledTaskRef.current = scheduler.scheduleAt(counter.nowMs() + (effect as Extract<JobEffect, { 'variant': 'scheduleAdvance' }>).delayMs, () => {
        scheduledTaskRef.current = undefined;
        void interpreterRef.current.send({ 'type': 'advance' });
      });
    };

    interpreterRef.current = EffectInterpreter.create<JobState, JobEvent, JobEffect>({
      'handlers': {
        'requestAck': requestAckFn,
        'scheduleAdvance': scheduleAdvanceFn
      },
      'machine': JobProcess.make()
    });

    const getScheduledTaskFn = (): ScheduledTaskType | undefined => {
      const result = scheduledTaskRef.current;
      return result;
    };

    return {
      'getScheduledTask': getScheduledTaskFn,
      'interpreter': interpreterRef.current
    };
  }
}

// Cancellation composed via the now-instantiable Signal: an AbortSignal drives a 'cancel'
// event into the interpreter and cancels any still-pending scheduled advance.
class CancellationWiring {
  static wire(
    interpreter: EffectInterpreter<JobState, JobEvent, JobEffect>,
    getScheduledTask: () => ScheduledTaskType | undefined,
    abortSignal: AbortSignal
  ): void {
    abortSignal.addEventListener('abort', () => {
      const pending = getScheduledTask();
      if (pending !== undefined) { pending.cancel(); }
      void interpreter.send({ 'type': 'cancel' });
    }, { 'once': true });
  }
}

Reach for ProcessKit when the StateMachine+EffectInterpreter+scheduler+Signal wiring repeats across call sites and the dispatch()/scheduleDispatch() split matters; hand-compose when a single call site needs a one-off variant of the wiring, or when pulling in @studnicky/process-kit as a dependency isn't worth it for a single reducer.