# 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**

```ts
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.catch(s1, () => s2)

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](/docs/v4/data-types/result/) data type before merging them, mapping `s1`'s elements into `Result.succeed` and `s2`'s elements into `Result.fail`:

```ts
import { Stream, Effect, Result } 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.map(s1, Result.succeed).pipe(
  Stream.catch(() => Stream.map(s2, Result.fail)),
)

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.

```ts
import { Stream, Effect } from "effect"

const s1 = 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 s2 = Stream.make("a", "b", "c")

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

const stream = Stream.catch(s1, (error): Stream.Stream<string | boolean> => {
  switch (error) {
    case "Uh Oh!":
      return s2
    case "Ouch":
      return s3
  }
})

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**

```ts
import { Stream, Effect } from "effect"

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

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

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

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`:

```ts
import { Stream, Effect, Filter } 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.catchFilter(
  s1,
  Filter.fromPredicate((error) => error === "Oh! Error!"),
  () => s2,
)

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:

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

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

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

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

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.

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

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

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
Stream application closed! We are doing some cleanup jobs.
Error: Oh! Boom!
*/
```

## 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**

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

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)) {
          return yield* Effect.fail("NaN")
        }
        return n
      }),
    ).pipe(Stream.retry(Schedule.exponential("1 second"))),
  ),
)

Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
Enter a number: a
Enter a number: b
Enter a number: c
Enter a number: 4
[ 1, 2, 3, 4 ]
*/

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)
        })
      }),
  )
```

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**

```ts
import { Stream, Option, Effect, Exit } from "effect"

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

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

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.

```ts
import { Stream, Effect } from "effect"

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

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.

```ts
import { Stream, Effect, Exit } from "effect"

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.fail("timeout"),
  }),
)

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.

```ts
import { Stream, Effect, Cause, Exit } from "effect"

const stream = Stream.fromEffect(Effect.never).pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.failCause(Cause.die("timeout")),
  }),
)

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.

```ts
import { Stream, Effect } from "effect"

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

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