DEV Community: AdmilsonCossa

Don't let a billion RAG docs drown your 25-result pipeline

AdmilsonCossa — Mon, 25 May 2026 10:17:41 +0000

Backpressure For Streaming Pipelines

Last time we showed how to terminate non-cooperative CPU work at the worker boundary. This article stays cooperative but adds the missing piece: backpressure, the runtime contract that lets a producer pause the moment the consumer can't keep up.

A RAG ingest pipeline has a billion candidate documents. You only need the 25 that match a downstream filter. A naive promise collection can materialize far more work than the consumer needs; a hand-rolled async iterator can still fill a prefetch buffer before the first result arrives. With WorkIt:

import { work } from "@workit/core";

async function* billionDocuments() {
  for (let i = 0; i < 1_000_000_000; i++) yield { id: i, text: `doc ${i}` };
}

const results = [];
for await (const processed of work(billionDocuments())
  .inParallel(16)
  .map(async (doc, ctx) => enrich(doc, { signal: ctx.signal }))
  .stream()) {
  results.push(processed);
  if (results.length === 25) break;
}

Two things to notice:

work() accepts an async iterable directly. No .from(), no Readable.from(...) shim. The signature is Iterable<I> | AsyncIterable<I> -> WorkBuilder<I, I>.
.map().stream() is the streaming pipeline form. .do(fn) returns a Promise<WorkOutput<R>> (full batch result). .map(fn) returns a new builder; .stream() on a builder returns an AsyncIterable<O> that respects backpressure. Both terminals exist; you pick by what the consumer is doing.

What the producer actually does:

Bench 09-stream-1b-lazy.mjs. 1,000,000,000-row generator. inParallel(16). Consumer takes 25, breaks.

Implementation Consumed Items pulled from the generator maxActive In-flight after break

Naïve eager prefetch buffer (256-deep) 25 281 1 0 (all let to settle)

work().inParallel(16).map().stream() 25 40 1 0 (cancelled at break)

Implementation	Consumed	Items pulled from the generator	maxActive	In-flight after break
Naïve eager prefetch buffer (256-deep)	25	281	1	0 (all let to settle)
`work().inParallel(16).map().stream()`	25	40	1	0 (cancelled at break)

These are representative captured values. The bench asserts the invariant: produced items stay bounded by TAKE + CONCURRENCY. The naïve baseline pulled 281 items because once the prefetch buffer is full it doesn't pause the producer -- it pauses the worker pool, which is a different question.

That's backpressure: the producer pauses when the consumer slows down or stops, not when the worker pool fills.

`work().stream()` -- bounded, lazy, cancellable

for await (const summary of work(documents)
  .inParallel(8)
  .withRetry(2)
  .withTimeout("15s")
  .map(async (doc, ctx) => summarize(doc, { signal: ctx.signal }))
  .stream()) {
  ui.append(summary);
}

Properties the runtime guarantees:

inParallel(N) is a hard cap. maxActive never exceeds N. Property test runs 1..20 wide x 1..100 items, asserts the cap holds across every shape.
stream() is lazy. The producer iterator pulls only when an inflight slot is free.
break is cancellation. The remaining inflight tasks abort with CancelReason { kind: "manual", tag: "stream_consumer_closed" }. Their ctx.defer runs. The producer iterator's return() runs.
A throw inside the body triggers CancelReason { kind: "manual", tag: "stream_failed" } for siblings -- typed, distinguishable from the consumer-break path on a dashboard.
Slow consumer pauses producer. Tracked under check:stream-memory: 1,000,000 logical items, slow consumer, bounded heap growth, and no unbounded producer advance.

Bench 10-stream-slow-consumer.mjs. 5,000-item source, inParallel(16), consumer ~5 ms per item, take 200.

Metric Value

Consumed 200

Produced 215

Producer overshoot 15 (bound: CONCURRENCY + 1 = 17)

maxActive 1

In-flight after break 0

Wall time ~3,108 ms

Metric	Value
Consumed	200
Produced	215
Producer overshoot	15 (bound: `CONCURRENCY + 1` = 17)
maxActive	1
In-flight after break	0
Wall time	~3,108 ms

The interesting detail: even with inParallel(16), maxActive stayed at 1 because the consumer was the bottleneck. The runtime didn't speculatively saturate the worker pool -- it paced the producer to consumer demand. That is what "backpressure" actually means. A pool that always runs at capacity isn't backpressure; it's a pool.

Streaming map: stop after 12, produce only what demand requires

The most practical reader-facing form of the same property -- a real summarizer pipeline, the size of a real prompt:

// samples/streaming-summarizer.sample.js
const TAKE = 12;
const CONCURRENCY = 5;

for await (const summary of work(documents())
  .inParallel(CONCURRENCY)
  .withRetry(2)
  .withTimeout("500ms")
  .map(async (doc, ctx) => `summary:${doc.id}`)
  .stream()) {
  summaries.push(summary);
  if (summaries.length === TAKE) break;
}

// Asserted by the sample:
//   summaries.length === TAKE
//   produced     <= TAKE + CONCURRENCY - 1
//   maxActive    === CONCURRENCY
//   active       === 0       // all in-flight cancelled cleanly on break

50-doc generator. Consume 12. Producer never advances past 16. Concurrency cap exact. Active count zero after break. Retry and timeout policy attached without breaking the pull cadence.

npm run sample:stream

Defaults that don't surprise

Setting	Default	Why
`inParallel`	`1` (sequential)	Auto-concurrency surprises rate-limited APIs. Sequential is correct.
`withRetry`	none	Retrying non-idempotent ops silently is a footgun.
`withTimeout`	none	Cancelling work the user didn't ask to cancel is worse than no timeout.
`onError`	`"fail"`	Matches `Promise.all` intuition. The discriminated `WorkOutput<R>` return type forces explicit handling on the others.

You opt into resilience. Nothing is implicit.

CSP-style channels -- `@workit/core/channel`

work().stream() is the right shape when the producer-consumer relationship is one fluent pipeline. When the producer and consumer are independent tasks running side by side -- fan-in, fan-out, work-queue -- you want a channel.

import { createChannel } from "@workit/core/channel";
import { group } from "@workit/core";

const orders = createChannel<Order>({ capacity: 100 });

await group(async (task) => {
  task(async (ctx) => {
    for await (const o of orderSource()) {
      await orders.send(o, { signal: ctx.signal });
    }
    orders.close();
  });

  task(async (ctx) => {
    for await (const o of orders) {
      await processOrder(o, { signal: ctx.signal });
    }
  });
});

Channel contract, all five rows verified by 11-channel-contract.mjs:

#	Scenario	Bench observation
A	`send` blocks when the channel is full	On a `capacity: 2` channel, the third `send` is still pending after a microtask turn and completes only after a `receive` frees a slot
B	`close()` drains buffered values	`[1, 2, 3]` delivered, then iteration ended cleanly
C	Pending `send` after `close(reason)` rejects	`ChannelClosedError` with `reason: { tag: "shutdown" }`
D	A `signal` cancels a pending `receive`	Pending receive rejects when the controller aborts
E	Capacity validation	`0`, `-1`, `0.5`, `NaN`, `Infinity` all rejected with `RangeError` at `createChannel(...)`

Cancellation composes with the parent scope. If the consumer task throws inside group, sibling cancellation aborts the producer's pending send. The producer's for await exits cleanly through the rejection. No orphaned sends, no leaked consumers, no half-drained buffer.

This is Go's chan with structured-concurrency parents. Kotlin's Channel without coroutines. It fills the gap between "raw async iterator" and "RxJS observable" for owned producer-consumer work.

Bad-batch bisection -- one rotten document doesn't poison the embedding

A real RAG pipeline failure mode: the provider returns 400 for a mixed batch because one of the documents is malformed. With Promise.all, the whole batch fails, the budget is spent on nothing, and the next 99 documents get re-embedded on retry.

WorkIt ships embedAllBisection that splits the failed batch and recovers the good vectors:

