processing-tools

Processing Tools

A TypeScript library of building blocks for data-processing and reconciliation pipelines: throttling, resource allocation, dependency-ordered task execution, hierarchical configuration stores, and a full model-driven "current vs. desired state" reconciliation engine.

npm install processing-tools

import { RateThrottler, DependencyProcessor, ConfigStore /* ... */ } from 'processing-tools';

All components are logger-agnostic. Almost every class takes an ILogger as its first constructor argument — you supply the implementation, the library never picks one for you.

---

Table of Contents

• Throttlers — cap concurrency or rate of async work
• Allocators — hand out port blocks and CIDR subnets
• Dependency Resolver — topological ordering
• Dependency Processor — run tasks respecting dependencies, labels, and readiness
• Config Store — hierarchical key/value with inheritance
• Repo Store — named nested repositories with persistence and dirty-processing
• Environment Tools — merge and substitute env dictionaries
• Config Reconciliation Engine — the model/delta/processor stack
• Logger Contract

---

Throttlers

Limit how async actions run. Each execute() call queues an action and returns a promise that resolves with the action's result once it is allowed to run and completes.

ConcurrentThrottler

Caps the number of actions running at the same time.

import { ConcurrentThrottler } from 'processing-tools';

const throttler = new ConcurrentThrottler(logger, { number: 4 }); // at most 4 in flight

const results = await Promise.all(
    urls.map((url) => throttler.execute(() => fetch(url), `fetch:${url}`)),
);

RateThrottler

Caps how many actions start within a sliding time window. Use it to respect rate limits (e.g. "no more than 10 requests per second").

import { RateThrottler } from 'processing-tools';

// at most 10 starts per 1000ms
const throttler = new RateThrottler(logger, { number: 10, interval: 1000 });

await throttler.execute(() => callApi(), 'api-call');

interval is in milliseconds. When the window is full the throttler schedules a timer and drains the queue as slots free up.

BaseThrottler

The shared base class. Subclass it and implement _canRun() (and optionally _onActionStart() / _processTaskChange()) to build a custom admission policy. Exposes:

• execute<T>(action: () => Promise<T> | T, name: string): Promise<T>
• hasWaitingActions: boolean

---

Allocators

PortAllocator

Hands out fixed-size, non-overlapping blocks of ports from one or more free ranges. Allocations are sticky: asking again for the same (service, sourcePort) returns the same block.

import { PortAllocator } from 'processing-tools';

const allocator = new PortAllocator(logger, 100); // block size of 100 ports
allocator.addFreeRange(30000, 40000);

allocator.reserve('web', 443, 30500);           // pin a specific block
const block = allocator.allocate('web', 8080);   // -> { start, end } or null when exhausted
// block === { start: 30100, end: 30199 }  (example)

allocator.output(); // logs all current reservations

SubnetAllocator

Carves CIDR subnets out of a larger range using a buddy-style split. Reserve subnets that are already in use, then allocate new ones by prefix length.

import { SubnetAllocator } from 'processing-tools';

const allocator = new SubnetAllocator('10.0.0.0/16');
allocator.reserve('10.0.1.0/24');          // mark as used

const cidr = allocator.allocate(24);        // -> '10.0.0.0/24' (or null if none fit)

allocate(maskLength) returns the smallest fitting free block, splitting larger free subnets as needed, or null when the range is exhausted.

---

Dependency Resolver

Lightweight topological sort. Register client → supplier edges, read back a valid processing order. Self-edges are ignored and clients with no edges are still included.

import { DependencyResolver } from 'processing-tools';

const resolver = new DependencyResolver();
resolver.add('app', 'database'); // app depends on database
resolver.add('app', 'cache');
resolver.add('cache');           // no dependency

console.log(resolver.order); // suppliers first, e.g. ['database', 'cache', 'app']

---

Dependency Processor

An asynchronous engine that runs a set of tasks while honouring dependencies, concurrency labels, and readiness checks. It is the workhorse behind the reconciliation engine, but is usable on its own.

