Error Handling in Streams

Recovering from Failure

When working with streams that may encounter errors, it’s crucial to know how to handle these errors gracefully. The Stream.catch function is a powerful tool for recovering from failures and switching to an alternative stream in case of an error.

Example

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

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.fail("Oh! Error!")),
5
  Stream.concat(Stream.make(4, 5)),
6
)
7

8
const s2 = Stream.make("a", "b", "c")
9

10
const stream = Stream.catch(s1, () => s2)
11

12
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", "c"]

In this example, s1 encounters an error, but instead of terminating the stream, we gracefully switch to s2 using Stream.catch. This ensures that we can continue processing data even if one stream fails.

You can also distinguish elements from the two streams based on success or failure by tagging each side with the Result data type before merging them, mapping s1’s elements into Result.succeed and s2’s elements into Result.fail:

1
import { Stream, Effect, Result } from "effect"
2

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.fail("Oh! Error!")),
5
  Stream.concat(Stream.make(4, 5)),
6
)
7

8
const s2 = Stream.make("a", "b", "c")
9

10
const stream = Stream.map(s1, Result.succeed).pipe(
11
  Stream.catch(() => Stream.map(s2, Result.fail)),
12
)
13

14
await Effect.runPromise(Stream.runCollect(stream)) // => [Result.succeed(1), Result.succeed(2), Result.succeed(3), Result.fail("a"), Result.fail("b"), Result.fail("c")]

The Stream.catch function provides advanced error handling capabilities compared to Stream.catch. With Stream.catch, you can make decisions based on both the type and value of the encountered failure.

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

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.fail("Uh Oh!" as const)),
5
  Stream.concat(Stream.make(4, 5)),
6
  Stream.concat(Stream.fail("Ouch" as const)),
7
)
8

9
const s2 = Stream.make("a", "b", "c")
10

11
const s3 = Stream.make(true, false, false)
12

13
const stream = Stream.catch(s1, (error): Stream.Stream<string | boolean> => {
14
  switch (error) {
15
    case "Uh Oh!":
16
      return s2
17
    case "Ouch":
18
      return s3
19
  }
20
})
21

22
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", "c"]

In this example, we have a stream, s1, which may encounter two different types of errors. Instead of a straightforward switch to an alternative stream, as done with Stream.catch, we employ Stream.catch to precisely determine how to handle each type of error. This level of control over error recovery enables you to choose different streams or actions based on the specific error conditions.

Recovering from Defects

When working with streams, it’s essential to be prepared for various failure scenarios, including defects that might occur during stream processing. To address this, the Stream.catchCause function provides a robust solution. It enables you to gracefully handle and recover from any type of failure that may arise.

Example

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

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.die(new Error("Boom!"))),
5
  Stream.concat(Stream.make(4, 5)),
6
)
7

8
const s2 = Stream.make("a", "b", "c")
9

10
const stream = Stream.catchCause(s1, () => s2)
11

12
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", "c"]

In this example, s1 may encounter a defect, but instead of crashing the application, we use Stream.catchCause to gracefully switch to an alternative stream, s2. This ensures that your application remains robust and continues processing data even in the face of unexpected issues.

Recovery from Some Errors

In stream processing, there may be situations where you need to recover from specific types of failures. The Stream.catchFilter and Stream.catchCauseFilter functions come to the rescue, allowing you to handle and mitigate errors selectively.

If you want to recover from a particular error, you can use Stream.catchFilter:

1
import { Stream, Effect, Filter } from "effect"
2

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.fail("Oh! Error!")),
5
  Stream.concat(Stream.make(4, 5)),
6
)
7

8
const s2 = Stream.make("a", "b", "c")
9

10
const stream = Stream.catchFilter(
11
  s1,
12
  Filter.fromPredicate((error) => error === "Oh! Error!"),
13
  () => s2,
14
)
15

16
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", "c"]

To recover from a specific cause, you can use the Stream.catchCauseFilter function:

1
import { Stream, Effect, Cause } from "effect"
2

3
const s1 = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.die(new Error("Oh! Error!"))),
5
  Stream.concat(Stream.make(4, 5)),
6
)
7

8
const s2 = Stream.make("a", "b", "c")
9

10
const stream = Stream.catchCauseFilter(s1, Cause.findDie, () => s2)
11

12
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3, "a", "b", "c"]

Running Cleanup on Failure

Stream.onError runs an effect when the stream fails, then preserves the original failure. Use it for cleanup or diagnostics, not for recovery.

1
import { Stream, Console, Effect } from "effect"
2

3
const stream = Stream.make(1, 2, 3).pipe(
4
  Stream.concat(Stream.die(new Error("Oh! Boom!"))),
5
  Stream.concat(Stream.make(4, 5)),
6
  Stream.onError(() =>
7
    Console.log(
8
      "Stream application closed! We are doing some cleanup jobs.",
9
    ).pipe(Effect.orDie),
10
  ),
11
)
12