// samples/embed-bisection.sample.js
const result = await group(
  async () => embedAllBisection(["alpha", "bad-doc", "gamma"], {
    async embedBatch(inputs) {
      if (inputs.includes("bad-doc")) throw new BadBatchError("provider rejected mixed batch");
      return inputs.map((input) => [input.length]);
    },
  }, {
    batchSize:   3,
    onError:     "continue",
    countTokens: (input) => input.length,
  }),
  { context }
);

// Asserted by the sample:
//   result.results contains the vectors for "alpha" and "gamma"
//   result.errors  contains exactly one entry pointing at "bad-doc"
//   tokensSpent reflects only the successful work

BadBatchError is the contract. Throw it from embedBatch and the helper bisects: split the batch in halves, retry each half, isolate the rotten document, keep the good vectors. Token budget accounting follows the actual successful work -- you don't pay for the failed mixed batch twice.

npm run sample:bisection

This is the difference between "batch job dies at 2 a.m. and the on-call resyncs the warehouse" and "batch job logs the bad ID and keeps going."

Streaming STT with disconnect cleanup (revisited)

Article 1 showed this. Now you can read the backpressure underneath it:

import { transcribeStream } from "@workit/core/ai";

for await (const text of transcribeStream(microphone, {
  async transcribe(chunk, ctx) {
    return provider.transcribe(chunk, { signal: ctx.signal });
  },
}, { signal: socket.signal })) {
  socket.send(text);
}

When the user closes their laptop:

socket.signal aborts.
transcribeStream propagates the abort to the inflight transcribe() body.
The provider's HTTP request aborts at the AbortSignal boundary.
The async generator's finally runs, closing the microphone source.
The for await loop exits.

Tracked sample: sample:stt-disconnect -- disconnects mid-second-chunk, asserts the provider was cancelled, the source was closed, and the cancel reason kind is manual.

How WorkIt's streaming primitives compare

Library	Backpressure	Cancellation	Structured concurrency	Note
WorkIt `work().stream()`	yes producer pauses on consumer	yes via `ctx.signal` and `break`	yes scope-owned	Backpressure between producer and consumer in one pipeline
WorkIt `createChannel`	yes blocking `send`/`receive`	yes via signal + scope cancel	yes scope-owned	Backpressure between independent tasks
Node.js `Readable` stream	yes via `highWaterMark`	partial via `destroy()`	no no scope	No structured cancel propagation
RxJS observable	no by default; pressure operators are opt-in	yes on `unsubscribe`	per-subscription, not per-scope	Different model: events, not owned tasks
`p-queue`	partial (concurrency limit)	no	no	Bounds in-flight, not producer pull
Async generator (raw)	yes pull-based	partial via `return()`	no	No bounded concurrency without manual scaffolding

WorkIt's streaming and channel primitives are the only ones in the table that tie backpressure to ownership -- cancel the scope, the channel closes, the in-flight work aborts, and cleanup runs.

Receipts

node benchmarks/articles/09-stream-1b-lazy.mjs        # naive 281 vs WorkIt 40
node benchmarks/articles/10-stream-slow-consumer.mjs  # producer overshoot 15 vs bound 17
node benchmarks/articles/11-channel-contract.mjs      # 5 channel scenarios
node benchmarks/articles/run-all.mjs                  # full article suite

Production-side gates that back the same primitives:

Claim	Evidence
1 B virtual stream consumed = 25	`sample:1b` produces <= TAKE+CONCURRENCY items, asserted in CI. Reproduced by `09-stream-1b-lazy.mjs`.
1 M item slow-consumer gate	`check:stream-memory` -- heap growth bounded, max active capped, and producer pull remains demand-limited.
Channel backpressure on capacity 2	`11-channel-contract.mjs` verifies the third send blocks until the first receive.
Channel close + drain	`tests/evidence/correctness/runtime-contracts.mjs` verifies buffered values drain before `done: true`.
Channel cancel via signal	Channel contract coverage verifies pending receives reject with the cancel reason.
Channel composes with `group()`	Channel contract coverage verifies producer/consumer pipelines deliver values in order.
`work().inParallel(N)` cap	Property test (`fast-check`): for any (N, total), `maxActive <= N`.
STT disconnect	`sample:stt-disconnect`: provider cancelled, source closed, reason kind = `manual`.

Run them:

npm run sample:1b
npm run sample:stream
npm run sample:embed100k
npm run sample:bisection
npm run sample:stt-disconnect

What's coming

Now you have a producer that paces itself to the consumer, a channel that closes when its scope cancels, and a stream that exits cleanly when the user closes the tab.

Tomorrow we add the next ownership primitive on top: the budget.

A $0.50 CostBudget. A 100,000-token OpenAITokens. A 5-tool-call AgentToolCalls. Atomic across all parallel children. Inheritable through scope context. Shadowed by inner scopes for sub-budgets. Overrun cancels with CancelReason { kind: "budget" } and partial results stay.

The runtime change underneath this is context overlay lookup: 100 .with() calls over a 5,000-key context bag moved from tens of milliseconds in the inline clone baseline to well under the 10 ms gate, without changing a line of public API. The bench in the next article shows the representative timing.

The point is not simply "we have budgets." Many frameworks expose budgets. The stronger claim is budgets that compose with cancellation, race, retry, hedge, fallback, channels, and streams under one ownership tree.

Source, Benchmarks, And Evidence

Source: https://github.com/WorkRuntime/workit
Article source: https://github.com/WorkRuntime/workit/blob/main/articles/04-backpressure-for-streaming-pipelines.md
Reproduce: npm run bench:articles and npm run test:evidence

Cancelling a Task Doesn’t Stop the CPU Work — Killing Non-Cooperative Jobs in worker_threads

AdmilsonCossa — Thu, 21 May 2026 09:04:21 +0000

Last time we showed nine composables that cancel siblings, retry with signal-aware backoff, and hedge tied requests. That's cooperative cancellation – it works when the body checks the signal and the I/O it makes is signal-aware. This article answers the hard question: what happens when code does not cooperate?

Drop this in a worker module:

// benchmarks/articles/lib/spinner.mjs – ignores every signal you throw at it
import { writeFileSync } from "node:fs";

export function spin({ durationMs, markerPath }) {
  const start = Date.now();
  while (Date.now() - start < durationMs) {
    Math.sqrt(Math.random() * 1e6);
  }
  writeFileSync(markerPath, "late-marker-written-by-worker");
  return { completed: true, elapsedMs: Date.now() - start };
}

This is the canonical "non-cooperative" body. Tight loop, no await, no signal.aborted check. In a single-threaded JS runtime, cooperative cancellation cannot stop it from the outside. AbortController can record cancellation, but the callback cannot run until the event loop yields.

import { offload } from "@workit/core/worker";
import { run } from "@workit/core";

await run.scope(async (scope) => scope.spawn(
  offload(spinnerURL, "spin", { durationMs: 5_000, markerPath }, { timeout: "200ms" }),
));

200 ms later: TimeoutError. The worker thread is terminated by the host. The late-marker file does not exist on disk. We stat() for it in CI and fail the gate if it does.

That's the only honest answer to "can JS forcibly stop work". The answer is no on the main thread – but you can move the work to a worker and have the host kill the thread.

WorkIt has both layers. They are labeled honestly.

Bench 07-worker-hard-kill.mjs. 5,000 ms spin loop. 200 ms timeout. Late-marker file written after the loop completes. Timings are representative; the invariant is the marker-file result.

Implementation Settled at Late-marker file on disk

Main-thread AbortController.abort() t=5,001 ms (full duration) yes exists – the abort callback never even fired; the event loop was starved

offload({ timeout: "200ms" }) t=206 ms with TimeoutError no does not exist – and stays absent through an 800 ms grace window

Implementation	Settled at	Late-marker file on disk
Main-thread `AbortController.abort()`	t=5,001 ms (full duration)	yes exists – the abort callback never even fired; the event loop was starved
`offload({ timeout: "200ms" })`	t=206 ms with `TimeoutError`	no does not exist – and stays absent through an 800 ms grace window

