effect-concurrency

Files (4)

• SKILL.md
• effect-concurrency-coordination.md · references/
• effect-concurrency-fibers.md · references/
• effect-concurrency-schedule.md · references/

SKILL.md (first lines)

# Effect concurrency

Checked against `effect@4.0.0`. When the project has a newer version, check each API you use in `node_modules/effect/src/<Module>.ts` and read `node_modules/effect/AGENTS.md` first. The source wins over this skill.

Every example imports each module from its subpath, `import * as Effect from "effect/Effect"`. The `effect` index import also works, but it loads every module: measured on Node, 216 modules in 220 to 270 ms cold, against 64 modules in 60 to 70 ms for `effect/Effect`.

Every effect runs on a fiber, a light thread that the Effect runtime schedules. Concurrency in Effect is structured: a fiber you start belongs to a parent fiber or a `Scope`, and interruption runs every finalizer before the interrupter resumes.

## Choose the tool

| Need | Use |
| --- | --- |
| Run a known set or a list of effects, with a limit | `Effect.all`, `Effect.forEach` with `concurrency` |
| Run work in the background and come back for it | `Effect.forkChild`, then `Fiber.join` / `Fiber.await` / `Fiber.interrupt` |
| Keep a changing set of background fibers | `FiberSet`, `FiberMap` (by key), `FiberHandle` (one at a time) |
| Take the first result | `Effect.race`, `Effect.raceAll`, `Effect.raceFirst`, `Effect.raceAllFirst` |
| Bound the time of an effect | `Effect.timeout`, `Effect.timeoutOption`, `Effect.timeoutOrElse` |
| Wrap a Promise or callback API that can be cancelled | `Effect.tryPromise` with its `signal`, `Effect.callback` |
| Signal once, gate, limit, hand off, broadcast, share state | `Deferred`, `Latch`, `Semaphore`, `Queue`, `PubSub`, `Ref` |
| Retry, repeat, poll, run at calendar times | `Effect.retry`, `Effect.repeat`, `Effect.schedule` with a `Schedule` |

Reach for `Effect.all`, `Effect.forEach` and the race functions first. Fork by hand only when the work must outlive the expression that starts it.

## Run effects in parallel

`Effect.all` takes a tuple, an iterable or a record of effects and keeps that shape. `Effect.forEach` maps a function over an iterable. Both take the same options:

| Option | Values | Effect |
| --- | --- | --- |
| `concurrency` | omitted, a number, `"unbounded"` | Omitted runs one at a time, in order. A number caps the running fibers. Results keep input order in every mode. |
| `discard` | `true` | Drops the results and succeeds with `void`. |
| `mode` (`Effect.all` only) | `"result"` | Turns each typed failure into a `Result` and runs the rest. A defect or an interruption still ends the combined effect, and effects not yet run do not run. |

When one effect fails (default mode):

- Sequential: the rest never start.
- Concurrent: the first failure makes the combined effect interrupt every sibling that is still running and wait for their finalizers. Effects not yet started never start. The `Cause` starts with that first failure, and can also hold the failure of an uninterruptible sibling and a defect from a sibling's finalizer. A sibling that is interrupted before it fails adds nothing.
- To keep going after a typed failure, use `mode: "result"`, `Effect.partition` (splits successes and failures) or `Effect.validate` (collects every failure). To keep going after defects too, map each effect through `Effect.exit` and inspect each `Exit`. An interruption of the whole call still stops everything.

```ts
import * as Effect from "effect/Effect"

declare const loadInvoice: (id: string) => Effect.Effect<{ readonly total: number }, Error>

export const totals = Effect.fn("totals")(function*(ids: ReadonlyArray<string>) {
  // At most 4 requests in flight. The first failure interrupts the others.
  const invoices = yield* Effect.forEach(ids, loadInvoice, { concurrency: 4 })
  return invoices.reduce((sum, invoice) => sum + invoice.total, 0)
})
```

`Effect.zip(a, b, { concurrent: true })` runs a pair concurrently. A bare `Effect.zip` is sequential.

## Fork fibers

| Fork | The fiber ends when |
| --- | --- |
| `Effect.forkChild(effect)` | It completes, or the fiber that forked it ends. |
| `Effect.forkScoped(effect)` | It completes, or the `Scope` in context closes. Adds `Scope` to the requirements. |
| `Effect.forkIn(effect, scope)` | It completes, or the given `scope` closes. |
| `Effect.forkDetach(effect)` | It completes, or something interrupts it. Nothing supervises it. |