You provide a handler(id) => boolean | Promise<boolean>. Returning false marks a task unqualified and skips everything that depends on it; throwing fails the run.

import { DependencyProcessor, TaskState } from 'processing-tools';

const processor = new DependencyProcessor(logger, 'my-run', async (id) => {
    await doWork(id);
    return true; // false => unqualified (dependents skipped)
});

processor.addTask('database');
processor.addTask('app');
processor.setDependency('app', 'database'); // app runs only after database completes

await processor.process(); // resolves when all done, rejects if any task failed

Capabilities:

Method	Purpose
addTask(id)	Register a task (id is any JSON-serializable value).
setDependency(id, predecessorId)	id waits for predecessorId to reach Complete.
setLabel(id, label)	Tag a task (used for tracking / non-concurrency).
setNonConcurrentLabels(id, labels)	Prevent tasks sharing a label from running simultaneously.
setPreRunChecker(id, () => boolean)	Gate a task; false marks it unqualified before it starts.
setCompletionChecker(id, cb)	Poll for readiness after the handler resolves — return { ready }, or { retry: true, timeout: <seconds> } to re-check later.
process()	Run to completion. Resolves on success, rejects if any task failed.
taskErrors	Collected TaskErrorInfo for failed tasks.
tasksByState	Map of TaskState → tasks, for reporting.
close()	Stop processing and clear the internal health-check timer.

Task lifecycle states (TaskState): Idle, Running, WaitingFinish, Complete, Error, Unqualified, Skipped.

---

Config Store

A hierarchical key/value store with inheritance along a path. Values set at a shallow path act as defaults; values set deeper override them. resolveValue walks the path as far as it can and returns the deepest value found, falling back toward the root.

import { ConfigStore } from 'processing-tools';

const store = new ConfigStore(logger);
store.setValue([], 'timeout', 30);                    // global default
store.setValue(['prod'], 'timeout', 60);              // override for prod
store.setValue(['prod', 'db'], 'ssl', 'true');

store.resolveValue(['prod', 'db'], 'timeout');        // 60 (inherited from 'prod')
store.resolveValue(['staging'], 'timeout');           // 30 (global default)
store.resolveBoolValue(['prod', 'db'], 'ssl');        // true ('true'/'yes' => true)

---

Repo Store

Manages several named repositories of nested dictionaries, with optional file persistence and a "dirty processing" workflow for reacting to changes.

import { RepoStore } from 'processing-tools';

const store = new RepoStore(logger, 'services');
store.setupPersistence('./state');

store.setupRepository('endpoints')
    .description('SERVICE ENDPOINTS')
    .handleDirty((service) => rebuildService(service), 1); // reprocess 1 level deep

store.set('endpoints', ['web', 'url'], 'https://example.com');
const url = store.get('endpoints', ['web', 'url']);   // read; null if missing
const node = store.at('endpoints', ['web']);          // read-or-create nested node

await store.persistStore(); // writes changed, persistable repos to ./state/*.json

Nested access: at (get-or-create), get (read, null if absent), set, delete, and loop(name, keyPath, cb) for serial iteration over a level.

Dirty processing lets you register a processor per repository and then:

• markDirtyRepo(name, path) — process immediately, or defer if the path is suppressed.
• markRepoSuppressProcess(name, path, delay) — suppress (delay = true) to batch changes, then release (delay = false) to flush everything that accumulated.
• unmarkDirtyRepo(name, path) — clear a pending dirty mark.

Repositories flagged markDoNotPersist() are kept in memory only; persistStore() writes the rest to <dir>/<name>.json and skips repos that haven't changed.

---

Environment Tools

Helpers for composing environment-variable dictionaries (EnvDict = Record<string, string>) and sets of them keyed by name (EnvSet, where a global entry acts as the shared base).

import {
    mergeEnvironment,
    resolveEnvironmentSet,
    mergeEnvironmentSets,
    substituteEnvironment,
} from 'processing-tools';

// active overrides base
mergeEnvironment({ HOST: 'a', PORT: '80' }, { HOST: 'b' });
// => { HOST: 'b', PORT: '80' }

