@tomhundley/condux

A composable Node.js toolkit for async workflows: pipelines, DAG jobs, retries, circuit breakers, rate limits, schema validation, caching, events, and an embeddable runtime.

Zero runtime dependencies. Node 18+.

Install

npm install @tomhundley/condux

Pipeline

import { pipeline } from "@tomhundley/condux";

const names = await pipeline({ name: "users" })
  .use(async (input) => input.split(","))
  .map((name) => name.trim())
  .filter((name) => name.length > 0)
  .retry({ times: 3, delay: 50 })
  .timeout(2_000)
  .run("Ada,  Grace, Alan");

Workflow graph

Independent steps run in parallel. Dependent steps wait automatically.

import { workflow } from "@tomhundley/condux";

const provision = workflow("provision")
  .step("config", async ({ input }) => ({ region: input.region, size: "s" }))
  .step("db", async ({ get }) => ({ id: `db-${get("config").region}` }), {
    dependsOn: ["config"],
    retry: { times: 3, delay: 25 },
  })
  .step("cache", async ({ get }) => ({ id: `cache-${get("config").region}` }), {
    dependsOn: ["config"],
  })
  .step("ready", async ({ get }) => ({ db: get("db"), cache: get("cache") }), {
    dependsOn: ["db", "cache"],
  });

const { results } = await provision.run({ region: "us-east-1" });

Resilience

import { retry, timeout, CircuitBreaker, RateLimiter, MemoryCache } from "@tomhundley/condux";

const breaker = new CircuitBreaker({ failureThreshold: 5, resetMs: 15_000 });
const limiter = new RateLimiter({ capacity: 20, refillPerSecond: 5 });
const cache = new MemoryCache({ max: 200, ttl: 30_000 });

const payload = await cache.wrap("users", () =>
  limiter.wrap(() =>
    breaker.exec(() =>
      retry(() => timeout(() => fetch("https://example.com/users"), 1_500), {
        times: 4,
        delay: 100,
        backoff: 2,
      }),
    ),
  )(),
);

Schema validation

import { s } from "@tomhundley/condux";

const User = s.object({
  id: s.string().min(1),
  email: s.string().email(),
  age: s.number().int().min(0).max(150).optional(),
  role: s.enum("admin", "user"),
  tags: s.array(s.string()).max(8),
});

const user = User.parse({
  id: "u_1",
  email: "[email protected]",
  role: "admin",
  tags: ["math"],
});

Runtime + plugins

import { createRuntime, definePlugin, s } from "@tomhundley/condux";

const metrics = definePlugin("metrics", (app) => {
  app.hook("afterWorkflow", ({ name, result }) => {
    app.logger.info("workflow finished", { name, ms: result.durationMs });
  });
});

const app = createRuntime({ name: "orders", level: "info" })
  .use(metrics)
  .registerWorkflow("checkout", (wf) =>
    wf
      .step("validate", ({ input }) =>
        s.object({ sku: s.string(), qty: s.number().int().positive() }).parse(input),
      )
      .step("charge", ({ get }) => ({ ok: true, sku: get("validate").sku }), {
        dependsOn: ["validate"],
      }),
  );

await app.runWorkflow("checkout", { sku: "book", qty: 2 });

HTTP client

import { createClient } from "@tomhundley/condux";

const api = createClient({
  baseUrl: "https://api.example.com",
  timeout: 3_000,
  retry: { times: 3, delay: 100 },
  breaker: { failureThreshold: 8, resetMs: 10_000 },
});

const { body } = await api.get("/health");

What is included

| Module | Purpose | | --- | --- | | pipeline | Fluent async transforms with retry, timeout, fallback | | workflow | DAG of named steps with dependency parallelism | | retry / timeout | Call-level resilience | | CircuitBreaker | Fail-fast after repeated errors | | RateLimiter | Token-bucket throttling | | ConcurrencyPool / mapPool | Bounded parallelism | | MemoryCache / memoize | LRU + TTL | | s | Runtime schema parser | | EventBus | Wildcard async events | | compose | Koa-style middleware | | loadConfig | Deep merge + prefixed env vars | | createLogger | Structured JSON logs | | Scheduler | Interval and one-shot jobs | | createClient | fetch wrapper | | createRuntime | Plugins, hooks, registered flows |

License

MIT