Reference

@uniflowed/effect

// @flow
import { effect, fail, retry, runPromise, sleep } from "@uniflowed/effect";

const fetchUser = (id: string) =>
  effect(function* () {
    const response = yield* tryPromise({
      try: () => fetch(`/users/${id}`),
      catch: () => ({ kind: "offline" }),
    });
    if (!response.ok) {
      return yield* fail({ kind: "notFound", id });
    }
    return yield* promise(() => response.json());
  });

await runPromise(retry(fetchUser("7"), { kind: "exponential", baseMillis: 100 }));

The three channels

An Effect<A, E, R> produces an A, may fail with an E, and needs an R. E and R default to empty, so Effect<number> reads as cannot fail, needs nothing.

The channels accumulate as a union rather than by subtraction, because Flow has no type-level set difference. flatMap(a, b) is Effect<B, E1 | E2, R1 | R2>, and a generator body accumulates the same way: Flow infers a generator's yield type as the union of everything yielded, so a pipeline's failure type is the sum of its steps' without anyone writing it down.

Failure is not the same as a defect

Three ways an effect can end without a value, and keeping them apart is the point of the package.

retry only retries a fail, because running a bug again runs the bug again. catchAll only catches a fail, because catching a defect is how a bug gets reported as a handled condition. orDie moves a failure into the defect channel when nobody downstream is going to handle it.

Running

An effect with an asynchronous step run under runSync fails as a defect rather than handing back a promise nobody is awaiting.

Fibers, and what a fiber owns

fork starts work beside the current fiber and hands back a handle. The child's lifetime is contained in its parent's, in both directions:

• interrupting the parent interrupts the child, and
• a parent that ends — returns, fails, or is interrupted — stops whatever it still had running.

const handler = effect(function* () {
  const refresh = yield* fork(refreshCache);
  const page = yield* render();
  yield* interrupt(refresh); // optional: it would be stopped anyway
  return page;
});

A cancelled request cannot leave a background refresh writing to a connection that is already closed, and a handler that returns early cannot either.

forkDaemon is the way out, for work whose lifetime genuinely is not the caller's — a cache warmer, a metrics flush. A daemon outlives its parent, so the handle forkDaemon returns is the only thing that can ever stop it. It is a separate name rather than an option because the two answers fail in opposite directions: a caller who wanted a daemon and got a child sees their work stop and goes looking, while a caller who wanted a child and got a daemon sees nothing at all until the process will not exit.

A daemon is detached from its parent, not from its own children — otherwise one forkDaemon would detach a whole subtree.

forkScoped is the third lifetime, and the one a request handler usually means: outlive this fiber, die with this request. The child is tied to the enclosing Scope rather than to a fiber, so a background refresh finishes its own work and stops when the scope closes.

await runPromise(
  scoped(
    effect(function* () {
      // Not stopped when this generator returns, and not left running for ever.
      yield* forkScoped(refreshCache);
      return yield* handle(request);
    }),
  ),
);

The scope's finalizer interrupts the child and waits for it to stop, so a scoped block has not returned until what it started has finished and released. The child gets a scope of its own for its own resources: finalizers run newest first, so registering them on the enclosing scope would release them before the finalizer that stops the child — a resource closed under a fiber still using it.

Effect's fourth fork, forkIn, is deliberately absent. It needs a Scope that is a value, and Scope here is a phantom that is never constructed, which is what makes scoped the only thing that can close one. A forkScoped inside a nested scoped says the same thing with the nesting visible in the code.

all, race and timeout open child fibers under the same rule, which is why a failing all stops its siblings before it returns and a timeout stops the effect it timed out on.

Interruption is cooperative

The flag is read between steps, never inside one: an effect that has started runs to its own end, and interruption decides whether the next step begins. A pending sleep is the exception — it registers a waker, so cancelling a fiber that is asleep for a minute takes effect now rather than in a minute.

interrupt returns the fiber's Exit rather than void, so a caller can tell "cancelled in time" from "too late, it had finished".

What two fibers can share

Starting and stopping fibers is the easy half of concurrency. Three types cover the coordinating half, and all three are the same waiting mechanism with a different list attached — which is why a fiber blocked on any of them can still be interrupted.

The operations are flat functions — refGet, deferredAwait, withPermit — rather than methods or namespaces, which is this package's convention throughout: nothing to dispatch at run time, and a bundler drops what you do not call.