The native baseline is the receipt for "AbortController cannot preempt a CPU loop." The body completed all 5 seconds and wrote the marker, and the setTimeout that was supposed to fire controller.abort() at 200 ms could not be delivered because the event loop never returned. The abort callback never ran.

Layer 1 -- Cooperative cancellation, in-process

async function callTool(input, ctx) {
  ctx.signal.throwIfAborted();                              // explicit checkpoint
  const res = await fetch(url, { signal: ctx.signal });     // signal-aware
  ctx.signal.throwIfAborted();                              // checkpoint after I/O
  return res.json();
}

Cooperative cancellation works when the body checks the signal at safe points and the I/O it makes is signal-aware. WorkIt handles the checkpoints automatically through await boundaries inside run.retry, run.timeout, run.race, and the work() builder. You just have to thread ctx.signal into the I/O calls.

This works for 95% of code: HTTP, database, filesystem, streams, child processes, sleeps, channel sends. They all take an AbortSignal.

It does not work for the 5% that does CPU loops, sync crypto, sync JSON.parse of a 200 MB string, or a fitness test in a genetic algorithm.

For that 5%, you need Layer 2.

Layer 2 -- Hard kill at the worker boundary

import { offload } from "@workit/core/worker";

const transcoded = offload(
  new URL("./ffmpeg-transcode.js", import.meta.url),
  "transcode",
  { input, format: "webm" },
  { timeout: "30s" },
);

await run.scope(async (scope) => scope.spawn(transcoded));

offload(...) returns a TaskFn. Spawn it into a scope. The named export runs in a Worker thread. When the timeout fires the worker is terminated -- not signalled, not asked nicely. The host process keeps running. The promise rejects with TimeoutError. If the parent scope cancels first, the worker is terminated with a CancellationError carrying the parent's reason.

What offload accepts:

Local file URLs (new URL("./mod.js", import.meta.url)).
Named export only.
Structured-cloneable input: primitives, arrays, plain objects, Map, Set, Date, RegExp, ArrayBuffer, SharedArrayBuffer, typed array views.

What offload rejects, before the worker spins up:

Remote and inline URL schemes (https:, data:, blob:).
Path traversal segments.
Functions, symbols, class instances, custom-prototype objects – including buried inside Map values, Set members, or cycles.

The worker boundary is covered by unit tests and by tests/evidence/security/worker-boundary.mjs. The two interesting subtleties: Object.create(null) is accepted (a null-prototype object is "plain enough"), and a class with a clean-looking shape is rejected at deep walk because the prototype check runs on the cloneable graph, not just the top level.

Worker offload -- the happy path

The hard-kill is the headline, but the everyday use of offload is mundane CPU work. The repo ships a sample that runs two Fibonacci computations on real worker threads through run.pool:

// samples/worker-offload.sample.js
const moduleURL = new URL("./cpu-worker.sample-worker.js", import.meta.url);

const results = await run.pool(2, [
  offload(moduleURL, "fibonacci", 20),
  offload(moduleURL, "fibonacci", 21),
]);

// Asserted by the sample:
//   results.map(r => r.value) === [6_765, 10_946]
//   results.every(r => r.threadId > 0)

Different OS thread per task. Both results returned. No try/catch around Worker. No parentPort plumbing. Just offload(modURL, "fnName", input) composed through the same run.pool you saw in article 02. The same primitive that terminates a CPU spinner at the worker boundary is also the one you use to take a heavy sync transform off the event loop.

npm run sample:worker

The shield: `run.uncancellable`

Some code must run to completion even when the parent scope is being cancelled. Database commit. Stripe webhook receipt. Distributed lock release. Audit log flush.

import { run } from "@workit/core";

const commit = run.uncancellable(async (ctx) => {
  await db.commit({ signal: ctx.signal });
  await flushReceipt({ signal: ctx.signal });
}, { timeout: "2s" });

await run.scope(async (scope) => scope.spawn(commit));

Inside the shielded body, ctx.signal is a fresh signal local to the shield -- the parent's cancel does not propagate in. The shield has its own bounded lifetime (timeout: "2s"). When the shield finishes, if the parent had cancelled during the shield, the original CancellationError rethrows after the body completes. Cancellation is delayed, not hidden.

What this is not: run.uncancellable is cooperative. It cannot stop a non-cooperative CPU loop inside the shielded body. For that, use offload.

Bench 08-uncancellable-shield.mjs. Three scenarios -- measured.

Scenario What we measure Result

A. Parent cancel mid-body Body started t=1 ms, parent cancelled at t=41 ms, body sleeping 120 ms Body completed naturally at t=136 ms (outlived cancel by 95 ms), bodyObservedAbort: false, outer settled cancelled with reason.kind === "manual"

B. Shield timeout while body runs Shield { timeout: "100ms" }, body sleeping 2,000 ms Body observed abort at ~100 ms, bodyAbortReasonClass === "TimeoutError", outer settled TimeoutError

C. Nested shields, outer scope cancels Inner sleep 80 ms, outer scope cancels at t=20 ms Inner completed at t=92 ms, outer-shield body completed at t=92 ms, outer settled cancelled at t=93 ms with reason.kind === "manual" -- preserved through both shields

Scenario	What we measure	Result
A. Parent cancel mid-body	Body started t=1 ms, parent cancelled at t=41 ms, body sleeping 120 ms	Body completed naturally at t=136 ms (outlived cancel by 95 ms), `bodyObservedAbort: false`, outer settled `cancelled` with `reason.kind === "manual"`
B. Shield timeout while body runs	Shield `{ timeout: "100ms" }`, body sleeping 2,000 ms	Body observed abort at ~100 ms, `bodyAbortReasonClass === "TimeoutError"`, outer settled `TimeoutError`
C. Nested shields, outer scope cancels	Inner sleep 80 ms, outer scope cancels at t=20 ms	Inner completed at t=92 ms, outer-shield body completed at t=92 ms, outer settled `cancelled` at t=93 ms with `reason.kind === "manual"` -- preserved through both shields

A is the "delayed cancel" contract. B is "the shield is bounded by its own timeout, which the body sees as a TimeoutError on its local signal". C is "nested shields don't lose the outer cancel reason."

Cancellation reasons are typed, not strings

type CancelReason =
  | { kind: "user"; message: string }
  | { kind: "deadline"; deadlineAt: number; elapsedMs: number }
  | { kind: "timeout"; timeoutMs: number }
  | { kind: "parent_failed"; error: unknown }
  | { kind: "sibling_failed"; siblingId: TaskId; error: unknown }
  | { kind: "race_lost"; winnerId: TaskId }
  | { kind: "budget"; budgetKey: string; limit: number; spent: number }
  | { kind: "scope_ended" }
  | { kind: "manual"; tag: string; data: unknown };

Every cancellation in WorkIt carries one of these. You can pivot a metric on cancelReason.kind. You can route a runbook on tag. You can build a "why did my agent stop" dashboard with seven buckets and an exhaustive switch. TypeScript will tell you when you forgot a case.

Compare:

controller.abort("user_clicked_stop");                  // string. lossy. arbitrary.
controller.abort(new DOMException("...", "AbortError")); // class with "Abort" name. that's it.

AbortSignal.reason was a stringly-typed escape hatch that won. WorkIt closes it with a discriminated union and tests that every kind is exercised in the suite.

How do other libraries handle non-cooperative work

Library	Cooperative cancellation	Hard kill (CPU loops)	Mechanism
WorkIt	yes signal-aware	yes built-in	`offload({ timeout })` terminates the worker thread
Effection	yes generator ops	no	bring your own worker
Effect-TS	yes fibers	no	bring your own worker
Native `AbortController`	yes	no	the event loop is single-threaded

If you need to kill a sync CPU loop today and you're not on WorkIt, you're hand-rolling worker management – module URL validation, structured-clone classification, timeout-driven termination, parent-cancel propagation, error propagation back to the host. WorkIt's offload is ~50 lines of public surface and the runtime contract is in CI.

Receipts

