@nfolds/chains-flow-core

v0.2.0

Published

2 months ago

Middleware-style execution framework for Node.js: chains of small components share a context, primitives stay declarative, transformers hold the logic. First-class support for per-chain transactions across heterogeneous resources, saga compensators, confi

flow-core

A middleware-style execution framework for Node.js. Chains of small declarative components share a context, primitives stay configuration-only, transformers hold the logic. Built-in support for best-effort distributed transactions, saga compensators, parallel/series composition, typed metrics, and structured logging — with zero runtime dependencies.

npm install flow-core

Mental model

A flow is a directed graph of small named components.

Logic lives in Transformer and CustomExecutor — anywhere you'd write arbitrary JavaScript.
I/O lives in connectors — MongoConnector, HttpConnector, etc. They are declarative: configure which ctx key supplies the input and which ctx key receives the output, nothing else.
Control flow lives in Conditional, Series, Parallel, onError, EXIT, and CustomExecutor.

Components communicate by reading and writing ctx — a plain JS object shared across every step in a single engine.run(). Components don't return values to each other; they mutate ctx. This makes inserting / removing / reordering a step a one-line change.

A chain is an ordered list of components. Chains can jump to other chains (transfer control) or be invoked as sub-flows (call-and-return). Cross-chain state is just ctx.

            ┌──────────────────────── ctx (shared dict) ────────────────────────┐
            │                                                                   │
   ┌────────▼────────┐  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐    │
   │   RequireVars   │─►│  Transformer │─►│  MongoWriter │─►│ KafkaPublisher│  ──┘
   │ (validate keys) │  │  (build doc) │  │ (insertOne)  │  │   (publish)  │
   └─────────────────┘  └──────────────┘  └──────────────┘  └──────────────┘
                                          │ (registers     │
                                          │ compensator)   │ (registers
                                                           │ compensator)
                                          │ ←── LIFO ──── ←┘
                                              on failure

Five things every component shares (via the Component base class):

| Option | Purpose | |---------------|--------------------------------------------------------------------------------------------------| | name | Human label used in logs/metrics. Defaults to the class name. | | onError | A Chain (or (ctx) => Chain) to jump to if this component throws. | | compensate | A function, Chain, or (ctx) => Chain to register as an inverse op when this component succeeds. | | metrics | A MetricSpec, an array of them, or a handler function. Component duration is auto-timed. | | run(ctx) | The subclass's behaviour. Either no return (continue), { jump }, or { exit: true }. |

Hello world

const { Engine, Chain, Transformer } = require('flow-core');

const greet = new Chain('greet', [
  new Transformer({ name: 'capitalise', fn: (ctx) => {
    ctx.greeting = `Hello, ${ctx.name[0].toUpperCase() + ctx.name.slice(1)}!`;
  } }),
]);

(async () => {
  const out = await new Engine().run(greet, { name: 'ada' });
  console.log(out.greeting); // → Hello, Ada!
})();

Three things to notice:

Components are constructed once. engine.run is called per request/event.
Initial state goes in as the second argument to run. Whatever's on ctx at the end is the result.
Components don't return data — they mutate ctx.

What's new in 0.2.0

Driven entirely by feedback from the first real consumer that ported a 9-endpoint production service onto chains-flow-core. Every item below answers a friction point that showed up in code review of that port. All changes are backward compatible — existing 0.1.x callers see no behaviour difference.

Critical correctness

A1 — Nested chain compensator preservation. Engine no longer wipes ctx._compensators at the end of inner engine.run calls. Saga compensators registered by outer chains used to be silently dropped if any sub-chain ran successfully. The wipe now happens only at the outermost engine.run.
A2 — Better error for fn-returning-fn compensators. Writing compensate: (ctx) => async () => { ... } (a function that returns a function) now throws at compensator-invoke time with a copy-paste fix, instead of failing late with an opaque ctx._engine is not set.
F3 — Constructor-time error for Chain-instance compensators. Passing compensate: new Chain(...) now throws at construction with a hint to wrap in a factory () => new Chain(...).

New components

const {
  AcquireLock, ReleaseLock,
  LoadState, PersistState, StateAdvancedError,
  BreakerRegistry,
} = require('@nfolds/chains-flow-core');

B1 — Redis SETNX distributed lock

new Chain('reprice', [
  new Transformer({ fn: (ctx) => { ctx.lockKey = `lock:${ctx.bookingId}`; } }),
  new AcquireLock({
    client: redis, keyVar: 'lockKey', ttlSec: 120,
    onAlreadyLocked: 'conflict',           // ctx.status when contention
    compensate: () => new Chain('rel', [   // safety: release on saga abort
      new ReleaseLock({ client: redis, keyVar: 'lockKey' }),
    ]),
  }),
  new Conditional({ branches: [[(c) => !c._lockAcquired, EXIT]] }),
  // ... saga steps ...
  new ReleaseLock({ client: redis, keyVar: 'lockKey' }),
]);

Lua EVAL atomic owner-check delete (with GET+DEL fallback). Never deletes a foreign worker's lock even if our TTL expired and someone else grabbed the key.

B2 / B3 — State-machine LoadState + PersistState (CAS, monotonic version)

new Chain('advance', [
  new LoadState({
    cache: { client: redis, keyVar: 'stateKey' },     // optional hot layer
    store: { read: async (ctx) =>
      mongo.collection('jobs').findOne({ _id: ctx.jobId }) },
    outputVar: 'job',                                  // ctx.job, ctx.jobState, ctx.jobVersion
  }),
  new PersistState({
    store: { write: async (ctx, { expected, version, newState, extra }) =>
      mongo.collection('jobs').findOneAndUpdate(
        { _id: ctx.jobId, processing_state: expected, processing_version: version },
        { $set: { processing_state: newState, processing_version: version + 1, ...extra } },
        { returnDocument: 'after' },
      ) },
    cache: { client: redis, keyVar: 'stateKey', ttlSec: 3600 },
    newState: 'PRICED',
    expectedStateVar: 'jobState',
    versionVar: 'jobVersion',
    onError: handleCASMiss,                            // catches StateAdvancedError
  }),
]);

CAS-fail throws StateAdvancedError (err.code === 'STATE_ADVANCED') so a chain-level onError can re-load + resume cleanly. Cache write is best-effort: a Redis outage logs a warning, doesn't fail the transition.

Connector ergonomics

C1 — ElasticsearchConnector pagination + aggregations

new ElasticsearchConnector({
  client: esClient, index: 'flights',
  queryVar: 'q',
  sizeVar: 'pageSize', fromVar: 'offset',
  sortVar: 'sort', aggsVar: 'aggs',
  source: ['origin', 'destination', 'price'],
  trackTotalHits: true,
  outputVar: 'hits',
});
// → ctx.hits, ctx.hitsTotal, ctx.hitsAggs (siblings auto-exposed)
// Escape hatch: requestVar overrides the entire request body.

