kahn-queue (TypeScript)

Getting started

• TypeScript: ~6 (devDependencies in package.json).
• Tests: cd typescript and npm test (Vitest).
• Build: npm run build (tsup → dist/).
• Import: ./dist/index.js (after build), or your bundler’s alias to that entry.

Pieces

| Piece | Role | |--------|------| | Dag / Dag.Builder | Immutable DAG: Dag.builder(), add, connect, build(). | | KahnScheduler | Drives execution: run, signalComplete / signalFailed; getResult() returns DagResult (sets of ids). | | KahnQueue | Single queue implementation backing the scheduler (see source for pop / prune / peek). | | IllegalGraphException | Thrown for invalid graphs (e.g. self-loop or cycle at build()). | | NodeProgressTracker | Optional per-node progress in [0, 1]; not required for scheduling. |

Example

Single-threaded use: call run() once, then drive completion from your callback (e.g. Temporal-style workflow logic).

import { Dag, KahnScheduler } from "./dist/index.js";

const b = Dag.builder<string>();
const lint = b.add("lint");
const compile = b.add("compile");
const test = b.add("test");
b.connect(lint, compile).connect(compile, test);
const dag = b.build();

const sched = KahnScheduler.create(dag, (id, s) => {
  try {
    runStep(dag.get(id));
    s.signalComplete(id);
  } catch {
    s.signalFailed(id);
  }
});
sched.run();
const result = sched.getResult();

Manual KahnQueue

import { Dag, KahnQueue } from "./dist/index.js";

const b = Dag.builder<string>();
const lint = b.add("lint");
const compile = b.add("compile");
const test = b.add("test");
b.connect(lint, compile).connect(compile, test);
const dag = b.build();

const q = new KahnQueue(dag);

let ready = [...q.peek()];

while (ready.length > 0) {
  const id = ready.shift()!;

  // do work for `id` (e.g. runStep(dag.get(id)))

  const promoted = q.pop(id);

  ready.push(...promoted);

  // If a node fails, you can prune it (and its descendants):
  // q.prune(id);
}