Two layers. Two benches. One evidence path per claim.

node benchmarks/articles/07-worker-hard-kill.mjs       # main-thread vs offload
node benchmarks/articles/08-uncancellable-shield.mjs   # 3 shield contracts
node benchmarks/articles/run-all.mjs                   # full article suite

Production-side gates that back the same contracts:

Claim	Evidence
Worker hard-kill on CPU spinner	`07-worker-hard-kill.mjs` runs `offload({ timeout: "200ms" })` against the spinner module, asserts bounded rejection, and verifies the late-marker file does not exist.
Worker hard-kill on parent cancel	`tests/evidence/security/worker-boundary.mjs` verifies parent cancellation terminates worker-owned CPU work.
5 concurrent offloads	Worker unit coverage exercises mixed fast and spinning workers without cross-talk between results.
Input validation	`tests/evidence/security/worker-boundary.mjs` verifies remote and executable worker URLs are rejected; unit coverage exercises structured-clone classification.
`run.uncancellable` semantics	`08-uncancellable-shield.mjs` covers parent cancel during body, shield timeout, nested shields, signal isolation, and reason preservation.
`CancelReason.kind` coverage	Every kind in the discriminated union has at least one tracked test that produces it.

The ergonomic version of cooperative cancellation:

await sleep(ms, ctx.signal);             // signal-aware sleep
await fetch(url, { signal: ctx.signal }); // signal-aware fetch

The ergonomic version of hard cancellation:

await scope.spawn(offload(modUrl, "fn", input, { timeout: "Xs" }));

That's the API. Two layers. Honest labels.

What's coming

Tomorrow: backpressure.

You're consuming a billion-row source you'll never materialize. You want to read 25 from the front, run them through a 16-wide map, and have the producer pause when the consumer can't keep up. You want a transcription stream that exits cleanly when the user closes the tab. You want CSP-style channels for the part of your pipeline that's actually a pipeline.

The slow-consumer memory gate runs a million items through a paused consumer in CI and asserts the heap doesn't move. That's the next bench.

Source, Benchmarks, And Evidence

Source: https://github.com/WorkRuntime/workit
Article source: https://github.com/WorkRuntime/workit/blob/main/articles/03-cancellation-and-worker-boundaries.md
Reproduce: npm run bench:articles and npm run test:evidence

AI agents do not fail in one place

AdmilsonCossa — Thu, 14 May 2026 10:00:15 +0000

They fail across concurrency, retries, timeouts, queues, tools, streams, and provider calls.

That is why the JavaScript ecosystem has huge demand for separate async primitives:

Package	Weekly Downloads
`p-limit`	~204M
`p-map`	~53M
`p-timeout`	~36M
`p-retry`	~38M
`async-retry`	~24M
`p-queue`	~23M
`bottleneck`	~10M

These libraries are good.

But the production pain is deeper:

You want concurrency → add p-limit

You want retries → add p-retry

You want timeouts → add p-timeout

You want queues → add p-queue

You want rate limits → add bottleneck

Now each primitive owns a different part of the lifecycle.

None coordinate cancellation together.

When an AI agent fails mid‑flight:

Who stops the retry?

Who clears the timeout?

Who drains the queue?

Who cleans up the tool call?

Who prevents the losing provider from continuing to bill?

WorkIt explores a different model:

work(items)
  .inParallel(8)
  .withRetry(3)
  .withTimeout("5s")
  .do(fn)

One scope. One owner. One cancellation path. One cleanup model.

Concurrency, retry, timeout – under the same ownership tree.

👉 Read the full article:

Concurrency, Retry, and Timeout Under One Owner

npm install @workit/core

AI agents do not fail in one place. They fail across retries, timeouts, tools, and provider calls.

AdmilsonCossa — Mon, 11 May 2026 14:43:53 +0000

Concurrency, Retry, And Timeout Under One Owner.

Last time we showed work(items).inParallel(8).withRetry(3).withTimeout("5s").do(fn) — the one-line fluent surface for processing a list. That handles the 80% case.

Check Owned Async Work in TypeScript — Promise.race does not cancel your work and Your AI Is Still Billing After the User Closed the Tab in case you missed them.

This article is about the other 20%: orchestrating heterogeneous tasks that race, fall back, hedge, and retry — with ownership.

Open package.json in enough AI codebases and you'll see combinations like these – or completely different helpers – depending on who wrote the stack:

"p-limit":      "^5.0.0",
"p-map":        "^7.0.0",
"p-retry":      "^6.2.0",
"p-timeout":    "^6.1.2",
"p-queue":      "^8.0.1",
"bottleneck":   "^2.19.5",
"async-retry":  "^1.3.3"

Six libraries is just one example. Some teams roll their own Promise.race loops. Others use async helpers from lodash or bluebird. The pattern is the same: scattered concurrency primitives with no shared ownership.

When a sibling throws, when a timeout fires, when the user hits stop, you have to stitch together queue state, retry delay, timeout wrapper, underlying I/O, cleanup, and error shape yourself.

That is the comparison in this article: not "those tools are useless", but "they are separate primitives." WorkIt's claim is ownership and composition. The runnable benches at the end of each section verify the WorkIt invariants on your machine.

WorkIt has five core composables, all sharing one runtime contract:

run.all      // Promise.all that actually cancels losers on first failure.
run.race     // Promise.race that actually cancels losers.
run.any      // Promise.any that actually cancels remaining tasks.
run.pool     // p-limit + p-map, but children belong to the scope.
run.series   // sequential, with shared cancellation.

Plus four more that compose with them:

run.retry    // backoff with signal-aware sleep.
run.timeout  // deadline that returns a TaskFn.
run.fallback // primary -> secondary, type-safe.
run.hedge    // bounded speculative execution for tail-latency control.

Same familiar names. Different runtime contract: everything below the call belongs to a scope, and the scope owns the cancel.

The composition property under all nine: every WorkIt resilience helper takes a TaskFn<T> and returns a TaskFn<T>. That makes the algebra closed – run.timeout(run.retry(callProvider, 3), "5s") is just function composition. Promise helpers usually return promises or independent wrapper functions, so crossing from timeout to retry to race means you own the glue and the signal threading.

Why `run.all` improves on Promise.all’s safety

import { run } from "@workit/core";

const [profile, plan, sources] = await run.all([
  (ctx) => fetchProfile({ signal: ctx.signal }),
  (ctx) => planLLM(question, { signal: ctx.signal }),
  (ctx) => retrieveContext(question, { signal: ctx.signal }),
]);

Promise.all rejects on first failure and leaves the other two requests running unless each branch has its own cancellation wiring. Their .then handlers can fire after your error handler already returned a 500, producing completion events that are no longer attached to the owning request.

run.all rejects on first failure and cancels the other two. ctx.signal aborts. defer cleanups run. The reason is typed: CancelReason { kind: "sibling_failed", siblingId, error }.

You can pivot a dashboard on that. You cannot pivot on Error: AggregateError.

Bench 01-run-all-vs-promise-all.mjs. A succeeds at 50 ms. B fails at 30 ms. C succeeds at 100 ms.

Implementation Outer rejected A still ran past reject C still ran past reject Defer ran for losers

Promise.all t=35 ms +16 ms +79 ms n/a

run.all t=32 ms 0 ms (cancelled at +1 ms) 0 ms (cancelled at +1 ms) yes before outer reject

Implementation	Outer rejected	A still ran past reject	C still ran past reject	Defer ran for losers
`Promise.all`	t=35 ms	+16 ms	+79 ms	n/a
`run.all`	t=32 ms	0 ms (cancelled at +1 ms)	0 ms (cancelled at +1 ms)	yes before outer reject

`run.race` – the race that actually races

const winner = await run.race([callOpenAI, callAnthropic, callGemini]);

Six tokens you wrote with Promise.race. Different runtime contract:

Each body receives a ctx.signal linked to the race.
First settlement cancels the rest at the AbortSignal boundary, before TCP completes.
Each loser sees CancelReason { kind: "race_lost", winnerId } — typed, exhaustively narrowed.
await run.race(...) returns only after losers have finished cleaning up.

