35 - Concurrency

Previous · Next

Effect models concurrency with fibers, lightweight virtual threads, and ties their lifetime to the surrounding scope. Effect.all runs a fixed collection, Effect.forEach maps over a dynamic one, Effect.forkChild/Fiber.join give explicit control, and Effect.race picks the first success.

The runtime cleans up automatically when a scope ends, so unfinished fibers never leak.

What you’ll work through

index.test.ts covers the four high-level operators in order: Effect.all, Effect.forEach, Effect.forkChild/Fiber.join, and Effect.race. The interesting test is the concurrency option — a few beats let you flip it between "unbounded" and a fixed number and watch the wall-clock time change.

By the end you should be able to choose the right operator for “I have N things to do at once” without reaching for the docs.

Fibers and Structured Concurrency

A fiber is the unit of concurrent work. Forking is non-blocking; the parent continues immediately and the fiber runs in the background.

Joining is the synchronisation point. When the parent scope ends, by returning, by failing, by being interrupted, every child fiber is interrupted in turn.

1
parent scope
2
  │
3
  ├─ Effect.forkChild(taskA) ── fiber A ── running ─── join → result A
4
  │
5
  ├─ Effect.forkChild(taskB) ── fiber B ── running ─── join → result B
6
  │
7
  │  (if the parent ends before joining)
8
  │     ↓
9
  │  fiber A and fiber B are interrupted automatically
10
  └─────────────────────────────────────────────────────

This is structured concurrency: fiber lifetimes follow lexical scope, the way local variables do. No leaks, no orphaned work.

`Effect.all`

Runs a tuple or record of effects and returns the corresponding tuple or record of results. Sequential by default, pass a concurrency option for parallelism.

1
import { Effect } from "effect"
2

3
const fetchUser = Effect.succeed({ id: 1 })
4
const fetchPosts = Effect.succeed([{ title: "Hello" }])
5

6
//      ┌─── Effect<[{ id: number }, { title: string }[]], never, never>
7
//      ▼
8
const sequential = Effect.all([fetchUser, fetchPosts])
9

10
const parallel = Effect.all([fetchUser, fetchPosts], {
11
  concurrency: "unbounded",
12
})

Details

The concurrency option takes a number (the bound), "unbounded" (no limit), or "inherit" (use the surrounding scope’s limit). Results stay in the original input order regardless of which fiber finished first.

When to Use

When the set of effects is fixed and known statically. For dynamic collections see Effect.forEach.

Your turn add `{ concurrency: "unbounded" }` so the three sleeping tasks run in parallel.

`Effect.forEach`

Maps a function returning an effect over an iterable, collecting results. Same concurrency options as Effect.all.

1
import { Effect } from "effect"
2

3
declare const fetchItem: (id: number) => Effect.Effect<string>
4

5
//      ┌─── Effect<string[], never, never>
6
//      ▼
7
const program = Effect.forEach([1, 2, 3, 4, 5], fetchItem, {
8
  concurrency: 2,
9
})

Details

A bounded concurrency limit is the production-safe default for hitting external services. The limit caps in-flight work; everything else queues until a slot opens up.

When to Use

When the collection size is dynamic, when a concurrency cap should protect a downstream service, or when ordering of results by input is required.

Your turn call `Effect.forEach` with `{ concurrency: 3 }` so at most three tasks run at once.

`Effect.forkChild` and `Fiber.join`

Effect.forkChild returns a RuntimeFiber<A, E> without blocking. Fiber.join waits for the fiber and produces its result.

1
import { Effect, Fiber } from "effect"
2

3
const program = Effect.gen(function* () {
4
  const fiber = yield* Effect.forkChild(
5
    Effect.sleep("20 millis").pipe(Effect.as(42))
6
  )
7

8
  // Other work runs here in parallel with the fiber...
9
  const local = "done locally"
10

11
  const value = yield* Fiber.join(fiber)
12
  return { local, value }
13
})

Details

A forked fiber lives until joined, interrupted, or the surrounding scope ends. Fiber.interrupt cancels it explicitly.