// fold the `global` env into every named env
resolveEnvironmentSet({
    global: { REGION: 'us' },
    prod: { TIER: 'high' },
});
// => { global: { REGION: 'us' }, prod: { REGION: 'us', TIER: 'high' } }

// layer one set over another (both resolved against their globals first)
mergeEnvironmentSets(baseSet, overrideSet);

// expand ${VAR} references
substituteEnvironment({ URL: '${HOST}:${PORT}' }, { HOST: 'db', PORT: '5432' });
// => { URL: 'db:5432' }

---

Config Reconciliation Engine

The largest capability: a framework for driving real-world resources toward a desired state by computing and applying a delta between current and desired configurations. It is built from several cooperating pieces (all exported from the package root and from processing-tools config types).

Concepts

• ConfigMeta / ConfigSectionMeta — the schema. A section describes one kind of resource and registers callbacks via a fluent API: how to list it (onQueryAll), how to read identity/config/runtime/relations off a raw object (onExtractId, onExtractConfig, onExtractRuntime, onExtractRelations), and how to change it (onCreate, onUpdate, onDelete, onRelationCreate, onRelationDelete). Additional hooks cover readiness (onCheckReady), auto-configuration (onAutoConfig), delta suppression (onCheckIgnoreDelta), and forced recreation (onUpdateRecreate). ConfigMeta.load(paths, logger, context) loads one metadata module per file from a directory.
• Config — a populated model: ConfigSections holding ConfigItems plus the ConfigRelations between them. extract() builds the current model by querying every section; a desired Config is constructed against the current one for resolution. produceDelta(base) yields a DeltaDict.
• ConfigItem / ConfigRelation — a single resource (identified by a dn like section://naming) and a typed link to another item. Relations drive dependency ordering and can auto-create their targets.
• DeltaProcessor — takes current + desired Config, computes the delta, and applies it through a DependencyProcessor: creates items in relation order, deletes in reverse, updates in place, and recreates where required. Returns a DeltaProcessResult with per-task error and count breakdowns.
• ModelProcessor — orchestrates a full iteration: extract current → build desired → auto-config → compute and output delta → process delta → decide whether another iteration is needed. Extend it, wire section metadata and stage callbacks, and drive the pipeline via runStage('process-iteration').

Delta types

produceDelta classifies each item with a DeltaItemStatus — Create, Update, Delete, or Recreate — and details property-level (ConfigPropertyDelta) and relation-level (RelationDelta) changes, each carrying a DeltaState (Create / Update / Delete).

Sketch

import { ConfigMeta, Config, DeltaProcessor } from 'processing-tools';

// 1. Load schema (one file per section under ./models)
const meta = ConfigMeta.load(['./models'], logger, context);

// 2. Extract the live/current state
const current = new Config(meta);
await current.extract();

// 3. Build the desired state against current
const desired = new Config(meta, current);
// ...populate desired sections/items/relations...
await desired.performAutoConfig();

// 4. Compute and apply the delta
const processor = new DeltaProcessor(logger, current, desired);
const result = await processor.process();
// result.failedTaskCount, result.skippedTaskCount, result.taskErrors, ...

For most uses you subclass ModelProcessor, which runs this loop for you and re-iterates until the model converges.

---

Logger Contract

The library depends on an injected logger rather than shipping one. Provide an object implementing ILogger:

interface ILogger {
    info(...args: any[]): void;
    warn(...args: any[]): void;
    error(...args: any[]): void;
    crit(...args: any[]): void;
    verbose(...args: any[]): void;
    debug(...args: any[]): void;
    silly(...args: any[]): void;
    exception(error: any): void;
    sublogger(name: string): ILogger;          // scoped child logger
    outputFile(fileName: string, data: any): any;
    outputStream(fileName: string): IOutputWriter | null;
}

sublogger lets components create named child loggers; outputFile / outputStream (returning an IOutputWriter with write / writeHeader / indent / unindent / close) are used for debug dumps of configs and deltas.