C2 — HttpConnector defensive header iteration. Now tolerates plain-object header stubs in test mocks (no more "headers.forEach is not a function" / "headers.get is not a function" surprises).

C3 — HttpConnector inline retry + breaker

// Before (still works):
new Retry({ step: new CircuitBreaker({ step: new HttpConnector({ ... }) }) });

// After:
new HttpConnector({
  url: 'https://partner.example/search',
  outputVar: 'partnerResp',
  retry:   { attempts: 3, backoff: 'exponential', delayMs: 100 },
  breaker: { failureThreshold: 5, openMs: 30_000 },
});

C4 — BreakerRegistry — shared breaker state across call sites

const reg = new BreakerRegistry();
reg.declare('promo', { failureThreshold: 5, openMs: 30_000 });

// Both call sites trip + close together, because they share the same upstream.
new CircuitBreaker({ step: productCall,  registry: reg, upstreamName: 'promo' });
new CircuitBreaker({ step: categoryCall, registry: reg, upstreamName: 'promo' });

Omit registry/upstreamName → per-instance state (today's behaviour).

C5 — CustomConnector fire-and-forget

new CustomConnector({
  client: kafkaProducer,
  action: (k, ctx) => k.send({ topic: 'audit', messages: [{ value: ctx.event }] }),
  outputVar: false,        // explicit skip — side effect only
});

Discoverability

F1 — EVENTS constants module. Lifecycle event names exported as frozen constants so consumers can hook the engine logger with autocomplete

refactor safety instead of string-matching 'engine.auto_abort_on_exit' by hand.

const { EVENTS } = require('@nfolds/chains-flow-core');

function engineLogger(rec) {
  switch (rec.event) {
    case EVENTS.LIFECYCLE.COMPONENT_ENTER: span.startChild(rec.component); break;
    case EVENTS.LIFECYCLE.COMPONENT_EXIT:  span.endChild(); break;
    case EVENTS.LIFECYCLE.ENGINE_AUTO_ABORT: metrics.increment('chains.aborted'); break;
  }
}

F2 — Conditional jump-vs-invoke recipe. Louder callout in the Conditional section above; sub-flow vs control-transfer is now the very first thing readers see.

Upgrade notes

npm install @nfolds/[email protected]

No code changes required. New components are opt-in; existing imports continue to work unchanged.

Core types

Context (`ctx`)

A plain JS object. Components read keys, write keys. The framework reserves a few underscored keys for itself — don't write to these directly:

| Reserved key | Set by | Meaning | |-------------------------|-------------------------------------------------|------------------------------------------------------------------| | _engine | Engine.run | Reference to the engine so sub-chains can recursively run | | _metricsAdapter | Engine.run | Adapter every component dispatches metrics through | | _logger | User code (typically request middleware) | Default Logger instance for Log components | | _participants | Begin*Transaction | Open transactional participants, per chain scope | | _compensators | Component base | Saga compensators registered after successful runs (LIFO list) | | _lastError | Component.execute on catch | The most recent error | | _lastErrorFrom | Same | The component name that errored | | _committed / _aborted | CommitAll / AbortAll / CommitTransaction | What just settled | | _compensated | RunCompensators | { succeeded: [...], failed: [...] } | | _cleared | ClearCompensator | Compensator names that were removed | | _autoCommittedOnExit | Engine auto-settle (success) | Participants the engine committed for you | | _autoAbortedOnExit | Engine auto-settle (error-jump) | Participants the engine aborted for you | | _parallelFailures | Parallel({ failFast: false }) | [{ branch, error }] of branches that rejected |

Everything else is yours.

`Component`

You rarely subclass this directly — but every option flows through this base.

const { Component } = require('flow-core');

class MyThing extends Component {
  constructor(opts = {}) {
    super({ ...opts, name: opts.name || 'MyThing' });
    this.cfg = opts.cfg;
  }
  async run(ctx) {
    ctx.didIt = true;
    // Return undefined to continue, { jump: chain } to jump,
    // { exit: true } to end the current chain.
  }
}

Pass the full opts to super(opts) so onError, compensate, and metrics are auto-propagated. If you forget, your subclass will silently ignore those features.

`Chain`

A named list of executables.

const c = new Chain('checkout', [
  new RequireVars({ vars: ['cartId'], onError: errorChain }),
  new MongoConnector({ /* ... */ }),
  // ... etc
], {
  inputValidator:  (ctx) => Boolean(ctx.cartId),       // optional
  outputValidator: (ctx) => Boolean(ctx.result),       // optional
});

| Argument | Type | |-----------------------------|---------------------------------| | name (or array) | string (or the components array if you skip a name) | | components | array of executables | | opts.inputValidator(ctx) | runs at chain entry | | opts.outputValidator(ctx) | runs only at natural exit |

A Chain has an execute(ctx) method so it can stand in anywhere a Component is accepted — see Chain as component.

`Engine`

const engine = new Engine({
  logger:        (e) => { /* receives lifecycle events */ },
  metricsAdapter: somePrometheusAdapter,
});

const ctx = await engine.run(chain, { /* initial state */ });

Lifecycle events the logger callback receives:

| event | When | |-----------------------------|-----------------------------------------------------------------| | chain.enter | Entering a chain | | component.enter / exit | Around each component | | chain.exit_requested | A component returned { exit: true } | | engine.auto_commit_on_exit| Auto-commit fired at natural chain exit | | engine.auto_abort_on_exit | Auto-abort fired at error-jump chain exit | | engine.auto_compensate | Uncaught error → engine ran registered compensators | | engine.auto_abort | Uncaught error → engine aborted open participants |

This is a plain function, not a logger object. Wire it to whatever logger you use.

Control flow

Component results

What Component.run(ctx) may return:

| Return | Engine effect | |-------------------------------------|-------------------------------------------------------------------| | undefined | Continue to next component | | { jump: targetChain } | Abandon current chain, run targetChain next | | { jump: target, viaError: true } | Same as above, but flagged as error → chain's open tx abort | | { exit: true } | End the current chain right here; auto-commit open tx |

Components don't usually craft these by hand. Conditional returns { jump }, onError is converted by the base Component, EXIT and CustomExecutor helpers produce them.

`Conditional` jump vs invoke

The same Conditional can either jump to a target chain (default; control transfers permanently) or invoke it (target runs as a sub-flow and control returns to the outer chain).

⚠️ Pick invoke when steps remain after the conditional. A common mis-use: writing mode: 'jump' (the default) when you actually need cleanup steps — lock release, span end, log emit, response build — to run after the branch. jump abandons the outer chain, so anything below the conditional is silently skipped.
// ❌ WRONG — `release` and `respond` never run when the predicate fires.
new Chain('book', [
  acquireLock,
  new Conditional({ mode: 'jump', branches: [[isPremium, premiumFlow]] }),
  release,
  respond,
]);

// ✅ RIGHT — premiumFlow runs as a sub-flow, control returns, cleanup fires.
new Chain('book', [
  acquireLock,
  new Conditional({ mode: 'invoke', branches: [[isPremium, premiumFlow]] }),
  release,
  respond,
]);
Rule of thumb: jump when the chain's job ends at the branch. invoke when the chain still has work to do afterwards.

const { Conditional, EXIT } = require('flow-core');

new Conditional({
  branches: [                                  // default mode: 'jump'
    [(ctx) => ctx.cart.empty, emptyCartChain], // → goes to emptyCartChain, never returns
    [(ctx) => ctx.cart.fraud, fraudChain],
  ],
});

new Conditional({
  mode: 'invoke',                              // runs target then continues
  branches: [
    [(ctx) => ctx.kind === 'premium', applyPremiumDiscountChain],
    [(ctx) => ctx.kind === 'student', applyStudentDiscountChain],
  ],
});

new Conditional({
  branches: [
    [(ctx) => ctx.done, EXIT],                 // end this chain here (any mode)
  ],
});

| Branch target form | Notes | |---------------------------------------|--------------------------------------------------------| | Chain instance | Most common | | Component (Series, Parallel, …) | Anything with .execute(ctx) | | Array | Treated as new Series({ steps: target }) | | (ctx) => one of the above | Dynamic — evaluated at predicate-match time | | EXIT | Special sentinel: end current chain immediately |

Quick worked example:

const { Engine, Chain, Transformer, Conditional, EXIT } = require('flow-core');

const dispatched = new Chain('dispatched', [
  new Transformer({ fn: (ctx) => { ctx.trace.push('dispatched'); } }),
]);

const main = new Chain('main', [
  new Transformer({ fn: (ctx) => { ctx.trace = ['start']; } }),

  // Jump mode: if mode === 'jump', leave this chain for `dispatched`.
  new Conditional({
    branches: [[(ctx) => ctx.mode === 'jump', dispatched]],
  }),

  new Transformer({ fn: (ctx) => { ctx.trace.push('between'); } }),

  // Invoke mode: run dispatched as a sub-flow, then continue.
  new Conditional({
    mode: 'invoke',
    branches: [[(ctx) => ctx.mode === 'invoke', dispatched]],
  }),

  // Exit at will.
  new Conditional({
    branches: [[(ctx) => ctx.mode === 'exit', EXIT]],
  }),

  new Transformer({ fn: (ctx) => { ctx.trace.push('end'); } }),
]);

(async () => {
  for (const mode of ['jump', 'invoke', 'exit', 'fallthrough']) {
    const out = await new Engine().run(main, { mode });
    console.log(mode.padEnd(11), out.trace);
  }
})();
// jump        [ 'start', 'dispatched' ]
// invoke      [ 'start', 'between', 'dispatched', 'end' ]
// exit        [ 'start', 'between' ]
// fallthrough [ 'start', 'between', 'end' ]

`EXIT` sentinel

EXIT ends the current chain right where it appears, without erroring. Auto-settlement treats this as a clean exit: open transactions commit, the chain's outputValidator is skipped (output may not be fully formed since we stopped early), no nextChain is followed.

const { EXIT } = require('flow-core');

new Conditional({
  branches: [
    [(ctx) => ctx.alreadyProcessed, EXIT],   // idempotency short-circuit
    [(ctx) => ctx.duplicateKey, EXIT],
  ],
});

`Series` and `Parallel`

Both accept executables (Component, Chain, another Series/Parallel). The enclosing chain continues after they finish.

const { Series, Parallel, Engine, Chain, Transformer } = require('flow-core');

const ordered = new Series({
  steps: [chainA, chainB, transformer],
});

const fannedOut = new Parallel({
  branches: [chainA, chainB, chainC],
  failFast: false,                  // wait for everything; record failures
});

const main = new Chain('main', [
  ordered,
  fannedOut,
  new Transformer({ fn: (ctx) => { ctx.afterFanout = true; } }),
]);

| Construct | failFast | Outcome | |-----------|----------|-------------------------------------------------------------------------| | Series | n/a | Stops on first thrown error; throws out | | Parallel| true (default) | Promise.all — first rejection wins; siblings keep running but their result is dropped | | Parallel| false | Promise.allSettled — wait for all; failures collected on ctx._parallelFailures. The Parallel itself only throws if every branch failed. |

Important warning on Parallel + shared ctx. All branches share the same ctx object. If two branches write to the same key, the result is racy. Either give each branch its own output key, or merge results in a follow-up Transformer.

Chain as component

A Chain can sit anywhere a Component does. Internally it dispatches via ctx._engine.run(this, ctx) (so participants/scopes are isolated as expected — see Per-chain scoping).

const inner = new Chain('charge', [
  new Transformer({ fn: async (ctx) => { ctx.charge = await paymentApi.charge(ctx.amount); } }),
]);

const outer = new Chain('checkout', [
  new Transformer({ fn: (ctx) => { ctx.starting = true; } }),
  inner,                                              // ← runs, then returns
  new Transformer({ fn: (ctx) => { ctx.done = true; } }),
]);

`CustomExecutor`

When a job doesn't fit the declarative model, this is the escape hatch.

const { CustomExecutor } = require('flow-core');

new CustomExecutor({
  name: 'dispatch',
  execute: async (ctx, helpers) => {
    // SYNCHRONOUS sub-flow (control returns when done):
    await helpers.runChain(reservationChain);

    // SEQUENTIAL / CONCURRENT batches of executables:
    await helpers.runSeries([stepA, stepB]);
    await helpers.runParallel([stepX, stepY], /* failFast */ true);

    // JUMP — abandon current chain:
    if (ctx.broken) return helpers.jump(errorChain);

    // EXIT — end current chain right here:
    if (ctx.done) return helpers.exit();

    // Otherwise fall through and the outer chain continues.
  },
});

helpers:

| Helper | Behaviour | |-------------------------------------|------------------------------------------------------------------------| | engine | Engine instance | | runChain(chain) | Sub-flow on same ctx; returns when the chain finishes | | runSeries(steps) | Sequential sub-flow | | runParallel(branches, failFast?) | Concurrent sub-flow | | jump(target) | Returns a marker — executor's .run() converts to { jump: target } | | exit() | Marker — converts to { exit: true } | | logger | ctx._logger (may be undefined) | | metricsAdapter | ctx._metricsAdapter (may be undefined) |

`onError` chains

Every component takes onError: Chain | (ctx) => Chain. When run() throws, the engine sets ctx._lastError and jumps to onError (with the error-jump flag — open transactions in the current chain will auto-abort).

const errors = new Chain('errors', [
  new Log({ level: 'error', message: 'flow failed', fields: ['_lastError'] }),
  new Transformer({ fn: (ctx) => { ctx.status = 'failed'; ctx.error = ctx._lastError.message; } }),
]);

new Chain('main', [
  new MongoConnector({ /* ... */, onError: errors }),
  new KafkaPublisher({ /* ... */, onError: errors }),
]);

If a component has no onError and nothing higher up handles it, the engine runs any registered compensators (LIFO), aborts any open participants, and re-throws to the caller of engine.run().

Logic components

`Transformer`

Where you put JavaScript.

const { Transformer } = require('flow-core');

new Transformer({
  name: 'computeTotal',
  fn: (ctx) => { ctx.total = ctx.items.reduce((s, i) => s + i.price * i.qty, 0); },
});

new Transformer({
  name: 'asyncFetch',
  fn: async (ctx) => { ctx.user = await db.users.findOne({ id: ctx.userId }); },
});

Options: name, fn (required), onError, compensate, metrics.

`RequireVars`

Fail fast if expected ctx keys are missing.

new RequireVars({
  vars: ['userId', 'amount'],
  onError: badInputChain,
});

Throws RequireVars: missing ctx keys: userId, amount — propagates as a normal error and gets caught by onError. Use this at chain entry to guarantee what later components can rely on.

`Validator`

Library-agnostic. Accept any function that follows one of the supported result shapes. Tested against ajv, joi, and zod patterns out of the box.

const { Validator } = require('flow-core');

// Plain predicate
new Validator({ targetVar: 'amount', validate: (n) => n > 0 });

// Boolean+errors (ajv style)
const Ajv = require('ajv');
const schema = new Ajv().compile({
  type: 'object',
  required: ['email'],
  properties: { email: { type: 'string', format: 'email' } },
});
new Validator({ targetVar: 'user', validate: schema, onError: badInput });

// {valid, errors} shape
new Validator({
  targetVar: 'phone',
  validate: (s) => /^\+?\d{10,}$/.test(s)
    ? true
    : { valid: false, errors: ['must be E.164'] },
});

// Throws (also fine — bubbles through onError)
new Validator({
  targetVar: 'user',
  validate: (u) => { if (!u.email) throw new Error('email required'); },
});

Accepted return shapes from your validate function:

| You return | Means | |-------------------------------------|-------| | true / undefined / void | Pass | | false | Fail (uses .errors if attached, ajv style) | | { valid: false, errors: [...] } | Fail with reported errors | | { valid: true } | Pass | | { error: undefined, value: ... } | Pass (joi-style success) | | throws | Fail with the thrown error |

`Log`

Drop-in structured log entry inside a chain.

const { Logger, Log } = require('flow-core');

const logger = new Logger({
  name: 'svc',
  level: 'info',
  contextFields: ['requestId', 'traceId'],   // auto-included on every emit
});

new Chain('handle', [
  // Make the logger available on ctx (typically done in HTTP middleware)
  new Transformer({ fn: (ctx) => { ctx._logger = logger; } }),

  // Plain message — includes auto-pulled requestId / traceId from ctx
  new Log({ message: 'received', level: 'info' }),

  // Pick specific ctx keys to include
  new Log({ message: 'order processed', fields: ['orderId', 'total'] }),

  // Dump the entire ctx (cycles, Errors, BigInt, Map/Set handled)
  new Log({ message: 'debug ctx', fields: 'all', level: 'debug' }),
]);

Connectors

A connector is a thin, declarative wrapper around an injected client. The framework doesn't bundle Mongo / Redis / Kafka / Postgres — you bring the driver, you bring an authenticated client, you wire it in once at boot.

Mongo

Three components for mongodb's driver. They all honour an open transaction session if one's been set via BeginMongoTransaction.

const { MongoClient } = require('mongodb');
const mongo = new MongoClient('mongodb://127.0.0.1:27017');
await mongo.connect();
const db = mongo.db('app');

const { MongoConnector, MongoWriter, MongoDeleter } = require('flow-core');

new Chain('orders', [
  new Transformer({ fn: (ctx) => { ctx.q = { customerId: ctx.userId }; } }),
  new MongoConnector({
    client: db,
    collection: 'orders',
    queryVar:   'q',
    outputVar:  'orders',
  }),
  new Transformer({ fn: (ctx) => { ctx.newOrder = { customerId: ctx.userId, total: 99 }; } }),
  new MongoWriter({
    client: db,
    collection: 'orders',
    docVar:    'newOrder',
    outputVar: 'newOrderId',
  }),
  new Transformer({ fn: (ctx) => { ctx.delFilter = { _id: ctx.cancelOrderId }; } }),
  new MongoDeleter({
    client: db,
    collection: 'orders',
    filterVar: 'delFilter',
    outputVar: 'deletedCount',
  }),
]);

| Component | Reads | Writes | |-----------------|-----------------------------------------------|---------------------------------| | MongoConnector| ctx[queryVar] → collection.find(query).toArray() | ctx[outputVar] ← array of docs | | MongoWriter | ctx[docVar] → collection.insertOne(doc) | ctx[outputVar] ← insertedId | | MongoDeleter | ctx[filterVar] → collection.deleteOne(filter) | ctx[outputVar] ← deletedCount |

All three pass { session: ctx._mongoSession } automatically when set.

`RedisConnector`

Works with ioredis or any client that exposes lowercased command methods.

const Redis = require('ioredis');
const redis = new Redis();
const { RedisConnector } = require('flow-core');

new Chain('cache', [
  new Transformer({ fn: (ctx) => {
    ctx.cmd = 'set';
    ctx.args = [`user:${ctx.userId}`, JSON.stringify(ctx.user), 'EX', '60'];
  } }),
  new RedisConnector({
    client: redis,
    commandVar: 'cmd',
    argsVar:    'args',
    outputVar:  'cacheResult',   // optional
  }),
]);

When ctx._redisMulti is set (by BeginRedisTransaction), commands queue on the MULTI pipeline rather than running immediately. The MULTI is EXEC'd at chain settlement.

Variadic commands work transparently — args is spread into the call:

// MSET k1 v1 k2 v2 k3 v3
ctx.cmd = 'mset';
ctx.args = ['k1', 'v1', 'k2', 'v2', 'k3', 'v3'];

// MGET a b c → returns aligned [valA, valB, valC]
ctx.cmd = 'mget'; ctx.args = ['a', 'b', 'c'];

// HMGET hash f1 f2 f3
ctx.cmd = 'hmget'; ctx.args = ['user:1', 'name', 'email', 'role'];

// ZADD lb score1 m1 score2 m2 …
ctx.cmd = 'zadd'; ctx.args = ['leaderboard', 100, 'ada', 95, 'linus'];

// SADD k m1 m2 m3
ctx.cmd = 'sadd'; ctx.args = ['tags', 'red', 'green', 'blue'];

`RedisPipeline` (non-transactional batching)

Distinct from BeginRedisTransaction. Redis offers two batching modes:

| Mode | What it gives you | When to use | flow-core | |------|------------------|-------------|-----------| | MULTI | Atomic — all queued commands EXEC together, none run if aborted | All-or-nothing semantics across commands | BeginRedisTransaction + RedisConnector + CommitAll/AbortAll | | Pipeline | Network batching — N commands in one round-trip, each runs independently on the server | Bulk unrelated reads/writes for latency reduction | RedisPipeline |

const { RedisPipeline } = require('flow-core');

new Chain('bulk', [
  new Transformer({ fn: (ctx) => {
    ctx.redisOps = [
      ['set',  'user:1:lastSeen', Date.now()],
      ['get',  'user:1:profile'],
      ['mget', 'feature:a', 'feature:b', 'feature:c'],
      ['hset', 'session:abc', 'expires', '300'],
      ['del',  'tmp:cleanup'],
    ];
  } }),
  new RedisPipeline({
    client: redis,
    commandsVar: 'redisOps',
    outputVar:   'pipelineResults',  // ioredis shape: [[err, result], ...]
    failOnAnyError: false,           // default — set true to throw if any sub-command errored
  }),
]);

pipelineResults is ioredis's standard shape — one [err, result] tuple per submitted command, in input order. With failOnAnyError: true the component throws an aggregate error if any tuple has a non-null err.

`KafkaPublisher`

Works with kafkajs's Producer.

const { Kafka } = require('kafkajs');
const producer = new Kafka({ clientId: 'svc', brokers: ['127.0.0.1:9092'] }).producer();
await producer.connect();

const { KafkaPublisher } = require('flow-core');

new Chain('publish', [
  new Transformer({ fn: (ctx) => {
    ctx.topic = 'orders.placed';
    ctx.body = { orderId: ctx.id, at: new Date().toISOString() };
    ctx.key  = ctx.id;
  } }),
  new KafkaPublisher({
    client:     producer,
    topicVar:   'topic',
    payloadVar: 'body',
    keyVar:     'key',         // optional
  }),
]);

If ctx._kafkaTx is set (BeginKafkaTransaction), messages route through the transactional producer instead.

SQL

Both connectors use named SQL via ctx vars. The library is driver-agnostic — pass any client whose interface matches.

const mysql = require('mysql2/promise');
const pool = await mysql.createPool({ host: 'localhost', user: 'root' });

const { MysqlConnector } = require('flow-core');

new Chain('q', [
  new Transformer({ fn: (ctx) => {
    ctx.sql = 'SELECT * FROM users WHERE id = ?';
    ctx.params = [ctx.userId];
  } }),
  new MysqlConnector({ client: pool, queryVar: 'sql', paramsVar: 'params', outputVar: 'rows' }),
]);

const { Pool } = require('pg');
const pg = new Pool({ /* … */ });

const { PostgresConnector } = require('flow-core');

new PostgresConnector({
  client: pg,
  queryVar: 'sql',          // 'SELECT * FROM users WHERE id = $1'
  paramsVar: 'params',
  outputVar: 'rows',
});

MysqlConnector calls client.execute(sql, params) and falls back to client.query if execute isn't present. PostgresConnector calls client.query(sql, params) and stores result.rows.

`ElasticsearchConnector`

const { Client } = require('@elastic/elasticsearch');
const es = new Client({ node: 'http://localhost:9200' });

const { ElasticsearchConnector } = require('flow-core');

new ElasticsearchConnector({
  client: es,
  index:  'flights',
  queryVar:  'query',         // ctx.query = { match: { origin: 'JFK' } }
  outputVar: 'hits',
  action: 'search',           // or 'index' / 'get' / etc.
});

For search the output is hits.hits flattened to [{ _id, ...source }, ...]. For other actions, the raw response body lands on outputVar.

`HttpConnector`

Uses global fetch (Node 18+).

const { HttpConnector } = require('flow-core');

// Static config
new HttpConnector({
  method: 'POST',
  url: 'https://api.example.com/charges',
  headers: { Authorization: 'Bearer xxx' },
  bodyVar:  'chargeBody',     // any JSON-serializable object — auto-stringified
  outputVar: 'chargeResp',    // { status, headers, body }
  expectedStatus: 200,        // throws if status !== 200
  timeoutMs: 5000,
});

// Or dynamic — pull url/headers from ctx
new HttpConnector({
  method:    'GET',
  urlVar:    'apiUrl',
  headersVar:'apiHeaders',
  outputVar: 'profile',
});

| Option | Type | Default | |-------------------|----------------------------------------|---------| | method | 'GET' / 'POST' / 'PUT' / etc. | GET | | url or urlVar | static string OR ctx key | one or the other required | | headers or headersVar | object OR ctx key | empty | | body or bodyVar | object/string OR ctx key | none for GET/HEAD | | outputVar | ctx key for { status, headers, body }| required | | expectedStatus | number; throws if status doesn't match | none | | timeoutMs | aborts via AbortController | 30000 |

Response body is auto-parsed as JSON when the response's Content-Type contains json, otherwise stored as text.

`CustomConnector`

Generic escape hatch for I/O we don't have a dedicated component for. Use this instead of a Transformer when the work is an integration point — the distinction signals readers "this talks to an external system, no business logic here."

const stripe = require('stripe')(process.env.STRIPE_KEY);

new CustomConnector({
  name: 'stripe.charge',
  client: stripe,
  action: async (client, ctx) => client.charges.create({
    amount: ctx.amount,
    currency: 'usd',
    customer: ctx.customerId,
  }),
  outputVar: 'charge',
});

| Option | Description | |--------------|------------------------------------------------------| | client | The integration's SDK / client (required) | | action | (client, ctx) => any — your I/O call (required) | | outputVar | ctx key for the action's return value (optional) |

Resilience

Three composable wrappers — each takes another executable (Component, Chain, Series, Parallel) as step. They compose freely:

Retry → CircuitBreaker → HttpConnector       // try several times, breaker counts the failures
CircuitBreaker → Retry → HttpConnector       // breaker counts each call (incl. retries) — different semantics
Retry → CacheRead( loader=Retry→HttpConnector ) // retry on cache miss too

Rule of thumb on onError: Retry and CircuitBreaker only see uncaught throws. If the wrapped step has its own onError, the error never reaches the wrapper. Strip the inner onError and use the wrapper's own onError for terminal-failure routing.

`Retry`

const { Retry } = require('flow-core');

new Retry({
  step: new HttpConnector({ method: 'GET', urlVar: 'partnerUrl', outputVar: 'partnerResp' }),
  attempts: 4,
  backoff: 'exponential',     // 'fixed' | 'linear' | 'exponential' | (attempt)=>ms
  delayMs: 100,               // base delay
  maxDelayMs: 5000,           // cap (post-jitter cap not applied — keep the cap reasonable)
  jitter: 0.25,               // 0..1 — ± multiplier
  retryOn: (err) => err.code !== 'INVALID_INPUT',
  onRetry: ({ attempt, error, nextDelayMs }) =>
    console.warn(`retry ${attempt} after ${nextDelayMs}ms: ${error.message}`),
  onError: failureChain,      // terminal-failure jump after the final attempt
});

| Option | Type | Default | |--------------|---------------------------------------------------|----------------| | step | any executable | required | | attempts | positive integer (total including first) | required | | backoff | 'fixed' \| 'linear' \| 'exponential' \| fn | 'exponential'| | delayMs | base delay | 100 | | maxDelayMs | cap on each individual delay | 30000 | | jitter | 0..1 randomisation factor | 0.2 | | retryOn | (err) => bool — false skips retry, throws now | () => true | | onRetry | ({ attempt, error, nextDelayMs, name }) => any | none |

Backoff formulas (jitter applied after):

fixed:       delayMs
linear:      delayMs * attempt
exponential: delayMs * 2^(attempt-1)
function:    backoff(attempt)

ctx._retryAttempt is set to the current attempt number while the wrapped step runs. Useful for the inner step to vary its behaviour by attempt (e.g., test-only deterministic failures).

`CircuitBreaker`

const { CircuitBreaker } = require('flow-core');

// One breaker per upstream — construct ONCE at boot and reuse everywhere.
const partnerBreaker = new CircuitBreaker({
  name: 'breaker.partner-airline',
  step: partnerHttp,
  failureThreshold: 5,        // failures within windowMs → open
  successThreshold: 2,        // successes in half-open → close
  windowMs: 60_000,           // failure-counting window
  openMs: 30_000,             // dwell time in open state
  isFailure: (err) => err.code !== 'CLIENT_ABORT',
  onError: fallbackChain,     // when breaker is open OR step throws
});

// Reuse in every chain that calls this upstream
new Chain('search', [...prelude, partnerBreaker, ...etc]);
new Chain('book',   [...prelude, partnerBreaker, ...etc]);

State machine:

                              failureThreshold within windowMs
                  closed ─────────────────────────────────────►  open
                    ▲                                              │
   successThreshold │                                              │ after openMs
                    │                                              ▼
                    └──── successful probe ──── half-open ─── probe call
                                                      │
                                                      │ fail
                                                      └──── re-open

Operations methods (call directly, not through a chain):

| Method | Purpose | |--------|---------| | breaker.status() | { state, recentFailures, openedAt, halfOpenSuccesses } | | breaker.reset() | Force back to closed (for ops / tests) | | breaker.now = customClock | Hookable for deterministic testing |

The breaker throws err.code === 'CIRCUIT_OPEN' with err.circuitOpen = true when it short-circuits. Don't retry on this error — set retryOn: (err) => !err.circuitOpen on any Retry that wraps a breaker:

new Retry({
  step: partnerBreaker,
  retryOn: (err) => !err.circuitOpen,   // pointless to retry while breaker is open
  attempts: 3,
});

`RateLimit`

Redis-backed. Three strategies trade off implementation complexity for behaviour quality.

const { RateLimit } = require('flow-core');

new RateLimit({
  client: redis,
  keyVar: 'rateKey',          // e.g. ctx.rateKey = `rl:user:${userId}`
  limit: 100,                 // requests
  windowSec: 60,              // window
  strategy: 'sliding-window', // 'fixed-window' (default) | 'sliding-window' | 'token-bucket'
  onExceeded: tooManyChain,   // optional jump
});

| Strategy | Implementation | When to use | |-------------------|-----------------------------------------------------|------------------------------------------------------------------------| | fixed-window | INCR + EXPIRE — single round-trip when cached | Cheap, predictable; allows up to 2× limit briefly at window boundaries | | sliding-window | ZSET of timestamps, ZREMRANGEBYSCORE + ZCARD | Smoother, no boundary burst; slightly heavier | | token-bucket | Lua EVAL atomic refill+take | Allows controlled bursts; ideal for "5/sec but up to 20 burst" |

Output:

ctx.rateState = {
  allowed: true | false,
  count: number,              // requests in window (fixed/sliding) OR tokens left (token-bucket)
  limit: number,
  retryAfterSec: number,      // suggested Retry-After on rejection
}

If onExceeded is omitted, the component leaves ctx.rateState.allowed = false and the chain continues — your next step decides what to do. A Conditional is the natural follow-up:

new Chain('protected', [
  new Transformer({ fn: (ctx) => { ctx.rateKey = `rl:${ctx.principal.userId}`; } }),
  new RateLimit({ client: redis, keyVar: 'rateKey', limit: 100, windowSec: 60 }),
  new Conditional({ branches: [[(ctx) => !ctx.rateState.allowed, tooManyChain]] }),
  // ... actual handler
]);

Caching

Two components for the five common cache patterns. Both accept Redis-like clients (anything with get/set/del).

`CacheRead` (cache-aside / read-through)

const { CacheRead } = require('flow-core');

new CacheRead({
  client: redis,
  keyVar: 'cacheKey',           // ctx.cacheKey = `flights:${origin}-${dest}`
  outputVar: 'flights',         // ctx.flights ← cached or loaded value
  loader: loadFromMongoChain,   // executable run on miss
  loaderValueVar: 'fromDb',     // optional; if loader writes elsewhere, name it
  ttlSec: 60,                   // TTL for the write-back
  serializer: JSON.stringify,
  deserializer: JSON.parse,
  skipCacheVar: 'noCache',      // optional — bypass cache when truthy
});

Behaviour:

| Phase | What happens | |----------|--------------------------------------------------------------------| | Lookup | client.get(ctx[keyVar]). Sets ctx.cacheHit = true/false. | | Hit | Deserialise, write to ctx[outputVar]. Loader never runs. | | Miss | Run loader.execute(ctx). Take the value from ctx[loaderValueVar] (or outputVar if not set). Write back with ttlSec. | | No loader| Miss leaves ctx[outputVar] = null. |

"Cache-aside" and "read-through" are commonly interchangeable in the literature when the framework owns the orchestration — flow-core does, so they're one component here.

`CacheWrite` (through / behind / around)

const { CacheWrite } = require('flow-core');

new CacheWrite({
  client: redis,
  strategy: 'through',                  // 'through' (default) | 'behind' | 'around'
  keyVar: 'cacheKey',
  valueVar: 'newValue',
  ttlSec: 60,
  persister: writeToMongoChain,         // required for all three strategies
  invalidateOnAround: false,            // strategy='around' only: DEL the cache key after persist
  serializer: JSON.stringify,
});

| Strategy | Cache write | Source write | When to use | |------------|------------------|-----------------------------------------|----------------------------------------------| | through | sync | sync (both must succeed) | strong consistency between cache and source | | behind | sync | async — setImmediate, errors logged | low write latency; eventual durability | | around | none (or DEL) | sync | rare writes, frequent reads; let cache repopulate on next read |

behind mode warning: the persister runs after the chain has already returned to the caller. Failures are caught and logged to ctx._logger (or stderr) — they do not fail the chain. A real production deployment needs an outbox table + worker for guaranteed delivery; this in-process queue loses pending writes on crash. The component is best for low-stakes writes (view counters, analytics events).

Authentication — `JwtVerify`

flow-core has no opinion on which JWT library you use. Inject the verify function and the component handles the policy checks.

const jwt = require('jsonwebtoken');
const { JwtVerify } = require('flow-core');

new JwtVerify({
  tokenVar: 'authToken',
  verify: (token) => jwt.verify(token, process.env.JWT_SECRET, {
    algorithms: ['HS256'],
  }),
  claimsVar: 'principal',
  issuer: 'auth.example.com',         // optional — must match claims.iss
  audience: 'flights-api',            // optional — checked against claims.aud (string or array)
  onError: unauthorizedChain,
});

A common pattern is to strip the Bearer prefix in a small transformer before the verifier:

new Chain('protected', [
  new Transformer({ fn: (ctx) => {
    const auth = ctx._headers && ctx._headers.authorization;
    ctx.authToken = auth && auth.startsWith('Bearer ') ? auth.slice(7) : null;
  } }),
  new JwtVerify({ tokenVar: 'authToken', claimsVar: 'principal', verify, onError: unauthorized }),
  // ... handler
]);

The verify function may return claims synchronously or as a Promise (for jose / async JWKS lookups). It should throw on invalid signature / malformed / expired. The component additionally enforces:

| Check | If claims.iss !== issuer | throws iss mismatch | |-------------|------------------------------------------------------------|-----------------------| | audience | claims.aud === audience or claims.aud.includes(audience) | else throws aud mismatch |

Observability — tracing, pagination, configuration

Three small but high-leverage utilities for production-shape services.

`TraceContext` and `Span`

W3C Trace Context plus an in-process span recorder, with an optional bridge to any OpenTelemetry-compatible tracer.

const { TraceContext, Span, HttpConnector, Chain } = require('flow-core');

// 1. Parse the inbound traceparent (or start a new trace).
const handler = new Chain('handler', [
  new Transformer({ fn: (ctx) => {
    ctx._traceContext =
      TraceContext.parse(ctx._headers.traceparent) || TraceContext.newRoot();
  } }),

  // 2. Wrap any expensive step in a Span. Nested spans share the trace.
  new Span({
    spanName: 'db.fetch.users',
    step: new MongoConnector({ /* ... */ }),
    attributes: { table: 'users' },
    attributesFromCtx: { tenantId: 'tenantId' },
  }),

  // 3. Outgoing HTTP automatically carries `traceparent`.
  new Span({
    spanName: 'http.partner.search',
    step: new HttpConnector({ method: 'GET', urlVar: 'partnerUrl', outputVar: 'partnerResp' }),
  }),
]);

After the chain runs, ctx._spans holds every span the chain recorded:

[
  {
    name: 'db.fetch.users',
    traceId: 'cb…64-hex…',
    spanId:  '7e…16-hex…',
    parentSpanId: '20…16-hex…',
    durationMs: 12,
    status: 'ok',
    attributes: { table: 'users', tenantId: 'acme' },
  },
  // … more spans, in finish order
]

| Method | Purpose | |---------------------------------|----------------------------------------------------------| | TraceContext.parse(header) | Returns a context, or null if header invalid | | TraceContext.newRoot(flags?) | New root context (fresh traceId/spanId) | | tc.child() | Child span — same traceId, new spanId, parent = this | | tc.toHeader() | Render as W3C traceparent for outgoing requests |

HttpConnector automatically injects traceparent from ctx._traceContext on every outgoing call — user-supplied headers win, so you can override.

Bridging to OpenTelemetry (or any tracer with startSpan(name, opts) → { setStatus, recordException, end }):

const { trace } = require('@opentelemetry/api');
const otelTracer = trace.getTracer('flights');

new Span({ spanName: 'search', step: …, tracer: otelTracer });

The component still records the span on ctx._spans and forwards to OTel — so local debugging tooling and Jaeger/Datadog see the same span.

`Pagination`

Offset-based input normaliser. Reads page and size from ctx, clamps to safe bounds, exposes skip/limit ready for connectors.

new Pagination({
  pageVar: 'page',          // ctx.page (typically from req.query.page)
  sizeVar: 'pageSize',
  outputVar: 'paging',      // ctx.paging = { page, size, skip, limit }
  defaultPage: 1,
  defaultSize: 20,
  maxSize: 100,
});

// then …
new Transformer({ fn: (ctx) => { ctx.q = { active: true }; } });
// MongoConnector that respects ctx.paging.skip + .limit (write a thin custom
// connector, or use CustomConnector for now — the built-in MongoConnector is
// query-only).

Build response meta with a follow-up transformer:

new Transformer({ fn: (ctx) => {
  const p = ctx.paging;
  ctx.pageMeta = {
    page: p.page,
    size: p.size,
    totalCount: ctx.totalCount,
    totalPages: Math.ceil(ctx.totalCount / p.size),
    hasMore: p.skip + (ctx.rows || []).length < ctx.totalCount,
  };
} });

Cursor-based pagination is too API-specific to ship as a component — encode/decode is a one-liner pair of transformers per route.

`ConsulGet` — Hashicorp Consul KV

Fetch a single key (or recurse a prefix) from Consul's KV store. Two modes:

Mode 1 — with the consul npm client:

const consul = require('consul')({ host: '127.0.0.1', port: '8500' });
new ConsulGet({ client: consul, keyVar: 'cfgKey', outputVar: 'cfg' });

Mode 2 — endpoint URL only (zero-dep, uses global fetch):

new ConsulGet({
  endpoint: 'http://consul.svc.cluster.local:8500',
  token: process.env.CONSUL_HTTP_TOKEN,
  datacenter: 'dc1',
  keyVar: 'cfgKey',          // ctx.cfgKey = 'config/airline-x/url'
  outputVar: 'partnerUrl',
  parse: 'json',             // 'raw' (default) | 'json'
  recurse: false,            // true → outputs array under a prefix
  required: true,            // false → outputVar = null on miss, no throw
  dataIndex: 'consulIndex',  // ctx[dataIndex] receives X-Consul-Index (for long-polling)
});

Add a TTL cache in front of Consul — single source of truth lookups shouldn't hit Consul on every request:

new CacheRead({
  client: redis,
  keyVar: 'consulCacheKey',
  outputVar: 'partnerUrl',
  ttlSec: 300,
  loader: new Chain('consul-load', [
    new Transformer({ fn: (ctx) => { ctx.cfgKey = `config/${ctx.tenant}/url`; } }),
    new ConsulGet({ endpoint, keyVar: 'cfgKey', outputVar: 'partnerUrlValue' }),
  ]),
  loaderValueVar: 'partnerUrlValue',
});

Scheduled chains — BullMQ, node-cron, k8s CronJob

ScheduledChainRunner bridges external schedulers to chains. Construct once at boot with the engine and a { name: Chain } map; the runner exposes .run(name, ctx) plus thin adapters for the two most common schedulers.

const { Engine, ScheduledChainRunner } = require('flow-core');

const engine = new Engine({ metricsAdapter });
const runner = new ScheduledChainRunner({
  engine,
  chains: { refreshRates, expireBookings, dailyReport, retryStuckOrders },
  logger,
});

node-cron

const cron = require('node-cron');

cron.schedule('*/5 * * * *', runner.toCron('refreshRates'));
cron.schedule('0 0 * * *',   runner.toCron('dailyReport', { reportDate: 'today' }));

toCron(name, ctx?) returns a zero-arg callback that runs the chain. Failures inside the chain are caught and logged — node-cron's callback is synchronous, so unhandled rejections would crash the process otherwise.

BullMQ

const { Worker, Queue } = require('bullmq');

const worker = new Worker('flights-jobs', runner.toBullMQ(), { connection: redis });

// Producer-side:
const queue = new Queue('flights-jobs', { connection: redis });
await queue.add('refreshRates', { source: 'cron' });
await queue.add('arbitrary-job-name', { chain: 'expireBookings', cutoff: '2026-01-01' });

The processor resolves the chain in this order:

job.data.chain (explicit)
job.name (fallback)

Job data becomes the initial ctx, with _job = { id, name, attempt } attached for traceability.

Kubernetes CronJob

flow-core's role here is just "be runnable from a process entrypoint":

// scripts/run-scheduled.js
const { runner } = require('./bootstrap');

const chainName = process.env.CHAIN;
const initialCtx = JSON.parse(process.env.CHAIN_INPUT || '{}');

runner.run(chainName, initialCtx)
  .then((ctx) => process.exit(ctx.status === 'ok' ? 0 : 1))
  .catch((err) => { console.error(err); process.exit(2); });

Then the CronJob manifest:

apiVersion: batch/v1
kind: CronJob
metadata:
  name: refresh-rates
spec:
  schedule: "*/5 * * * *"
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: scheduler
              image: my-flights-app:latest
              command: ["node", "scripts/run-scheduled.js"]
              env:
                - name: CHAIN
                  value: refreshRates
                - name: CHAIN_INPUT
                  value: '{"source":"k8s-cron"}'

Exit codes:

| Code | Meaning | |------|------------------------------------------------------| | 0 | Chain ended with ctx.status === 'ok' | | 1 | Chain finished but status was something else | | 2 | Chain threw — see container logs |

Same chain definitions back all three schedulers — only the trigger differs.

Transactions

The library models best-effort distributed transactions across heterogeneous resources (Mongo, Redis, Kafka — and anything else you write a Begin* for). Inspired by Spring's ChainedTransactionManager.

Begin / Commit / Abort lifecycle

const {
  Engine, Chain, Transformer,
  BeginMongoTransaction, BeginKafkaTransaction, BeginRedisTransaction,
  MongoWriter, RedisConnector, KafkaPublisher,
  CommitAll, AbortAll,
} = require('flow-core');

const audit = new Chain('audit', [
  // Open one tx per resource. Each pushes a "participant" onto ctx._participants.
  new BeginMongoTransaction({ client: mongo }),
  new BeginRedisTransaction({ client: redis }),
  new BeginKafkaTransaction({ producer: txProducer }),

  new Transformer({ fn: (ctx) => { ctx.doc = { event: 'login', at: new Date().toISOString() }; } }),
  new MongoWriter({ client: db, collection: 'audit', docVar: 'doc' }),

  new Transformer({ fn: (ctx) => { ctx.cmd = 'set'; ctx.args = ['last_login', ctx.doc.at]; } }),
  new RedisConnector({ client: redis, commandVar: 'cmd', argsVar: 'args' }),

  new Transformer({ fn: (ctx) => { ctx.topic = 'auth.login'; ctx.body = ctx.doc; } }),
  new KafkaPublisher({ client: producer, topicVar: 'topic', payloadVar: 'body' }),

  new CommitAll(),
]);

| Component | Pushes participant with | |--------------------------|------------------------------------------------------| | BeginMongoTransaction | client.startSession() + session.startTransaction(). Session lives on ctx._mongoSession. | | BeginRedisTransaction | client.multi() queue. Lives on ctx._redisMulti. | | BeginKafkaTransaction | producer.transaction(). Lives on ctx._kafkaTx. | | CommitAll | Walks ctx._participants in order, calls commit() on each. Clears the list. | | AbortAll | Walks ctx._participants in reverse order, calls abort() on each. Swallows abort errors. |

Tagged transactions

Open multiple transactions and settle them individually with CommitTransaction({ tag }) / AbortTransaction({ tag }).

const {
  BeginMongoTransaction, BeginKafkaTransaction,
  CommitTransaction, AbortTransaction,
} = require('flow-core');

new Chain('selective', [
  new BeginMongoTransaction({ client: mongo, tag: 'orders' }),
  new BeginKafkaTransaction({ producer: txProducer, tag: 'events' }),

  // … do work …

  new CommitTransaction({ tag: 'orders' }),   // commits one
  new AbortTransaction({ tag: 'events' }),    // aborts the other
]);

A participant's tag defaults to the component's name when omitted. After CommitTransaction / AbortTransaction runs, that participant is removed from ctx._participants — the engine's chain-exit auto-settle ignores it.

Per-chain scoping

The chain is the unit of work. When a chain begins, the engine snapshots ctx._participants and resets to []. When the chain ends, the chain's own participants are settled (commit or abort), then the outer participants are restored.

const outer = new Chain('outer', [
  new BeginMongoTransaction({ client: mongo }),       // outer participant
  inner,                                              // chain-as-component
  new CommitAll(),                                    // commits outer only
]);

const inner = new Chain('inner', [
  new BeginMongoTransaction({ client: secondMongo }), // inner participant — invisible to outer
  new Transformer({ fn: () => { /* work */ } }),
  // no CommitAll — engine auto-commits inner participants at inner's exit
]);

What this prevents:

CommitAll inside inner accidentally committing outer's open transactions.
AbortTransaction({ tag: 'orders' }) inside a compensator chain hitting an outer-chain participant that happens to share the tag.

Auto-settlement

If a chain ends with ctx._participants non-empty:

| How the chain ended | Engine action | |--------------------------------------|-----------------------------------------------------| | Ran every component, no throw | Commit all dangling participants (in order) | | Conditional jump (no error) | Commit (jump is intentional control flow) | | EXIT / CustomExecutor.exit() | Commit | | onError jump (error-jump) | Abort all dangling participants | | Uncaught error escaping the chain | Run saga compensators (LIFO), then abort participants, then rethrow |

The engine emits engine.auto_commit_on_exit, engine.auto_abort_on_exit, engine.auto_compensate, and engine.auto_abort log events so you can spot this happening.

Partial commit failures

CommitAll (and the engine's auto-commit) is best-effort 2PC: commits run sequentially. If commit #2 of 3 fails, commits #1 already succeeded, and the remaining are aborted.

try {
  await engine.run(chain, {});
} catch (err) {
  // err.message is like:
  // 'auto-commit failed at "BeginKafkaTransaction": ...
  //  Commit

Published

Vulnerabilities

Links

Maintainers

Keywords

Readme

flow-core

Table of contents

Mental model

Hello world

What's new in 0.2.0

Critical correctness

New components

Connector ergonomics

Discoverability

Upgrade notes

Core types

Context (ctx)

Component

Chain

Engine

Control flow

Component results

Conditional jump vs invoke

EXIT sentinel

Series and Parallel

Chain as component

CustomExecutor

onError chains