Ref

const total = effect(function* () {
  const counter = yield* ref(0);
  yield* forEach(items, (item) => refUpdate(counter, (n) => n + item), { concurrency: 4 });
  return yield* refGet(counter);
});

ref, refGet, refSet, refModify, refUpdate, refUpdateAndGet, refGetAndUpdate, refGetAndSet — all with synchronous kernels, so a program that uses a Ref can still be answered by runSync.

refModify reads, computes and writes without yielding, so it needs no lock. A transform that throws leaves the ref alone and becomes a defect: half of a read-modify-write is worse than none of it.

An effectful update does yield, so it needs one. refUpdateEffect(ref, lock, transform) is Effect's SynchronizedRef — a Ref and a Semaphore held together rather than a third type. Passing the lock in makes the serialisation visible, and lets two refs that must move together share one.

Deferred

Fiber covers "wait for what this fiber returns". Deferred covers the other one: waiting for a value nobody has promised yet.

const handshake = yield* deferred();
yield* fork(effect(function* () {
  yield* deferredSucceed(handshake, yield* openConnection());
}));
const connection = yield* deferredAwait(handshake); // interruptible

It completes at most once: the first deferredSucceed or deferredFail returns true and every later one returns false, rather than overwriting an answer somebody may already have acted on. deferredIsDone asks without waiting.

deferredAwait registers a waker, the same way a pending sleep does, so interrupting a fiber blocked on a handshake stops it now. A hand-written Promise in the same place would not — which is the reason this type exists rather than the convenience.

Semaphore

all's concurrency bounds one call. A Semaphore bounds a budget shared across call sites, which is what a rate-limited API needs:

const budget = yield* semaphore(2);
const request = (item) => withPermit(budget, fetchItem(item));
// Two forEaches, one budget of two between them.
yield* all([forEach(pageOne, request), forEach(pageTwo, request)]);

withPermit and withPermits give the permits back however the body ends — success, failure, defect, interruption — with the same guarantee ensuring gives. A permit that leaked on cancellation would shrink the budget every time somebody cancelled a request, until nothing could run at all.

The queue is served strictly from the head, so a later small request cannot overtake a waiting large one. Asking for more permits than the semaphore has is a defect rather than a wait, because nothing will ever release enough and a permanent hang is the least debuggable way to say so.

Queue

The same mechanism with a buffer in front, and the one the rest was waiting on: all collects into an array, so a producer that outruns its consumer grows one with no seam to put a limit in.

const work = yield* queue(64);               // bounded: the producer waits
const readings = yield* queue(1, "sliding"); // the newest reading, never a wait

queueOffer, queueTake, queueTakeAll, queueTakeUpTo, queueSize, queueShutdown and queueIsShutdown. Both ends are served strictly from the head, so a later taker cannot overtake a waiting one.

A fiber blocked in either direction can be interrupted out of it, and an interrupted taker leaves the value it was waiting for for the next taker — it is out of the queue of takers before anything can hand it one, so a cancelled request does not eat a piece of work nobody then does.

Every operation but queueShutdown and queueIsShutdown reports an interruption on a queue that has been shut down, waiting or not. A shutdown is a decision somebody took, which is what interrupt already means and what fail does not.

queueOffer and queueTake have synchronous kernels, unlike deferredAwait: taking from a queue with something in it is reading a buffer, not waiting for a value nobody has produced. One that would wait fails the way any other asynchronous effect does under runSync — and it is not a race it might have won, because under runSync nothing else is running to fill the queue.

PubSub

The same buffer, with several subscribers. A subscription is a Queue, so queueTake and queueTakeAll are already the operations a subscriber needs.

const topic = yield* pubSub(16);
const listener = yield* pubSubSubscribe(topic); // needs a Scope
yield* pubSubPublish(topic, tick);

pubSubSubscribe requires a Scope, and it is not decoration: a subscription nothing unsubscribes is a buffer a publisher keeps filling and nobody drains, which with bounded stops the publisher for ever. A subscriber sees what is published after it subscribed and nothing before it, because its buffer is its own and starts empty. With bounded, the publisher waits for the slowest subscriber.

STM is deliberately not planned: it needs a transaction log and a retry-on-conflict scheduler, which is larger than everything above put together, and its uses can be written with a serialised Ref at a cost worth measuring before it is paid.