13
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
14
/*
15
Output:
16
Stream application closed! We are doing some cleanup jobs.
17
Error: Oh! Boom!
18
*/

Retry a Failing Stream

Sometimes, streams may encounter failures that are temporary or recoverable. In such cases, the Stream.retry operator comes in handy. It allows you to specify a retry schedule, and the stream will be retried according to that schedule.

Example

1
import { Stream, Effect, Schedule } from "effect"
2
import * as NodeReadLine from "node:readline"
3

4
const stream = Stream.make(1, 2, 3).pipe(
5
  Stream.concat(
6
    Stream.fromEffect(
7
      Effect.gen(function* () {
8
        const s = yield* readLine("Enter a number: ")
9
        const n = parseInt(s)
10
        if (Number.isNaN(n)) {
11
          return yield* Effect.fail("NaN")
12
        }
13
        return n
14
      }),
15
    ).pipe(Stream.retry(Schedule.exponential("1 second"))),
16
  ),
17
)
18

19
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
20
/*
21
Output:
22
Enter a number: a
23
Enter a number: b
24
Enter a number: c
25
Enter a number: 4
26
[ 1, 2, 3, 4 ]
27
*/
28

29
const readLine = (message: string): Effect.Effect<string> =>
30
  Effect.promise(
31
    () =>
32
      new Promise((resolve) => {
33
        const rl = NodeReadLine.createInterface({
34
          input: process.stdin,
35
          output: process.stdout,
36
        })
37
        rl.question(message, (answer) => {
38
          rl.close()
39
          resolve(answer)
40
        })
41
      }),
42
  )

In this example, the stream asks the user to input a number, but if an invalid value is entered (e.g., “a,” “b,” “c”), it fails with “NaN.” However, we use Stream.retry with an exponential backoff schedule, which means it will retry after a delay of increasing duration. This allows us to handle temporary errors and eventually collect valid input.

Refining Errors

When working with streams, there might be situations where you want to selectively keep certain errors and terminate the stream with the remaining errors. You can achieve this by pattern matching on the error inside Stream.catch, re-failing with the errors you want to keep and dying (Stream.die) on the rest.

Example

1
import { Stream, Option, Effect, Exit } from "effect"
2

3
const stream = Stream.fail(new Error())
4

5
const res = Stream.catch(stream, (error) => {
6
  const refined =
7
    error instanceof SyntaxError ? Option.some(error) : Option.none()
8
  return Option.isSome(refined) ? Stream.fail(refined.value) : Stream.die(error)
9
})
10

11
await Effect.runPromiseExit(Stream.runCollect(res)) // => Exit.die(new Error())

In this example, stream initially fails with a generic Error. However, res filters and keeps only errors of type SyntaxError, re-failing the stream with them. Any other error is turned into a defect via Stream.die, terminating the stream.

Timing Out

When working with streams, there are scenarios where you may want to handle timeouts, such as terminating a stream if it doesn’t produce a value within a certain duration. In this section, we’ll explore how to manage timeouts using various operators.

timeout

The Stream.timeout operator allows you to set a timeout on a stream. If the stream does not produce a value within the specified duration, it terminates.

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

3
const stream = Stream.fromEffect(Effect.never).pipe(Stream.timeout("2 seconds"))
4

5
await Effect.runPromise(Stream.runCollect(stream)) // => []

timeoutFail

The Stream.timeoutOrElse operator combines a timeout with a custom failure message. If the stream times out, it fails with the specified error message.

1
import { Stream, Effect, Exit } from "effect"
2

3
const stream = Stream.fromEffect(Effect.never).pipe(
4
  Stream.timeoutOrElse({
5
    duration: "2 seconds",
6
    orElse: () => Stream.fail("timeout"),
7
  }),
8
)
9

10
await Effect.runPromiseExit(Stream.runCollect(stream)) // => Exit.fail("timeout")

timeoutFailCause

Similar to Stream.timeoutOrElse, Stream.timeoutOrElse combines a timeout with a custom failure cause. If the stream times out, it fails with the specified cause.

1
import { Stream, Effect, Cause, Exit } from "effect"
2

3
const stream = Stream.fromEffect(Effect.never).pipe(
4
  Stream.timeoutOrElse({
5
    duration: "2 seconds",
6
    orElse: () => Stream.failCause(Cause.die("timeout")),
7
  }),
8
)
9

10
await Effect.runPromiseExit(Stream.runCollect(stream)) // => Exit.die("timeout")

timeoutTo

The Stream.timeoutOrElse operator allows you to switch to another stream if the first stream does not produce a value within the specified duration.

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

3
const stream = Stream.fromEffect(Effect.never).pipe(
4
  Stream.timeoutOrElse({
5
    duration: "2 seconds",
6
    orElse: () => Stream.make(1, 2, 3),
7
  }),
8
)
9

10
await Effect.runPromise(Stream.runCollect(stream)) // => [1, 2, 3]