Bench 02-run-race-vs-promise-race.mjs. Anthropic at 10 ms, OpenAI at 50 ms, Gemini at 80 ms.

Implementation Winner at OpenAI loser still ran Gemini loser still ran Loser reason

Promise.race t=14 ms +47 ms (61 ms total) +77 ms (91 ms total) none

run.race t=17 ms 0 ms (cancelled at t=16 ms) 0 ms (cancelled at t=16 ms) race_lost

Implementation	Winner at	OpenAI loser still ran	Gemini loser still ran	Loser reason
`Promise.race`	t=14 ms	+47 ms (61 ms total)	+77 ms (91 ms total)	none
`run.race`	t=17 ms	0 ms (cancelled at t=16 ms)	0 ms (cancelled at t=16 ms)	`race_lost`

That loser runtime x N parallel agents x P requests per second is the line on your invoice that nobody wrote.

`run.any` – first success, rest cancelled

const cheapest = await run.any([callExpensive, callCheap, callCheaper]);

Promise.any resolves with the first success and ignores the rest. The slower siblings keep running. The faster failing ones got logged and forgotten. run.any does the same – except the slower siblings actually stop.

Bench 03-run-any-vs-promise-any.mjs. A fails at 30 ms. B succeeds at 50 ms. C succeeds at 100 ms.

Implementation Resolved at C kept running Defer ran for C

Promise.any t=61 ms +47 ms (108 ms total) n/a

run.any t=65 ms 0 ms (cancelled at t=65 ms) yes

Implementation	Resolved at	C kept running	Defer ran for C
`Promise.any`	t=61 ms	+47 ms (108 ms total)	n/a
`run.any`	t=65 ms	0 ms (cancelled at t=65 ms)	yes

`run.pool` – bounded concurrency that cancels

const results = await run.pool(8, files.map((file) => async (ctx) => {
  return uploadOne(file, { signal: ctx.signal });
}));

p-limit(8) is a semaphore. That's useful, and current versions can clear pending queue items when you ask them to. What it is not is a structured scope: it does not automatically turn a sibling failure into in-flight cancellation, typed cancel reasons, cleanup, and a partial-result contract.

run.pool(8, tasks) is a semaphore + a scope. Default policy is Promise.all-style fail-fast: first throw cancels queued and in-flight. Results are positionally indexed regardless of completion order. Switch policy with one line and the return type changes so you can't ignore failures:

const out = await work(files).inParallel(8).onError("collect").do(uploadOne);

if (out.mode === "collect") {
  for (const r of out.results) {
    if (r.status === "rejected") logFailure(r.reason);
  }
}

WorkOutput<R> is a discriminated union – mode: "fail" | "continue" | "collect". Change .onError("continue") and the return type forces you to handle errors[]. The compiler is your audit log.

Bench 04-pool-vs-semaphore.mjs. 10 items, concurrency 4. Item 3 throws at 20 ms; the rest take 100 ms each.

Implementation Outer rejected Started Fulfilled AFTER rejection Cancelled Never started Longest post-rejection run

local pLimitLike(4) semaphore baseline t=31 ms 10 9 0 0 +295 ms

run.pool(4, ...) t=33 ms 4 0 3 6 0 ms

Implementation	Outer rejected	Started	Fulfilled AFTER rejection	Cancelled	Never started	Longest post-rejection run
local `pLimitLike(4)` semaphore baseline	t=31 ms	10	9	0	0	+295 ms
`run.pool(4, ...)`	t=33 ms	4	0	3	6	0 ms

295 ms of post-rejection work, multiplied across a fleet, becomes avoidable runtime and provider cost.

`run.retry` – composable, cancel-aware backoff

const callWithRetry = run.retry(callProvider, {
  times: 4,
  backoff: "exponential",
  initialDelay: "200ms",
  maxDelay: "5s",
  jitter: true,
  retryIf: (err) => isTransient(err),
});

const answer = await callWithRetry(ctx);

Three things WorkIt makes part of the retry contract:

Stop retrying on scope cancellation. When the parent scope cancels mid-attempt, run.retry does not enqueue another attempt. The task settles as cancelled, not failed.
Validate input at the boundary. run.retry({ times: 1e9 }) would create an unbounded retry policy. run.retry rejects it: RangeError: retry attempts must be an integer between 1 and 1000. Bound is MAX_RETRY_ATTEMPTS.
Sleep with the scope signal. Backoff sleep is interruptible – abort the signal, the sleep rejects, the loop exits. The benchmark below compares against a signal-unaware retry loop; current retry libraries may expose their own abort hooks, but they still do not own WorkIt's scope tree, cleanup, and cancel-reason contract.

Bench 05-retry-on-cancel.mjs. Body throws on every attempt. External cancel fires around t=50 ms. Up to 8 retries with 50 ms backoff.

Implementation Cancel observed Outer settled Cancel latency Extra attempts after cancel Settled as

signal-unaware retry loop t=63 ms t=701 ms 638 ms 7 rejected

run.retry t=61 ms t=61 ms 0 ms 0 cancelled (kind: manual)

Implementation	Cancel observed	Outer settled	Cancel latency	Extra attempts after cancel	Settled as
signal-unaware retry loop	t=63 ms	t=701 ms	638 ms	7	`rejected`
`run.retry`	t=61 ms	t=61 ms	0 ms	0	`cancelled` (kind: `manual`)

638 ms of wasted retry work after the user already cancelled. Per request. Multiply by the agent fan-out.

`run.timeout`– composes with retry, race, and pool

const fastest = await run.race([
  run.timeout(callPrimary,   "800ms"),
  run.timeout(callSecondary, "800ms"),
]);

run.timeout(task, "800ms") returns a TaskFn. It composes. You can wrap it in run.retry. You can put it inside run.race. You can hand it to run.pool. The signature is closed under composition.

Promise timeout helpers return promises or decorated promises. Some expose AbortSignal support. They still do not return a WorkIt TaskFn, so crossing timeout, retry, race, pool, and cleanup means you own the composition boundary.

`run.fallback` – primary, secondary, type-safe

const callWithFallback = run.fallback(
  run.retry(callProvider, 3),
  callBackupProvider,
);

Primary fails (after retries) -> secondary runs. Same ctx.signal. Same scope. Same cancel reason if the parent stops. No nested try/catch. No "did I forget to await the fallback" Slack message at 2 a.m.

`run.supervise` – restart policy for long-lived work

run.retry is for one operation that may fail transiently and then succeed. run.supervise is for a long-lived task – a heartbeat, a queue consumer, a connection watcher, an agent keep-alive – that may need restart semantics with bounded backoff.

import { run } from "@workit/core";

const result = await run.supervise(async () => {
  attempts++;
  if (attempts < 3) throw new Error("transient worker failure");
  return "stable";
}, {
  restartOn:    "error",
  maxRestarts:  3,
  backoff:      () => 1,
});

// samples/supervision.sample.js – asserted in CI:
//   result === "stable"
//   attempts === 3

The supervised body fails twice, restarts each time under the policy, and stabilises on the third attempt. The parent scope can still cancel everything at once, and the cancel reason carries down through the supervision wrapper. Restart policies cap at maxRestarts per resetWindow so a permanently broken body doesn't infinite-loop.

npm run sample:supervise

The decision rule: use run.retry for a single call that can hiccup; use run.supervise for a process that should keep running.

`run.hedge` – bounded speculative requests

const ranked = await run.hedge(
  (ctx) => reranker.rank(question, sources, { signal: ctx.signal }),
  { after: "2s", max: 2 },
);

If the first call hasn't returned in 2 seconds, fire a second one. First success wins; the rest cancel. Bounded by max, this is a measured way to reduce tail latency without paying for every speculative fan-out.

Bench 06-hedge-tied-requests.mjs. Two scenarios, opts { after: "50ms", max: 3 }.

Scenario Body latency Attempts fired (timestamps) Winner Losers cancelled Cancel reason