For a fiber that must outlive its parent scope, use Effect.forkDetach, but prefer structured concurrency wherever possible.

When to Use

When fine-grained control is needed: starting work before it is consumed, doing other work between fork and join, or holding the fiber to interrupt it conditionally.

Your turn join the fiber with `Fiber.join(a)` so the parent waits for its result.

`Effect.race`

Runs two effects concurrently and returns the result of the first to succeed. The loser is interrupted.

If both fail, the race fails with both errors combined.

1
import { Effect } from "effect"
2

3
const slow = Effect.sleep("200 millis").pipe(Effect.as("slow"))
4
const fast = Effect.sleep("20 millis").pipe(Effect.as("fast"))
5

6
//      ┌─── Effect<"slow" | "fast", never, never>
7
//      ▼
8
const winner = slow.pipe(Effect.race(fast))
9
// "fast"

Details

A single failure does not end the race, only a success does. With one fast failure and one slow success, race returns the success.

If both effects fail, the resulting Cause is a parallel combination of both errors.

When to Use

Primary/fallback with the same return type, hedged requests across regions, or a deadline implemented as Effect.race(operation, Effect.sleep(deadline).pipe(Effect.as(fallback))).

@see Effect.raceAll for racing more than two effects.

Your turn race `slow` against `fast` so the faster one wins.

A Full Example

A bounded forEach that captures the maximum number of in-flight fibers. The cap protects the downstream service; results stay in input order.

1
import { Effect } from "effect"
2

3
let inFlight = 0
4
let maxInFlight = 0
5

6
const work = (n: number) =>
7
  Effect.gen(function* () {
8
    inFlight += 1
9
    maxInFlight = Math.max(maxInFlight, inFlight)
10
    yield* Effect.sleep("10 millis")
11
    inFlight -= 1
12
    return n * 2
13
  })
14

15
//      ┌─── Effect<number[], never, never>
16
//      ▼
17
const program = Effect.forEach([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], work, {
18
  concurrency: 3,
19
})
20

21
const results = await Effect.runPromise(program)
22
// results === [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]
23
// maxInFlight <= 3

Common Pitfalls

1, Effect.all is sequential by default. Omitting the concurrency option produces a sequential effect that looks parallel but is not. Add { concurrency: "unbounded" } for full parallelism or a number for a bounded cap.

2, Fiber.join without Effect.forkChild. Forking is what starts the work concurrently. Without it, calling Fiber.join on a non-existent fiber makes no sense, the program is sequential.

3, Assuming forked fibers outlive their parent. Fibers from Effect.forkChild are interrupted when the parent scope ends. Use Effect.forkDetach only when an outliving fiber is genuinely needed, and manage its lifetime explicitly.

4, Unbounded parallelism on external services. concurrency: "unbounded" over a thousand items fires a thousand simultaneous requests. Pick a number that matches the service’s capacity.

`"unbounded"`, a number, and the default. The wall-clock assertion is the point: each setting changes the runtime, not the result. - [ ] Then `Effect.forEach` with a dynamic collection. Confirm that `concurrency: 2` lets only two items run at once. - [ ] `Effect.forkChild` + `Fiber.join` for explicit control. Notice the parent fiber continues immediately after `fork`. - [ ] `Effect.race` — the winner returns; the loser is *interrupted*, not just abandoned. Step 37 picks up interruption directly.

The full test suite

The index.test.ts for this lesson, verbatim from steps/35-concurrency/.

1
import { describe, it, expect } from "vite-plus/test"
2
import { Effect, Fiber } from "effect"
3

4
// =============================================================================
5
// 20 - Concurrency
6
// =============================================================================
7
//
8
// Effect.all       — runs a fixed collection; sequential by default.
9
// Effect.forEach   — maps over a dynamic collection with concurrency control.
10
// Effect.forkChild      — starts a fiber without blocking; returns RuntimeFiber.
11
// Fiber.join       — waits for a fiber and returns its result.
12
// Effect.race      — first success wins; loser is interrupted.
13