Resources

acquireRelease puts Scope in the requirement channel, so an effect that acquires something cannot be run until somebody has said where the release belongs. scoped is what says it, and it closes the scope however the effect ends — success, failure, defect, interruption.

Finalizers run in a detached fiber, because a cleanup that is itself cancelled is not a cleanup.

ensuring is the same guarantee without the Scope, for the ordinary "close this when the block is done" case.

Services

A Tag is both the name of a service and the effect that reads it, so yield* Clock inside a generator produces the service and there is no second accessor to learn. provideService satisfies one requirement with a value; provide satisfies them by building a Layer.

Requirement subtraction is the experimental part. Flow has no type-level set difference, so provide is typed as (Effect<A, E, R | Out>, Layer<Out, …>) => Effect<A, E, R> and the checker solves for R. That works when the provided service is a distinct member of the union, and silently leaves it in R when it is not. layerProvide and layerScoped subtract the same way and carry the same caveat.

Layers compose

layerProvide is what makes a layer's In mean something. Without it, a Layer<Database, ConfigError, Config> could only be used by an effect that still required Config, so the requirement was a label rather than a debt anything could pay.

const configLayer = layerEffect(Config, readConfigFile);
const databaseLayer = layerProvide(layerEffect(Database, openPool), configLayer);
const loggerLayer = layerProvide(layerEffect(Logger, openLog), configLayer);

// Config is read once, not twice.
await runPromise(provide(program, layerMerge(databaseLayer, loggerLayer)));

One build per provide

A layer graph is a graph, not a tree: Database and Logger both wanting Config is the ordinary shape. One provide is one build pass with one memo keyed by layer identity, so a layer reached twice is built once — a pool is opened once, and a config file read once gives one answer.

The memo lives for one build and not for the process, so a layer handed to two provides is built twice. That is right for what provide is — one build, one scope, closed when the effect ends — and wrong for what an application wants, which is managedRuntime.

One build for an application

const runtime = await runPromise(managedRuntime(appLayer));
// …one connection pool, many requests…
await runtimeRunPromise(runtime, handle(request));
await runPromise(runtimeDispose(runtime));

Effect-TS calls this ManagedRuntime.make. The build is the same one pass with the same memo; the difference is entirely in who closes the scope and when. runtimeRunPromise, runtimeRunPromiseExit, runtimeRunSync, runtimeRunSyncExit and runtimeRunFork mirror the five runners, each with a root fiber of its own so two effects running against one runtime do not own each other.

This is the one effect in the package that deliberately leaves something open when it returns, which is why dispose is a named function rather than a finalizer somebody might not have registered. It is idempotent, and every run against a disposed runtime is a defect: the services are gone, so asking for one is a bug in the shutdown order rather than a condition to recover from.

Runtime<R> is invariant in R, unlike Effect and Layer. With a covariant one the checker could satisfy a run by widening both sides to a union containing a service the runtime does not have; pinned to what the layer produced, the effect has to be a subtype of it.

A memoised layer keeps the services it was first built with, so the same layer under two different layerProvides in one graph gets one of the two outers. The answer is two layers rather than one used twice.

Layers do not force a program asynchronous

provide has a synchronous kernel, so runSync still answers for a program that uses a layer — which matters most in a test, where runSync earns its keep. A layer whose build genuinely has to wait fails the same way any other asynchronous effect does under runSync: a defect saying so, rather than a promise nobody is awaiting.

A layer that acquires something

layerScoped builds its service with an effect that may acquireRelease. The resource is released when the provide using it ends — on success, on failure, and on interruption — rather than when the build finishes, which is the whole reason a connection pool belongs in a layer.

That scope is the layer's, not the body's. provide deliberately does not hand the body a scope: doing so would silently discharge the Scope an acquireRelease inside the body requires while the requirement channel went on saying otherwise. scoped is still the only thing that answers for an effect's own resources.

Retrying

A Schedule<In> is a state machine written as data: scheduleStep takes a state and an input and answers with a decision — continue, with an output and a delay, or stop. One type drives both retry, whose input is the error, and repeat, whose input is the value. It is importable on its own from @uniflowed/effect/schedule and testable with no runtime at all.

fixed is a period and spaced is a gap: a poll that took 400ms of a one-second period waits 600ms, where a one-second gap would run it every 1.4 seconds. windowed is the same period aligned to boundaries from the start, so an attempt that overran one window waits for the next boundary rather than starting immediately.

