Error Handling in Streams

This guide demonstrates techniques for managing errors and implementing recovery strategies in Effect’s Stream module.

onError

The Stream.onError function allows for cleanup or other tasks if a stream encounters an error.

Example (Executing Cleanup on Stream Error)

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

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Oh! Boom!")),
  Stream.concat(Stream.make(4, 5)),
  Stream.onError(() => Console.log("Some cleanup job..."))
)

Effect.runPromise(Stream.runCollect(stream))
/*
Output:
Some cleanup job...
RuntimeException: Oh! Boom!
*/

Recovering from Errors

orElse

Stream.orElse lets you recover from errors by switching to an alternative stream.

Example (Switching to an Alternative Stream on Error)

When s1 encounters an error, Stream.orElse switches to s2, allowing continued processing.

import { Stream, Effect } from "effect"

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

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

const stream = Stream.orElse(s1, () => s2)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b', 'c' ] }
*/

orElseEither

Stream.orElseEither enables error recovery by switching to an alternative stream if the original one fails, and wraps each element in an Either to indicate its source stream. This lets you distinguish between values from the primary and fallback streams.

Example (Differentiating Elements from Primary and Fallback Streams)

In this example, if s1 fails, Stream.orElseEither switches to s2, with each element wrapped in an Either to indicate whether it came from the original (Left) or fallback (Right) stream.

import { Stream, Effect } from "effect"

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

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

const stream = Stream.orElseEither(s1, () => s2)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: 'Chunk',
  values: [
    { _id: 'Either', _tag: 'Left', left: 1 },
    { _id: 'Either', _tag: 'Left', left: 2 },
    { _id: 'Either', _tag: 'Left', left: 3 },
    { _id: 'Either', _tag: 'Right', right: 'a' },
    { _id: 'Either', _tag: 'Right', right: 'b' },
    { _id: 'Either', _tag: 'Right', right: 'c' }
  ]
}
*/

catchAll

Stream.catchAll offers detailed error handling that allows for tailored recovery actions based on the specific error encountered. Unlike Stream.orElse, which provides a general fallback stream, Stream.catchAll lets you respond selectively to different error types and values.

Example (Handling Specific Errors with Different Streams)

import { Stream, Effect } from "effect"

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

const recovered = Stream.catchAll(
  stream,
  (error): Stream.Stream<string | boolean> => {
    switch (error) {
      case "Uh Oh!":
        return Stream.make("a", "b", "c")
      case "Ouch":
        return Stream.make(true, false, false)
    }
  }
)

Effect.runPromise(Stream.runCollect(recovered)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b', 'c' ] }
*/

Recovery from Some Errors

catchSome

To handle specific errors in a stream, you can use Stream.catchSome. This function allows you to recover from particular errors by switching to an alternative stream when a match is found.

Example (Handling a Specific Error with an Alternative Stream)

import { Stream, Effect, Option } from "effect"

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

const recovered = Stream.catchSome(stream, (error) => {
  if (error === "Oh! Error!") {
    return Option.some(Stream.make("a", "b", "c"))
  }
  return Option.none()
})

Effect.runPromise(Stream.runCollect(recovered)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b', 'c' ] }
*/

catchSomeCause

To handle specific causes of stream failures, you can use Stream.catchSomeCause. This function allows you to selectively recover from certain failure causes, such as defects or other specific issues, by switching to an alternative stream.

Example (Recovering from a Defect in a Stream)

import { Stream, Effect, Option, Cause } from "effect"

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Oh! Error!")),
  Stream.concat(Stream.make(4, 5))
)

const recovered = Stream.catchSomeCause(stream, (cause) => {
  if (Cause.isDie(cause)) {
    return Option.some(Stream.make("a", "b", "c"))
  }
  return Option.none()
})

Effect.runPromise(Stream.runCollect(recovered)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b', 'c' ] }
*/

Recovering from Defects

catchAllCause

The Stream.catchAllCause function provides a way to recover from any failure, whether it’s a standard error or an unexpected defect, by allowing you to switch to an alternative stream.

Example (Handling Defects in a Stream)

import { Stream, Effect } from "effect"

const s1 = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.dieMessage("Boom!")),
  Stream.concat(Stream.make(4, 5))
)

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