slow 200 ms 3 (t=2 ms, 62 ms, 107 ms) id=1 at 217 ms 2 race_lost

fast 30 ms 1 (no hedge fired) id=1 at 31 ms 0 n/a

Scenario	Body latency	Attempts fired (timestamps)	Winner	Losers cancelled	Cancel reason
slow	200 ms	3 (t=2 ms, 62 ms, 107 ms)	id=1 at 217 ms	2	`race_lost`
fast	30 ms	1 (no hedge fired)	id=1 at 31 ms	0	n/a

The fast path doesn't pay for hedging at all. The slow path bounded by max. Every loser tagged with race_lost.

Side-by-side – who actually cancels

// 3 tasks. B fails at 30 ms. A succeeds at 50 ms. C succeeds at 100 ms.

await Promise.all([A, B, C]);     // rejects at 30 ms. A and C keep running for ~16/63 ms.
await Promise.race([A, B, C]);    // rejects at 30 ms. A and C keep running.
await Promise.any([A, B, C]);     // resolves at 50 ms. C keeps running for ~44 ms.

await run.all([A, B, C]);         // rejects at 30 ms. A and C cancelled in 1 ms, defer ran.
await run.race([A, B, C]);        // rejects at 30 ms. A and C cancelled in 0-1 ms.
await run.any([A, B, C]);         // resolves at 50 ms. C cancelled in 0 ms, defer ran.

Same shape. Different contract. Native primitives return a value. WorkIt primitives own the tree underneath the value.

How do other libraries compare

Tool	Cancels on sibling failure	Signal-aware retry	Composable timeout (returns `TaskFn`)	Hedged requests	Bundle
WorkIt	yes	yes	yes	yes built-in	14,175 B / 4,835 B gz for all nine composables
`Promise.all` / `race` / `any`	no	n/a	n/a	n/a	0
`p-limit` + `p-retry` + `p-timeout`	partial/manual wiring	partial/manual wiring	separate abstractions	no	three deps
`RxJS`	yes on `unsubscribe`	partial via operators	yes via operators	no	large
Effection	yes structured (generator ops)	yes	yes	no	medium
Effect-TS	yes structured (fibers + typed `Cause`)	yes	yes	no	large

For simple array processing where nothing else can fail, p-limit is fine. For full-stack apps where you want a broader effect or operation model, Effection and Effect-TS are solid. WorkIt's distinction is narrower: structured-concurrency composition without leaving async/await, with nine composables in one ownership tree and the bundle size shown above.

Receipts

The WorkIt runtime claims above are verified by either npm run verify (the production gate) or npm run bench:articles (the side-by-side suite that produced the representative timing tables in this article).

npm run bench:articles
# full article suite: 19 passed, 0 failed

This article cites benches 01-06 from the full suite. Each bench script is ~100 lines, zero external dependencies, and asserts the WorkIt invariant in-line. Timings are representative captured runs; the assertions guard semantic invariants, not exact milliseconds. The folder has its own package.json so the published package's dependency graph stays empty. Read the README at benchmarks/articles/ for how the promise-helper baselines stay honest.

Production-side gates that back the same composables:

Claim	Evidence
Cancellation safety, all composables	Benches 01-06 plus `tests/evidence/lifecycle/owned-work.mjs` verify parent cancellation, sibling failure, retry cancellation, race loser cleanup, and owned background work.
`run.retry` validation	`RangeError` for `times` <= 0, > 1000, NaN, Infinity, fractional. Identical rejection on numeric and object form.
`run.race` / `run.any` loser cleanup	LIFO `defer` blocks observed in test for every loser; outer promise does not resolve until cleanup completes
Bundle of all nine composables	Included in 14,175 B min / 4,835 B gzip core-group-import. Tree-shaken if unused.

What's next

Nine composables share one engine. Tomorrow we open the engine.

We'll look at what cooperative cancellation can do, what it cannot do, and where the hard boundary starts. Then we put a CPU spin loop that ignores every signal in front of offload({ timeout: "200ms" }) and verify that worker termination prevents a late marker file from appearing. The CI gate runs stat() on it.

AbortController cannot preempt a CPU loop. WorkIt cannot change that language boundary, but a worker thread can be terminated by its host.

Source, Benchmarks, And Evidence

Source: https://github.com/WorkRuntime/workit
Article source: https://github.com/WorkRuntime/workit/blob/main/articles/02-concurrency-retry-timeout.md
Reproduce: npm run bench:articles and npm run test:evidence

Your AI Is Still Billing After the User Closed the Tab

AdmilsonCossa — Sat, 09 May 2026 07:10:11 +0000

That's not a bug. It's a missing owner.

The user closed the browser 30 seconds ago.

Your logs show the response was never delivered. But the LLM stream is still running. The vector search is still scanning. The reranker is still scoring. The tool calls are still executing.

The invoice will arrive tomorrow.

This is not hypothetical. It is the default behaviour of many AI backends today.

app.post("/chat", async (req, res) => {
  const stream = await openai.chat.completions.create({
    model: "gpt-4.1",
    stream: true,
    messages: req.body.messages,
  });

  for await (const chunk of stream) {
    res.write(chunk.choices[0]?.delta?.content ?? "");
  }

  res.end();
});

That code looks reasonable.

It even works — until the user refreshes the page, closes the tab, loses signal, or navigates away.

At that moment:

the HTTP response is dead
the client no longer exists
the user no longer cares

But the async work continues anyway.

The LLM keeps generating tokens. The vector search keeps scanning. Background tasks run with no remaining consumer.

The work outlived the reason it existed.

That sentence is the real problem.

The real root cause: ownership is missing

Most async systems treat cancellation as an optional convention rather than a runtime guarantee.

You can pass an AbortController if you remember.

You can manually wire cleanup if every developer remembers.

You can hope every dependency correctly propagates cancellation.

But structurally, there is no single owner for the tree of async work created by a request.

A single AI request may spawn:

LLM streaming
vector search
reranking
tool execution
background audit writes
metrics
cleanup handlers
retries
observability traces

Every one of these should stop when the user disconnects.

But native Promise does not provide ownership semantics.

So the work survives.

The hidden cost of abandoned AI work

At small scale this is invisible.

At production scale it is expensive.

100,000 abandoned requests/day
× 3–5 seconds of unnecessary downstream execution
= millions of wasted tokens
= unnecessary GPU time
= avoidable API spend
= infrastructure pressure

This is not a code-quality issue.

This is infrastructure waste.

The fix: one scope owns the work

@workit/core introduces one ownership boundary: the scope. Every child task belongs to it. When the scope cancels, everything below it stops with a typed reason and runs registered cleanup before the scope resolves.

import { run } from "@workit/core";

app.post("/chat", async (req, res) => {
  await run.scope(async (scope) => {
    // Client disconnects → cancel everything underneath
    req.signal.addEventListener("abort", () =>
      scope.cancel({ kind: "manual", tag: "client_disconnected" })
    );

    const llm = scope.spawn(async (ctx) =>
      streamLLM(req.body.messages, { signal: ctx.signal }),
      { name: "llm-stream", kind: "llm" });

    const tools = scope.spawn(async (ctx) =>
      runTools(req.body.input, { signal: ctx.signal }),
      { name: "tools", kind: "tool" });

    const vector = scope.spawn(async (ctx) =>
      searchVectorDB(req.body.query, { signal: ctx.signal }),
      { name: "vector-search", kind: "io" });

    const [text, toolResult, sources] = await Promise.all([llm, tools, vector]);
    res.json({ text, toolResult, sources });
  });
});

Now the request owns the work.

When the client disconnects:

scope.cancel({ kind: "manual", tag: "client_disconnected" }) fires.
Every child task receives ctx.signal.aborted with the typed reason.
The OpenAI stream call sees its signal abort and stops at the TCP layer.
Tool execution stops.
Vector search stops.
Registered ctx.defer(...) cleanup runs LIFO.
The scope resolves deterministically.

No orphaned work. No zombie tasks. No invisible token burn.