Reach for jittered on anything a crowd of clients runs. Without it, every client that failed at the same moment retries at the same moment, for ever, and maxDelay does not help — it caps the wait, it does not spread it. The default range is 80% to 120% of what the inner schedule said.

Time and randomness are arguments, not imports

scheduleStep is handed the clock and the random factor rather than reading either, which is what keeps a policy a pure function and testable across its whole behaviour without a clock to stub, a sleep to flake or a seed to hope about:

const backoff = { kind: "jittered", schedule: { kind: "exponential", baseMillis: 100 } };
const start = scheduleStart(0);
scheduleStep(backoff, start, error, 0, 0); // delayMillis 80  — the bottom of the range
scheduleStep(backoff, start, error, 0, 1); // delayMillis 120 — the top
scheduleStep(backoff, start, error, 0); //    delayMillis 100 — the midpoint

retry and repeat read Date.now() once per decision and pass it in.

Why the output is a number

Effect's type is Schedule<Out, In, R>. R is deliberately absent — a schedule that needs a service forces the runtime dependency back into the module — and Out is absent because Flow cannot check it.

A schedule here is a literal, and a union arm cannot constrain a type parameter it does not mention: with an Out parameter, { kind: "recurs", times: 2 } would inhabit Schedule<string, In> as happily as Schedule<number, In> and repeat would promise a string where the run time has a number. A phantom field has to be optional for the literal to stay writable, and an optional phantom constrains nothing.

In has no such problem, and the difference is variance rather than luck: In is contravariant, so an arm that ignores its input genuinely is usable at any input type. The parameter that can be checked is there and the one that cannot is not.

The loss is real — Schedule.map over an output type is not expressible — and it buys one thing back: compose has to name the type between two schedules, and can only because the output type is fixed. Effect's mapInput is absent for exactly the reason that would have made compose absent too: it needs a type variable for the intermediate input, and a union arm cannot bind one.

Which failures are worth another attempt

A Schedule<E> can read the error, so whileInput says this inside a policy — one a config file could name. retry also takes an optional third argument for the condition the call site knows, and both have to allow an attempt for one to happen:

retry(request, backoff, { while: (error) => error.kind !== "forbidden" });

while keeps going only while the predicate holds; until stops as soon as it does. Both compose with the line the runtime already draws — a defect is a bug and an interruption is a decision already taken, so neither is ever retried. A predicate that throws becomes a defect rather than one more attempt.

Effect takes { schedule, while, until, times } in one argument. Flow's objects are exact, so a union of a schedule and an options object cannot be refined by reading a property one of them lacks; a third parameter is the shape the checker can follow. times is absent for a different reason: it is intersect with recurs, which the schedule union already says.

repeat

The other half of what a schedule is for: run again on success, on a timetable. A poll, a heartbeat, a cache refresh.

const poll = repeat(retry(checkHealth, backoff), { kind: "spaced", millis: 1000 });
const fiber = runFork(poll);
// …later
await runPromise(interrupt(fiber));

The value is the schedule's input, so "poll until the job reports finished" is a schedule rather than a loop:

await runPromise(
  repeat(pollJob, {
    kind: "untilInput",
    schedule: { kind: "spaced", millis: 1000 },
    predicate: (job) => job.finished,
  }),
);

A failure ends the repetition and is the result — an effect that failed produced no value to repeat from, and swallowing it to keep polling would hide the outage the poll exists to notice. retry is what tolerates a blip; the two compose.

Interruption is checked before each wait and after it, so a fiber polling once a minute stops when it is cancelled rather than at the top of the next minute.

repeat gives back the schedule's output — the number of repetitions, the elapsed time, whichever number its last decision reached — which is what Effect's does. The effect's own last value is one tap or one Ref away, and was never the interesting half: a poll that ran eleven times wants to say eleven.

Streams

An Effect produces one value. @uniflowed/effect/stream is for the programs that produce many — a large file, a paginated API, a response body — without buying the whole thing into memory first, and without leaving the error channel, the interruption protocol and Scope behind to do it.

import { streamFromReadableStream, streamRunForEach, streamTake } from "@uniflowed/effect/stream";

const firstTen = streamTake(streamFromReadableStream(() => response.body), 10);
await runPromise(streamRunForEach(firstTen, handleChunk));