14
describe("20 - Concurrency", () => {
15
  // @beat
16
  //   id: all
17
  //   shape: one-todo
18
  //   title: Run a fixed set of effects with Effect.all
19
  //   teaches: Effect.all runs a tuple sequentially by default; pass concurrency: "unbounded" or a number for parallelism.
20
  //   hint: Effect.all([task(1), task(2), task(3)], { concurrency: "unbounded" })
21
  describe("Effect.all", () => {
22
    it("runs effects sequentially by default", async () => {
23
      const order: number[] = []
24

25
      const task = (id: number) =>
26
        Effect.sync(() => {
27
          order.push(id)
28
          return id
29
        })
30

31
      //      ┌─── Effect<[number, number, number], never, never>
32
      //      ▼
33
      const program = Effect.all([task(1), task(2), task(3)])
34

35
      const results = await Effect.runPromise(program)
36
      expect(results).toEqual([1, 2, 3])
37
      expect(order).toEqual([1, 2, 3])
38
    })
39

40
    it("runs effects in parallel with concurrency: unbounded", async () => {
41
      const task = (id: number) =>
42
        Effect.gen(function* () {
43
          yield* Effect.sleep("10 millis")
44
          return id
45
        })
46

47
      const results = await Effect.runPromise(
48
        /* @stub */ Effect.all([task(1), task(2), task(3)], { concurrency: "unbounded" }) /* @end */
49
      )
50

51
      // Results preserve input order regardless of finish order.
52
      expect(results).toEqual([1, 2, 3])
53
    })
54

55
    it("respects a numeric concurrency limit", async () => {
56
      let inFlight = 0
57
      let maxInFlight = 0
58

59
      const task = (id: number) =>
60
        Effect.gen(function* () {
61
          inFlight += 1
62
          maxInFlight = Math.max(maxInFlight, inFlight)
63
          yield* Effect.sleep("20 millis")
64
          inFlight -= 1
65
          return id
66
        })
67

68
      const items = Array.from({ length: 6 }, (_, i) => i + 1)
69

70
      await Effect.runPromise(
71
        Effect.all(
72
          items.map((i) => task(i)),
73
          { concurrency: 2 }
74
        )
75
      )
76

77
      expect(maxInFlight).toBeLessThanOrEqual(2)
78
    })
79
  })
80
  // @end-beat
81

82
  // @beat
83
  //   id: for-each
84
  //   shape: one-todo
85
  //   title: Map a dynamic collection with Effect.forEach
86
  //   teaches: Effect.forEach applies an effectful function to every element; a numeric concurrency cap protects downstream services and results stay in input order.
87
  //   hint: Effect.forEach(items, work, { concurrency: 3 })
88
  describe("Effect.forEach", () => {
89
    it("maps a function over an iterable", async () => {
90
      const results = await Effect.runPromise(
91
        Effect.forEach([1, 2, 3, 4, 5], (n) => Effect.succeed(n * 2), {
92
          concurrency: "unbounded",
93
        })
94
      )
95

96
      expect(results).toEqual([2, 4, 6, 8, 10])
97
    })
98

99
    it("preserves input order even with variable durations", async () => {
100
      const results = await Effect.runPromise(
101
        Effect.forEach(
102
          [5, 4, 3, 2, 1],
103
          (n) =>
104
            Effect.gen(function* () {
105
              yield* Effect.sleep(`${n * 10} millis`)
106
              return n
107
            }),
108
          { concurrency: "unbounded" }
109
        )
110
      )
111

112
      expect(results).toEqual([5, 4, 3, 2, 1])
113
    })
114

115
    it("caps in-flight work with a bounded concurrency limit", async () => {
116
      let inFlight = 0
117
      let maxInFlight = 0
118

119
      const results = await Effect.runPromise(
120
        /* @stub */ Effect.forEach(
121
          Array.from({ length: 10 }, (_, i) => i + 1),
122
          (n) =>
123
            Effect.gen(function* () {
124
              inFlight += 1
125
              maxInFlight = Math.max(maxInFlight, inFlight)
126
              yield* Effect.sleep("10 millis")
127
              inFlight -= 1
128
              return n * 2
129
            }),
130
          { concurrency: 3 }
131
        ) /* @end */
132
      )
133

134
      expect(results).toEqual([2, 4, 6, 8, 10, 12, 14, 16, 18, 20])
135
      expect(maxInFlight).toBeLessThanOrEqual(3)
136
    })
137
  })
138
  // @end-beat
139

140
  // @beat
141
  //   id: fork-join
142
  //   shape: one-todo
143
  //   title: Fork a fiber and join it
144
  //   teaches: Effect.forkChild starts a fiber without blocking; Fiber.join waits for its result. Forked fibers are interrupted when the parent scope ends.
145
  //   hint: yield* Effect.forkChild(...); yield* Fiber.join(fiber)
146
  describe("Effect.forkChild and Fiber.join", () => {
147
    it("fork starts a fiber; join waits for its result", async () => {
148
      const result = await Effect.runPromise(
149
        Effect.gen(function* () {
150
          const a = yield* Effect.forkChild(Effect.succeed("A"))
151
          const b = yield* Effect.forkChild(Effect.succeed("B"))
152

153
          const aValue = yield* /* @stub */ Fiber.join(a) /* @end */
154
          const bValue = yield* Fiber.join(b)
155

156
          return `${aValue}-${bValue}`
157
        })
158
      )
159

160
      expect(result).toBe("A-B")
161
    })
162

163
    it("lets the parent do other work while the fiber runs", async () => {
164
      const result = await Effect.runPromise(
165
        Effect.gen(function* () {
166
          const fiber = yield* Effect.forkChild(
167
            Effect.gen(function* () {
168
              yield* Effect.sleep("20 millis")
169
              return 42
170
            })
171
          )
172

173
          // Other work happens here in parallel with the fiber.
174
          const local = "done locally"
175

176
          const value = yield* Fiber.join(fiber)
177
          return { local, value }
178
        })
179
      )
180

181
      expect(result).toEqual({ local: "done locally", value: 42 })
182
    })
183
  })
184
  // @end-beat
185

186
  // @beat
187
  //   id: race
188
  //   shape: one-todo
189
  //   title: First success wins with Effect.race
190
  //   teaches: Effect.race runs two effects concurrently; the first success returns and the loser is interrupted. A single failure does not end the race — only a success does.
191
  //   hint: slow.pipe(Effect.race(fast))
192
  describe("Effect.race", () => {
193
    it("returns the result of the first effect to succeed", async () => {
194
      const slow = Effect.sleep("200 millis").pipe(Effect.as("slow"))
195
      const fast = Effect.sleep("20 millis").pipe(Effect.as("fast"))
196

197
      const result = await Effect.runPromise(/* @stub */ slow.pipe(Effect.race(fast)) /* @end */)
198
      expect(result).toBe("fast")
199
    })
200

201
    it("a single failure does not end the race — only a success does", async () => {
202
      const failing = Effect.fail("primary error")
203
      const succeeding = Effect.succeed("fallback result")
204

205
      const result = await Effect.runPromise(
206
        failing.pipe(Effect.race(succeeding))
207
      )
208

209
      expect(result).toBe("fallback result")
210
    })
211
  })
212
  // @end-beat
213

214
  describe("structured concurrency", () => {
215
    it("interrupts child fibers when the parent scope ends", async () => {
216
      let childCompleted = false
217

218
      const parent = Effect.gen(function* () {
219
        yield* Effect.forkChild(
220
          Effect.gen(function* () {
221
            yield* Effect.sleep("10 seconds")
222
            childCompleted = true
223
          })
224
        )
225
        return "parent done"
226
      })
227

228
      const result = await Effect.runPromise(parent)
229
      expect(result).toBe("parent done")
230
      expect(childCompleted).toBe(false)
231
    })
232
  })
233
})

07·01Concurrency

35 - Concurrency

What you’ll work through

Fibers and Structured Concurrency

`Effect.all`

`Effect.forEach`

`Effect.forkChild` and `Fiber.join`

`Effect.race`

A Full Example

Common Pitfalls

See Also

The full test suite