streamLLM, runTools, and searchVectorDB above are your application's wrappers — anything that accepts an AbortSignal (the official OpenAI SDK, the Vercel AI SDK, pg, mongodb, fetch) plugs straight in.

This is not theory

The repository ships a runnable sample for exactly this scenario:

// samples/stt-disconnect.sample.js — same shape, applied to live audio
const disconnect = new AbortController();
const iterator = transcribeStream(microphone, {
  async transcribe(chunk, ctx) {
    return provider.transcribe(chunk, { signal: ctx.signal });
  },
}, { signal: disconnect.signal })[Symbol.asyncIterator]();

await iterator.next();                // first chunk: "FIRST"
const pending = iterator.next();      // second chunk starts
disconnect.abort(new CancellationError({ kind: "manual", tag: "client_disconnect" }));

When the sample runs, it asserts and prints:

{
  "sample": "stt-disconnect",
  "first": "FIRST",
  "providerCancelled": true,
  "sourceClosed": true,
  "reasonKind": "manual"
}

The receipt is what did not continue running:

providerCancelled: true — the provider's HTTP request observed the abort.
sourceClosed: true — the async generator's finally ran and the microphone source closed.
reasonKind: "manual" — the cancel reason was typed end to end. Your dashboard can pivot a metric on it.

Reproduce: npm run sample:stt-disconnect. The sample is at samples/stt-disconnect.sample.js.

Why this matters

The AI ecosystem is rapidly moving toward:

streaming
agents
tool execution
multi-provider inference
background workflows
long-lived realtime sessions

All of these create trees of async work. Most of them still lack clear ownership semantics.

The result is:

abandoned compute
leaking streams
runaway retries
zombie tool execution
incomplete shutdowns
hidden infrastructure cost

WorkIt treats async work as something that must have an owner.

When the owner disappears, the work disappears with it.

The deeper idea

This is not really about cancellation.

It is about lifecycle ownership.

The request should own the work it creates.

The WebSocket should own its subscriptions.

The agent should own its tools.

The stream should own its producers.

Without ownership, async systems slowly leak compute and complexity.

Try it

npm install @workit/core

import { run } from "@workit/core";

await run.scope(async (scope) => {
  req.signal.addEventListener("abort", () =>
    scope.cancel({ kind: "manual", tag: "client_disconnected" })
  );

  const result = await scope.spawn(async (ctx) =>
    openai.chat.completions.create({
      model: "gpt-4.1",
      stream: true,
      messages,
      signal: ctx.signal,            // pass ctx.signal — the chain handles the rest
    }),
    { name: "ai-call", kind: "llm" }
  );
});

That's the contract. One scope owns the work. The OpenAI client receives the signal. The tab closes; the bill stops.

The larger problem

Every senior engineer has seen some version of this question in production:

"Why is this still running?"

That question appears in:

AI streaming
WebSockets
Kafka consumers
background workers
multiplayer game loops
Discord bots
server-side rendering
agent runtimes

The problem is the same every time:

The work outlived the reason it existed.

WorkIt is an attempt to fix that at the runtime level.

The series

Owned Async Work in TypeScript — Promise.race does not cancel your work
You are here — Your AI is still billing after the user closed the tab
Nine composables. One ownership contract. — coming this week
AbortController cannot preempt a CPU loop. WorkIt uses a worker boundary. — coming this week
A 1,000,000,000-row pipeline. 25 consumed. The producer noticed. — coming
A 50¢ agent. A connection that closes on Ctrl-C. — coming
100K agent runs a day. Bounded observability cost without core bloat. — coming
An agent loop in 12 lines. A typed tool contract. A 50¢ ceiling. — coming

Sources and reproducibility

GitHub: github.com/WorkRuntime/workit
The disconnect sample: samples/stt-disconnect.sample.js
The bench suite: node benchmarks/articles/run-all.mjs reports passed: 19, failed: 0
The production gate: npm run verify — typecheck, 214 unit tests at 100% coverage, no-network gate, vulnerability audit, SBOM, bundle gate, soak, exporter stress, package-consumer fixtures across runtimes

Owned Async Work In TypeScript. Promise.race Does Not Cancel Your Work

AdmilsonCossa — Fri, 08 May 2026 17:35:22 +0000

Why Promise.race, Promise.all, and async helpers need an ownership model for cancellation, cleanup, and production agent work.

Three providers. One winner. Three invoices.

const winner = await Promise.race([
  fetch(OPENAI,    { body }),
  fetch(ANTHROPIC, { body }),
  fetch(GEMINI,    { body }),
]);

It is easy to assume a race cancels the losers. Native Promise.race resolves on the first settlement and leaves every other promise running unless the caller wires cancellation into each branch. TCP completes. Tokens bill. .then callbacks fire. Cleanup remains a manual convention. Promise.any and Promise.all have the same ownership gap.

Native promises model values, not ownership. There is no ownership parent, no built-in cancellation path, and no scope cleanup contract.

There is still work running after the value settles.

The one line that fixes the 80% case

Before we fix racing, fix the thing every codebase does first: process a list of items with concurrency, retries, and a timeout.

import { work } from "@workit/core";

await work(items)
  .inParallel(8)
  .withRetry(3)
  .withTimeout("5s")
  .do(async (item, ctx) => {
    ctx.report({ message: `processing ${item.id}` });
    return apiCall(item, { signal: ctx.signal });
  });

That isn't a custom Promise chain. It isn't a standalone concurrency helper where cancellation, retry, timeout, and cleanup have to be wired separately. It's a runtime contract:

At most 8 inflight at any instant. The cap is a hard property test, not a hint.
Each item retries up to 3 times, exponential backoff with jitter, signal-aware sleep.
Any item exceeding 5 seconds cancels its own operation with TimeoutError.
First uncaught failure cancels the queued and the in-flight, by default. Switch policy on a single line and the return type changes so you can't ignore failures:

const out = await work(items).inParallel(8).onError("collect").do(fn);
//    ^^^ WorkOutput<R> -- discriminated union: "fail" | "continue" | "collect"
if (out.mode === "collect") {
  for (const r of out.results) if (r.status === "rejected") log(r.reason);
}

Every body receives a ctx.signal linked to the scope, so the fetch, the database query, or the LLM call actually aborts at the I/O boundary.
Progress events flow to your logger, metrics, or UI through ctx.report(...) -- zero allocation when nobody listens.

That is the surface. Five chained methods. No new vocabulary. Now stack it.

`run.race`: same shape, different contract

import { run } from "@workit/core";

const winner = await run.race([callOpenAI, callAnthropic, callGemini]);

Same six tokens you wrote with Promise.race. Different runtime contract:

Each body receives a ctx.signal linked to the race.
First settlement cancels the rest at the AbortSignal boundary, before TCP completes.
Each loser sees CancelReason { kind: "race_lost", winnerId } -- typed, exhaustively narrowed, not a string.
Each loser's ctx.defer(...) cleanup runs LIFO before run.race resolves.
await run.race(...) returns only after losers have finished cleaning up.

That is the first ownership boundary. Now stack it again.

Cancel a 200-tool agent on client disconnect

import { run } from "@workit/core";

await run.scope(async (scope) => {
  request.signal.addEventListener("abort", () =>
    scope.cancel({ kind: "manual", tag: "client_disconnect" }));

  for (const step of plan.steps) {
    scope.spawn(async (ctx) => callTool(step, ctx.signal),
      { name: step.name, kind: "tool" });
  }
  scope.spawn.background(async () => auditLog(plan));
}, { deadline: "30s" });

Every in-flight tool aborts. Every ctx.defer runs LIFO. The audit task flushes. The reason -- { kind: "manual", tag: "client_disconnect" } -- carries down the tree so your dashboard distinguishes a stop from a deadline from a budget overrun.

A socket close cancels an STT stream and closes the microphone

import { transcribeStream } from "@workit/core/ai";

for await (const text of transcribeStream(microphone, {
  async transcribe(chunk, ctx) {
    return provider.transcribe(chunk, { signal: ctx.signal });
  },
}, { signal: socket.signal })) {
  socket.send(text);
}