- `forkChild` ties the child to the parent fiber, not to the `Effect.gen` block or function that forked it. A helper that forks and returns leaves the child running until its caller's whole fiber ends. When a block must own its children, run it as its own fiber, wrap it in `Effect.scoped` with `forkScoped`, or interrupt the child before it returns.
- A forked fiber starts after the current fiber yields. Pass `{ startImmediately: true }` when it must start before the next line, for example to subscribe before you publish.
- `Fiber.join(fiber)` resumes with the fiber's value or re-raises its failure. `Fiber.await(fiber)` returns its `Exit` and never fails. `Fiber.interrupt(fiber)` returns `void` after the fiber's finalizers finish.
- `Effect.awaitAllChildren(effect)` delays completion until every child that `effect` forked has finished.
- Forked fibers in a long-running service belong in a `FiberSet`, `FiberMap` or `FiberHandle` created in the service's scope, so that closing the scope interrupts them.

Lifetimes, the `Fiber` API and the fiber collections are in [effect-concurrency-fibers.md](references/effect-concurrency-fibers.md).

## Interruption

- Interruption comes from outside the fiber: `Fiber.interrupt`, a lost race, a timeout, a failed sibling, a closed scope, or the `signal` passed to `Effect.runPromise`. `Effect.interrupt` interrupts the current fiber.
- The interrupter waits for the target's finalizers (`Effect.onInterrupt`, `Effect.ensuring`, `Effect.acquireRelease`). A slow finalizer delays the caller.
- `Effect.timeout(duration)` interrupts the source and fails with `Cause.TimeoutError`. If the source is uninterruptible, the timeout returns only when the source ends, and still fails with `TimeoutError`: the result is lost. Keep uninterruptible regions short.
- Mark a critical section with `Effect.uninterruptibleMask((restore) => ...)` and wrap only the waits that may stop in `restore`. `Effect.uninterruptible` protects the whole effect.
- An uninterruptible region delays an interrupt. It does not cancel it. The region runs to its end, then the fiber ends interrupted and the region's value is dropped. So when the signal of `Effect.runPromiseExit(program, { signal })` fires after the first suspension, the `Exit` is an interrupt even when `Effect.uninterruptible` wraps the rest of the program. The region keeps a write whole, but it cannot keep an answer. A program whose answer must survive a client close runs with no interrupting signal and reads the close as a value. Example: [effect-concurrency-fibers.md](references/effect-concurrency-fibers.md#an-uninterruptible-region-does-not-keep-the-answer).
- Interruption appears in an `Exit` as an `Interrupt` reason in the `Cause`, not as a typed error. See [effect-errors](skill:effect-errors).

## Wrap a cancellable API

When a fiber that waits on `Effect.tryPromise` or `Effect.promise` is interrupted, it ends at once: `Effect.runPromiseExit` resolves without waiting for the Promise to settle. The underlying call stops only if it observes the `AbortSignal` that Effect passes to the `try` function.

- Declare the `signal` parameter in the function you pass, and pass it to the call (`fetch`, an SDK option, a stream). Effect creates the `AbortController` only when that function declares a parameter: a `(...args) => call(...args)` wrapper receives `undefined`.
- An API with no cancellation runs on after the interruption. Its result is dropped.

### A call that takes no signal

Some SDKs take no abort signal, for example a database admin SDK. An interrupt cannot stop such a call. So choose for each call what the interrupt does, and give each choice its own helper. Then a reader sees the choice where the call is made:

- **A read stays interruptible.** A closed request stops waiting at once. The call runs to its end, and nothing reads its result.
- **A write or a transaction is uninterruptible.** A closed request waits for the commit, then ends. Otherwise the caller cannot know whether the write happened.
- A read inside an uninterruptible region waits like a write.

```ts
import * as Effect from "effect/Effect"
import * as Schema from "effect/Schema"

class StoreError extends Schema.TaggedError<StoreError>()("StoreError", { cause: Schema.Defect() }) {}

// The SDK client. No call takes an abort signal.
interface Db {
  get(path: string): Promise<unknown>
  set(path: string, value: unknown): Promise<void>
}
declare const db: Db

const call = <A>(f: (db: Db) => Promise<A>) =>
  Effect.tryPromise({ try: () => f(db), catch: (cause) => new StoreError({ cause }) })

// A read: an interrupt ends the wait at once.
export const storeRead = <A>(f: (db: Db) => Promise<A>) => call(f)

// A write or a transaction: an interrupt waits for the commit.
export const storeWrite = <A>(f: (db: Db) => Promise<A>) => Effect.uninterruptible(call(f))

export const rename = Effect.fn("rename")(function*(id: string, title: string) {
  const before = yield* storeRead((db) => db.get(`items/${id}`))
  yield* storeWrite((db) => db.set(`items/${id}/title`, title))

From andabove/skills, MIT.

More from andabove/skills