A Stream<A, E, R> is a pull: opening one hands back a step that produces the next batch, or nothing when there is no next batch, plus the way to close what the traversal opened. Batches rather than elements, because one Effect allocated per row is the whole cost of a stream on a large file. Effect calls the batch a Chunk; here it is a $ReadOnlyArray<A>, which is the honest equivalent in a language with no Chunk and no reason to grow a lookalike.

Every runner produces an ordinary Effect, so a stream ends where the rest of the package begins.

streamToReadableStream(stream, make) is the other boundary: the half @uniflowed/server needs, because streaming SSR and RSC produce a ReadableStream.

const body = streamToReadableStream(page, (source) => new ReadableStream(source));

Three decisions, none of them about the body of the function. Each host pull runs on its own fiber started with runFork, whose handle cancel reaches — a ReadableStream's pull is a callback the host calls, so there is no fiber to inherit. A typed failure crosses as its own value, because E is the type the consumer named; a defect and an interruption cross as an Error, because neither has a value anybody named. And the host's constructor is reached by being handed it, which is the same avoidance streamFromReadableStream makes by taking open, for the same reason: this module cannot name a global that four hosts declare in four places.

A consumer that cancels stops the pull it interrupted and closes the traversal, so a ReadableStream abandoned halfway releases what its source opened.

The guarantees that matter

A traversal is closed on success, on a failure from a pull, on a defect, and on the fiber being interrupted — the same guarantee acquireRelease gives, reached the same way, because the runners are written with ensuring. streamTake(3) of an infinite source releases what the source opened rather than leaking it, and a fiber draining a stream stops between pulls rather than in the middle of one.

A failure mid-stream ends the traversal with that failure in the error channel, and the elements before it are not reported. A defect stays a defect.

streamMapEffect fills its window across batch boundaries, so the concurrency a caller asks for is the concurrency they get whatever the source's batching was. There is no "unbounded", unlike all and forEach: a stream has no length, so unbounded would mean buffering the whole of it, which is the thing a stream exists to avoid.

streamBuffer and streamMerge are the only two combinators that own a fiber, because letting a source run ahead of its consumer and running two sources at once cannot be done by pulling. Both use a fiber per source and a bounded Queue, so the faster side stops at the bound rather than growing an array. The end of a source is a value in the queue and not a shutdown — a shutdown discards what is still buffered — and anything else does shut it down, which wakes a consumer waiting for a batch nobody will produce. What went wrong then lives on the fiber, and join puts it back on the stream's error channel with its cause intact.

Why it is a separate module and the runtime is not

index.js argues that the runtime cannot be split, because moving a combinator out means handing makeEffect to a sibling and giving away the opaque type's guarantee. A stream looked like it fell under that rule — it produces effects and runs them — and it does not: a Stream needs the Effect interface and never the Effect carrier. Every line of stream.js is written with the public exports, and index.js has no seam for it.

What streams deliberately do not have

Channel, Sink and GroupBy. Effect defines Stream and Sink in terms of Channel, a bidirectional primitive; a Stream defined directly cannot express a Sink as the dual of a source. That is a real cost and the right one to pay before porting a primitive nothing yet needs — the runners are the nearest honest shape.

A list of transforms, which is a different kind of absence: streamFlatMap, streamScan, streamMapAccum, streamTakeWhile, streamDrop, streamGrouped, streamAcquireRelease, streamRetry over a Schedule, streamTimeout, streamInterruptWhen, streamThrottle, streamDebounce and streamRechunk. None of them changes the type or needs a decision, which is why they can wait: adding them later costs nothing that adding them now would save.

What is not here

STM; Channel, Sink, GroupBy and Chunk; forkIn, and a schedule's Out as a type parameter, both for the reasons above; cron schedules; a fiber scheduler of its own — this runs on the host's microtask queue and its sleep is setTimeout; tracing, spans and metrics; a typed defect channel; and a heterogeneous all, which in Effect-TS keeps a tuple's element types and here takes and returns one array type.

Effect-TS is built on a higher-kinded encoding. Flow has no higher-kinded types — a type parameter cannot itself take type arguments — so there is no pipe with inference and no typeclass dispatch here. Composition is monomorphic functions over Effect, which is a smaller surface and one with nothing to resolve at run time.

Next: Internationalisation →