Socket disconnects. ctx.signal aborts. The provider's HTTP request aborts. The async generator's finally runs and closes the microphone. The sample asserts that the source closes and no provider call remains active after disconnect.

100,000 documents under a hard token cap

import { group, run } from "@workit/core";
import { OpenAITokens, embedAll } from "@workit/core/ai";

await run.context.with(
  OpenAITokens, { spent: 0, limit: 1_000_000, unit: "tokens" },
  () => group(() => embedAll(documents, { concurrency: 32 })),
);

Bounded concurrency. Per-item retry. Token budget enforced atomically across all 32 inflight workers. Blow the cap mid-pipeline and the scope cancels with CancelReason { kind: "budget", limit, spent }. Partial results stay. The rest abort.

No Orphans Means No Unowned Background Work

A background child is still scoped. The parent operation does not finish while owned background work keeps running. The receipt is one of the smallest samples in the repo:

// samples/no-orphan.sample.js
const result = await group(async (task) => {
  task.background(async (ctx) => {
    await sleep(20, ctx.signal);
    backgroundCompleted = true;
  });
  return "body-returned";
});

// Asserted by the sample:
//   result === "body-returned"
//   backgroundCompleted === true
//   elapsedMs >= 15

The body returns its value at t=0. The owned background task takes 20 ms. The await group(...) does not resolve until both finish. If you want to escape the scope, you call run.detached(...) and accept the orphan trade-off explicitly. There is no third option.

npm run sample:no-orphan

Why not just use X

The right tool depends on what part of the lifecycle you actually own.

Tool	Bounded concurrency	Scope-owned loser / sibling cancellation	Typed cancel reason	Scope cleanup
WorkIt	yes `work().inParallel(N)` / `run.pool(N, ...)`	yes at the `AbortSignal` boundary	yes `CancelReason` discriminated union	yes `ctx.defer` LIFO
`Promise.all` / `race` / `any`	no	no	no	no
`p-limit`	yes	manual; queue ownership is separate from task cancellation	no	no
`p-map`	yes	partial/manual; queue and in-flight work have separate policies	no	no
`RxJS.mergeMap`	yes	yes on unsubscribe	partial	per-subscription, not per-scope
Effection	yes via generator ops	yes (structured)	partial	yes
Effect-TS	yes via fibers	yes	yes (typed `Cause`)	yes

If your problem is "process this array with N concurrency" and nothing else ever fails, p-limit is fine. If your problem is "this list is part of a request that can time out, the user can disconnect, and one bad item must cancel the rest with cleanup", you want a runtime contract. Effection and Effect-TS provide one -- through generators and a fiber DSL respectively. WorkIt provides one without leaving async / await.

Receipts

The release-readiness claim above is a CI gate, not a tagline. Each row maps to a command in npm run verify.

Measurement	Value	What it includes
`core-group-import` bundle	*14,175 B min 4,835 B gzip**	The full `group` + `run` + `work` + retry/timeout/race/all/any/pool surface, tree-shaken
Runtime dependencies	0	Zero. The compiled core does not import `node:http`, `node:https`, or `fetch`. Static check enforced.
Tests / coverage	*214 tests 100% statements / branches / functions / lines**	Cancellation invariants, channel semantics, AI-subpath mocks, exporter stress, scope tree, budget atomicity
Hot-path heap, 100k tasks, no signal read	0.9 MB post-GC	Was 298 MB before lazy `AbortController` allocation -- ~330x reduction
Tracked soak gate, 100k tasks @ concurrency 128	126,136 B max heap growth	The `npm run check:soak` gate fails the build if this regresses
Stream backpressure, 1,000,000 logical items, slow consumer	maxActive <= inParallel(N), producer paused, heap bounded	The `npm run check:stream-memory` gate
`offload({ timeout: "200ms" })` against an infinite CPU spin loop	rejects at the worker timeout boundary, late-marker file does not exist	AbortController cannot preempt a CPU loop; the worker is terminated at the host boundary. CI `stat()`s for the marker.
Claim evidence suite	`npm run test:evidence`	Curated lifecycle, correctness, security, release, and performance proofs mapped in `evidence/claims.json`

The series

You are here -- Promise.race does not own the losing work. The fluent surface and why ownership matters.
Nine composables. One ownership contract.
AbortController cannot preempt a CPU loop. WorkIt uses a worker boundary.
A 1,000,000,000-row pipeline. 25 consumed. The producer noticed.
A 0.50 USD agent. A connection that closes on ctrl-C. A receipt the user never sees.
100K agent runs a day. Bounded observability cost without core bloat.
An agent loop in 12 lines. A typed tool contract. A 50-cent ceiling.

Try it

npm install @workit/core

The API is stable. The tests pass. The bundle is tiny.

Next: run.all, run.race, run.any, run.pool, run.series side-by-side with Promise.all, Promise.race, Promise.any, p-limit, and p-map. We measure which contracts still hold when one sibling throws mid-flight.

Source, Benchmarks, And Evidence

Source: https://github.com/WorkRuntime/workit
Article source: https://github.com/WorkRuntime/workit/blob/main/articles/01-owned-async-work.md
Reproduce: npm run bench:articles and npm run test:evidence

DEV Community: AdmilsonCossa

Don't let a billion RAG docs drown your 25-result pipeline

Backpressure For Streaming Pipelines

work().stream() -- bounded, lazy, cancellable

Streaming map: stop after 12, produce only what demand requires

Defaults that don't surprise

CSP-style channels -- @workit/core/channel

Bad-batch bisection -- one rotten document doesn't poison the embedding

Streaming STT with disconnect cleanup (revisited)

How WorkIt's streaming primitives compare

Receipts

What's coming

Source, Benchmarks, And Evidence

Cancelling a Task Doesn’t Stop the CPU Work — Killing Non-Cooperative Jobs in worker_threads

Layer 1 -- Cooperative cancellation, in-process

Layer 2 -- Hard kill at the worker boundary

Worker offload -- the happy path

The shield: run.uncancellable

Cancellation reasons are typed, not strings

How do other libraries handle non-cooperative work

Receipts

What's coming

Source, Benchmarks, And Evidence

AI agents do not fail in one place

AI agents do not fail in one place. They fail across retries, timeouts, tools, and provider calls.

Why run.all improves on Promise.all’s safety

run.race – the race that actually races

run.any – first success, rest cancelled

run.pool – bounded concurrency that cancels

run.retry – composable, cancel-aware backoff

run.timeout– composes with retry, race, and pool

run.fallback – primary, secondary, type-safe

run.supervise – restart policy for long-lived work

run.hedge – bounded speculative requests

Side-by-side – who actually cancels

How do other libraries compare

Receipts

What's next

Source, Benchmarks, And Evidence

Your AI Is Still Billing After the User Closed the Tab

The real root cause: ownership is missing

The hidden cost of abandoned AI work

The fix: one scope owns the work

This is not theory

Why this matters

The deeper idea

Try it

The larger problem

The series

Sources and reproducibility

Owned Async Work In TypeScript. Promise.race Does Not Cancel Your Work

The one line that fixes the 80% case

run.race: same shape, different contract

Cancel a 200-tool agent on client disconnect

A socket close cancels an STT stream and closes the microphone

100,000 documents under a hard token cap

No Orphans Means No Unowned Background Work

Why not just use X

Receipts

The series

Try it

Source, Benchmarks, And Evidence

`work().stream()` -- bounded, lazy, cancellable

CSP-style channels -- `@workit/core/channel`

The shield: `run.uncancellable`

Why `run.all` improves on Promise.all’s safety

`run.race` – the race that actually races

`run.any` – first success, rest cancelled

`run.pool` – bounded concurrency that cancels

`run.retry` – composable, cancel-aware backoff

`run.timeout`– composes with retry, race, and pool

`run.fallback` – primary, secondary, type-safe

`run.supervise` – restart policy for long-lived work

`run.hedge` – bounded speculative requests

`run.race`: same shape, different contract