const stream = Stream.catchAllCause(s1, () => s2)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3, 'a', 'b', 'c' ] }
*/

Retry a Failing Stream

Sometimes, a stream may encounter temporary or recoverable errors. The Stream.retry operator provides a way to handle these cases by specifying a retry schedule, allowing the stream to attempt recovery according to the given schedule.

Example (Retrying a Stream with Exponential Backoff)

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

const readLine = (message: string): Effect.Effect<string> =>
  Effect.promise(
    () =>
      new Promise((resolve) => {
        const rl = NodeReadLine.createInterface({
          input: process.stdin,
          output: process.stdout
        })
        rl.question(message, (answer) => {
          rl.close()
          resolve(answer)
        })
      })
  )

const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(
    Stream.fromEffect(
      Effect.gen(function* () {
        const s = yield* readLine("Enter a number: ")
        const n = parseInt(s)
        if (Number.isNaN(n)) {
          // Fail if input is not a number
          return yield* Effect.fail("NaN")
        }
        return n
      })
    ).pipe(
      // Retry with exponential backoff
      Stream.retry(Schedule.exponential("1 second"))
    )
  )
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Example Output:
Enter a number: a
Enter a number: b
Enter a number: c
Enter a number: 4
{ _id: 'Chunk', values: [ 1, 2, 3, 4 ] }
*/

Terminate a Stream

In some cases, you may want to selectively handle specific types of errors in a stream while terminating on others. The Stream.refineOrDie function allows you to filter and retain only certain error types, terminating the stream if a non-matching error occurs.

Example (Filtering and Retaining Specific Error Types)

import { Stream, Option } from "effect"

//      ┌─── Stream<number, Error, never>
//      ▼
const stream = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail(new Error()))
)

//      ┌─── Stream<number, SyntaxError, never>
//      ▼
const refined = Stream.refineOrDie(stream, (error) => {
  if (error instanceof SyntaxError) {
    return Option.some(error)
  }
  return Option.none()
})

In this example, stream begins with a failure due to a generic Error. By applying Stream.refineOrDie, we filter out non-SyntaxError errors. If the error is a SyntaxError, it is retained in refined; otherwise, the stream terminates, ensuring only specific error types are processed further.

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 function allows you to impose a time limit on a stream. If the stream does not emit a value within the specified duration, it will terminate.

Example (Applying a Timeout to a Stream)

import { Stream, Effect } from "effect"

// Create a stream that will never emit a value
const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeout("2 seconds")
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [] }
*/

timeoutFail

The Stream.timeoutFail function allows you to set a timeout on a stream with a custom failure message. If the stream does not emit a value within the specified duration, it will fail with the provided error message.

Example (Timeout with Custom Failure Message)

import { Stream, Effect } from "effect"

// Create a stream that never emits a value
const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutFail(() => "timeout", "2 seconds")
)

Effect.runPromiseExit(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: 'Exit',
  _tag: 'Failure',
  cause: { _id: 'Cause', _tag: 'Fail', failure: 'timeout' }
}
*/

timeoutFailCause

The Stream.timeoutFailCause function allows you to set a timeout on a stream with a custom failure cause. If the stream does not produce a value within the specified duration, it fails with the provided cause.

Example (Timeout with Custom Failure Cause)

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

// Create a stream that never emits a value
const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutFailCause(() => Cause.die("timeout"), "2 seconds")
)

Effect.runPromiseExit(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: 'Exit',
  _tag: 'Failure',
  cause: { _id: 'Cause', _tag: 'Die', defect: 'timeout' }
}
*/

timeoutTo

The Stream.timeoutTo function lets you specify a fallback stream that will be used if the original stream fails to emit a value within the given duration. This can be useful for setting default behavior when a stream becomes unresponsive.

Example (Switching to a Fallback Stream on Timeout)

import { Stream, Effect } from "effect"

// Create a stream that never emits a value
const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutTo("2 seconds", Stream.make(1, 2, 3))
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2, 3 ] }
*/