Error Handling

A stream’s failure type E is part of its signature, just like an Effect’s. When a stream fails, every downstream operator short-circuits and the failure surfaces wherever you run it. Effect gives you the same recovery vocabulary you use elsewhere — switch to a fallback stream, match on a specific tagged error, retry on a schedule, or run cleanup on failure — all expressed inside the stream so the result type reflects what can still go wrong.

import { Effect, Stream } from "effect"

// A stream that emits a couple of values and then fails.
const flaky = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom" as const)),
  Stream.concat(Stream.make(4, 5))
)

const fallback = Stream.make(10, 20, 30)

// `Stream.catch` is the catch-all: on any failure it switches to the stream you
// return. Values emitted *before* the failure are kept; the rest come from the
// fallback. The recovered stream's error channel is now `never`.
export const recovered = flaky.pipe(
  Stream.catch(() => fallback),
  Stream.runCollect
)
// => [1, 2, 3, 10, 20, 30]

Note

In v4 the catch-all on streams is Stream.catch (not catchAll), mirroring Effect.catch. For defects and interruptions, use Stream.catchCause, which receives the full Cause rather than just the typed error.

Recovering from specific errors

When a stream can fail with several tagged errors, recover from each one individually with Stream.catchTag (or catchTags for several at once). This keeps the other error types in the signature so the compiler still forces you to handle them.

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

class NotFound extends Schema.TaggedErrorClass<NotFound>()("NotFound", {
  id: Schema.String
}) {}

class RateLimited extends Schema.TaggedErrorClass<RateLimited>()("RateLimited", {
  retryAfterMillis: Schema.Number
}) {}

declare const records: Stream.Stream<string, NotFound | RateLimited>

// Handle `NotFound` by substituting an empty stream; `RateLimited` still
// propagates, so the result type is `Stream<string, RateLimited>`.
export const handled = records.pipe(
  Stream.catchTag("NotFound", () => Stream.empty)
)

// Handle several tags at once with a record of handlers.
export const handledBoth = records.pipe(
  Stream.catchTags({
    NotFound: () => Stream.empty,
    RateLimited: (e) => Stream.fromEffect(
      Effect.as(Effect.sleep(e.retryAfterMillis), "retried-placeholder")
    )
  })
)

For richer matching there is also Stream.catchIf (recover when a predicate or refinement matches) and Stream.catchCause (inspect the full Cause, including defects). See Error Management for how tagged errors, Cause, and these combinators relate across all of Effect.

Retrying a failing stream

Transient failures — a flaky socket, a rate-limited API — are best handled by retrying rather than giving up. Stream.retry takes a Schedule and restarts the stream from the beginning each time it fails, according to the policy.

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

const readSensor = Effect.fn("readSensor")(function*() {
  // Imagine an effect that occasionally fails to read a hardware sensor.
  return yield* Effect.fail("sensor unavailable" as const)
})

// Retry with exponential backoff, giving up after 5 attempts. `Schedule.both`
// combines the two policies with AND semantics: it keeps recurring only while
// *both* the exponential backoff and the `recurs(5)` cap still want to recur.
export const sensorStream = Stream.fromEffect(readSensor()).pipe(
  Stream.retry(Schedule.exponential("100 millis").pipe(
    Schedule.both(Schedule.recurs(5))
  ))
)

Caution

Stream.retry re-runs the stream from its source on each attempt, so it re-executes all of the stream’s acquire operations. For an unbounded source this restarts iteration; for a one-shot effect it simply re-runs the effect. Bound the policy (e.g. with Schedule.recurs) so a permanently-broken source does not retry forever.

Cleanup and timeouts

Stream.onError registers a finalizer that runs if (and only if) the stream fails, receiving the Cause. It does not change the failure — use it for cleanup, logging, or metrics, not recovery.

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("oops" as const)),
  // Runs on failure; the original error still propagates afterwards.
  Stream.onError((cause) =>
    Effect.logError(`stream failed: ${Cause.pretty(cause)}`)
  )
)

Two operators guard against streams that stall rather than fail. Stream.timeout ends the stream if no value is produced within a duration; Stream.timeoutOrElse switches to a fallback stream instead.

import { Effect, Stream } from "effect"

const slow: Stream.Stream<number> = Stream.fromEffect(
  Effect.as(Effect.never, 0)
)

// End the stream (emit nothing more) if no element arrives within 2 seconds.
export const bounded = slow.pipe(Stream.timeout("2 seconds"))

// Or switch to a fallback stream after the timeout.
export const orFallback = slow.pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.make(1, 2, 3)
  })
)

Together these let a long-running pipeline degrade gracefully: retry the transient failures, time out the stalls, fall back where recovery makes sense, and clean up on the paths that cannot recover.

Recovery reference

Every recovery and lifecycle combinator on Stream, grouped by purpose. Each returns a new stream; they compose with .pipe(...).

Catching

Stream.catch

Switches to the stream returned by f on any typed failure. The recovered error channel is whatever the fallback stream contributes.

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("Oops!")),
  Stream.catch(() => Stream.make(999)),
  Stream.runCollect
)
// => [1, 2, 999]

Stream.catchCause

Recovers from the full Cause, so it sees defects and interruptions as well as typed failures. Use this when a fallback should fire on any abnormal exit.

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("Oops!")),
  Stream.catchCause(() => Stream.make(999)),
  Stream.runCollect
)
// => [1, 2, 999]

Stream.catchIf

Recovers only when a predicate (or type refinement) matches the error. Non-matching failures propagate, so the error type is preserved unless the refinement narrows it. An optional third handler covers the non-matching case.

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(42)),
  Stream.catchIf(
    (error): error is 42 => error === 42,
    () => Stream.make(999)
  ),
  Stream.runCollect
)
// => [1, 2, 999]

Stream.catchFilter

Recovers using a reusable Filter, which can also narrow or transform the error before choosing the recovery stream. Successful filter results go to f; failed results re-fail (or go to the optional orElse).

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

// Filter keeps only negative numbers, passing them to the handler.
const negatives = Filter.fromPredicate((n: number) => n < 0)

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail(-5)),
  Stream.catchFilter(negatives, (n) => Stream.make(`recovered ${n}`)),
  Stream.runCollect
)
// => [1, 2, "recovered -5"]

Stream.catchTag

Recovers from failures whose _tag matches the given value, switching to the stream returned by the handler. Other tags stay in the error channel. Pass an array of tags to match several at once.

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

class HttpError extends Data.TaggedError("HttpError")<{ message: string }> {}

const program = Stream.fail(new HttpError({ message: "timeout" })).pipe(
  Stream.catchTag("HttpError", (e) => Stream.make(`Recovered: ${e.message}`)),
  Stream.runCollect
)
// => ["Recovered: timeout"]

Stream.catchTags

Recovers from multiple tagged failures with a record of _tag-keyed handlers. Tags you omit stay in the error channel.

import { Effect, Stream } from "effect"

class NotFound {
  readonly _tag = "NotFound"
  constructor(readonly resource: string) {}
}
class Unauthorized {
  readonly _tag = "Unauthorized"
  constructor(readonly user: string) {}
}

const program = Stream.fail(new NotFound("profile")).pipe(
  Stream.catchTags({
    NotFound: () => Stream.succeed("fallback"),
    Unauthorized: () => Stream.succeed("login")
  }),
  Stream.runCollect
)
// => ["fallback"]

Stream.catchReason

Catches a specific nested reason inside a tagged error without removing the parent error from the channel. The handler receives the unwrapped reason. Pass the parent error tag, then the reason tag, then the handler.

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

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}
class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError
}> {}

const program = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
).pipe(
  Stream.catchReason("AiError", "RateLimitError", (reason) =>
    Stream.succeed(`retry: ${reason.retryAfter}`)
  ),
  Stream.runCollect
)
// => ["retry: 60"]

Stream.catchReasons

Catches several reasons within one tagged error using an object of handlers keyed by reason tag.

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

class RateLimitError extends Data.TaggedError("RateLimitError")<{
  retryAfter: number
}> {}
class QuotaExceededError extends Data.TaggedError("QuotaExceededError")<{
  limit: number
}> {}
class AiError extends Data.TaggedError("AiError")<{
  reason: RateLimitError | QuotaExceededError
}> {}

const program = Stream.fail(
  new AiError({ reason: new RateLimitError({ retryAfter: 60 }) })
).pipe(
  Stream.catchReasons("AiError", {
    RateLimitError: (reason) => Stream.succeed(`retry: ${reason.retryAfter}`),
    QuotaExceededError: (reason) => Stream.succeed(`quota: ${reason.limit}`)
  }),
  Stream.runCollect
)
// => ["retry: 60"]

Stream.catchCauseIf

Recovers from a failure when a predicate over the full Cause matches. Non-matching causes are re-emitted as failures, so the error type is preserved.

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

const program = Stream.fail("NetworkError").pipe(
  Stream.catchCauseIf(
    (cause) => Cause.hasFails(cause),
    (cause) => Stream.make(`Recovered: ${Cause.squash(cause)}`)
  ),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]

Stream.catchCauseFilter

Recovers from causes selected by a Filter applied to the full Cause. The handler receives both the filtered value and the original cause; a failed filter result re-fails with the residual cause.

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

// Keep only causes that contain a typed failure.
const onlyFails = Filter.fromPredicate(Cause.hasFails)

const program = Stream.fail("NetworkError").pipe(
  Stream.catchCauseFilter(onlyFails, (cause) =>
    Stream.make(`Recovered: ${Cause.squash(cause)}`)
  ),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]

Mapping and observing errors

Stream.mapError

Transforms the typed error with f without recovering. Useful for wrapping low-level errors in a domain error before they reach a boundary.

import { Effect, Stream } from "effect"

const program = Stream.fail("bad").pipe(
  Stream.mapError((error) => `mapped: ${error}`),
  Stream.catch((error) => Stream.make(`recovered from ${error}`)),
  Stream.runCollect
)
// => ["recovered from mapped: bad"]

Stream.tapError

Runs an effect when the stream fails, peeking at the typed error without changing the values or error — unless the tap effect itself fails.

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

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapError((error) => Console.log(`tapError: ${error}`)),
  Stream.catch(() => Stream.make(999)),
  Stream.runCollect
)
// logs: tapError: boom
// => [1, 2, 999]

Stream.tapCause

Like tapError, but the tap effect receives the full Cause, so it also fires on defects and interruptions.

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

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.tapCause((cause) => Effect.logError(Cause.pretty(cause))),
  Stream.catch(() => Stream.succeed(0)),
  Stream.runCollect
)
// => [1, 2, 0]

Stream.result

Surfaces each failure as a Result value inside the stream, turning the error channel into never. Successful elements become Result.succeed; the failure becomes a trailing Result.fail.

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

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.result,
  Stream.map(Result.match({
    onFailure: (error) => `failure: ${error}`,
    onSuccess: (value) => `success: ${value}`
  })),
  Stream.runCollect
)
// => ["success: 1", "success: 2", "failure: boom"]

Fallbacks

Stream.orElseIfEmpty

Switches to a fallback stream only when the source completes without emitting any value. Failures still propagate.

import { Effect, Stream } from "effect"

const program = Stream.empty.pipe(
  Stream.orElseIfEmpty(() => Stream.make(1, 2)),
  Stream.runCollect
)
// => [1, 2]

Stream.orElseSucceed

Recovers from a failure by emitting a single fallback value computed from the error. The error channel becomes never.

import { Effect, Stream } from "effect"

const program = Stream.fail("NetworkError").pipe(
  Stream.orElseSucceed((error) => `Recovered: ${error}`),
  Stream.runCollect
)
// => ["Recovered: NetworkError"]

Stream.orDie

Turns typed failures into defects, making the stream infallible (E becomes never). Use when a failure here is a bug, not a recoverable condition.

import { Effect, Stream } from "effect"

const infallible = Stream.make(1, 2, 3).pipe(Stream.orDie)
const program = Stream.runCollect(infallible)
// => [1, 2, 3]   (a failure would now surface as a defect)

Stream.ignore

Ignores typed failures and ends the stream gracefully, keeping whatever was emitted before. The log option (default on) controls whether the failure is logged first.

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.ignore({ log: false }),
  Stream.runCollect
)
// => [1, 2, 3]

Stream.ignoreCause

Like ignore, but suppresses the entire Cause — including defects — rather than only typed failures, then ends the stream.

import { Effect, Stream } from "effect"

const program = Stream.make(1, 2).pipe(
  Stream.concat(Stream.die("boom")),
  Stream.ignoreCause({ log: false }),
  Stream.runCollect
)
// => [1, 2]

Retry and resilience

Stream.retry

Re-runs the whole stream on failure according to a Schedule. The schedule resets once the first element passes through again.

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

const program = Stream.make(1).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.retry(Schedule.recurs(1)),
  Stream.take(2),
  Stream.runCollect
)
// => [1, 1]

Stream.withExecutionPlan

Applies an ExecutionPlan: on failure it retries with each step’s provided resources until one succeeds or the plan is exhausted. Set preventFallbackOnPartialStream: true to fail rather than mix partial output with a later step’s fallback.

import { Context, Effect, ExecutionPlan, Layer, Stream } from "effect"

class Service extends Context.Service<Service>()("Service", {
  make: Effect.succeed({
    stream: Stream.fail("A") as Stream.Stream<number, string>
  })
}) {
  static Bad = Layer.succeed(Service, Service.of({ stream: Stream.fail("A") }))
  static Good = Layer.succeed(Service, Service.of({ stream: Stream.make(1, 2, 3) }))
}

const plan = ExecutionPlan.make(
  { provide: Service.Bad },
  { provide: Service.Good }
)

const stream = Stream.unwrap(Effect.map(Service, (_) => _.stream))

const program = stream.pipe(Stream.withExecutionPlan(plan), Stream.runCollect)
// => [1, 2, 3]

Timeouts

Stream.timeout

Ends the stream (emitting nothing more) if no element is produced within the duration. The timeout is checked on each pull.

import { Effect, Stream } from "effect"

const slow = Stream.fromEffect(Effect.as(Effect.never, 0))

const program = slow.pipe(Stream.timeout("2 seconds"))
// => []  (after 2s with no element)

Stream.timeoutOrElse

Switches to a fallback stream if no value arrives within the duration. The fallback is not itself timed after the switch.

import { Effect, Stream } from "effect"

const slow = Stream.fromEffect(Effect.as(Effect.never, 0))

const program = slow.pipe(
  Stream.timeoutOrElse({
    duration: "2 seconds",
    orElse: () => Stream.make(1, 2, 3)
  }),
  Stream.runCollect
)
// => [1, 2, 3]

Lifecycle and finalizers

Stream.onStart

Runs an effect once, before the stream starts pulling. Its error and requirements are added to the stream.

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

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.onStart(Console.log("Stream started")),
  Stream.runCollect
)
// logs: Stream started
// => [1, 2, 3]

Stream.onFirst

Runs an effect with the first element emitted by the stream (and only the first).

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

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.onFirst((value) => Console.log(`first=${value}`)),
  Stream.runDrain
)
// logs: first=1

Stream.onEnd

Runs an effect when the stream ends successfully (not on failure).

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.onEnd(Console.log("Stream ended")),
  Stream.runCollect
)
// logs: Stream ended
// => [1, 2, 3]

Stream.onError

Runs a cleanup effect only when the stream fails, receiving the Cause. It does not change the failure.

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.concat(Stream.fail("boom")),
  Stream.onError((cause) => Effect.logError(`Stream failed: ${Cause.squash(cause)}`))
)
// logs on failure: Stream failed: boom (then the failure propagates)

Caution

Unlike Effect.onError, the Stream.onError cleanup effect carries no guarantee that it will not itself be interrupted. For guaranteed-to-run cleanup, prefer Stream.ensuring.

Stream.onExit

Runs a finalizer when the stream exits for any reason, receiving the Exit so you can branch on success vs failure.

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

const program = Stream.make(1, 2, 3).pipe(
  Stream.onExit((exit) =>
    Exit.isSuccess(exit)
      ? Console.log("completed successfully")
      : Console.log("failed")
  ),
  Stream.runCollect
)
// logs: completed successfully
// => [1, 2, 3]

Stream.ensuring

Runs a finalizer after the stream’s own finalizers, regardless of how it exits. This is the resilient cleanup hook.

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

const program = Stream.fromArray([1, 2]).pipe(
  Stream.ensuring(Effect.orDie(Console.log("cleanup"))),
  Stream.runCollect
)
// logs: cleanup
// => [1, 2]

Providing context

These mirror the Effect.provide* family and remove the corresponding service requirements from a stream. See Services for the full story.

Stream.provide

Provides a Layer (or a built Context) to the stream, discharging its requirements. Layers are shared between provide calls unless you pass { local: true }.

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

class Env extends Context.Service<Env, { readonly name: string }>()("Env") {}

const layer = Layer.succeed(Env)({ name: "Ada" })

const program = Stream.fromEffect(
  Effect.map(Effect.service(Env), (env) => `Hello, ${env.name}`)
).pipe(
  Stream.provide(layer),
  Stream.runCollect,
  Effect.tap((values) => Console.log(values))
)
// => ["Hello, Ada"]

Stream.provideContext

Provides a whole prebuilt Context at once, eliminating several requirements.

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

class Config extends Context.Service<Config, { readonly prefix: string }>()("Config") {}
class Greeter extends Context.Service<Greeter, { greet: (name: string) => string }>()("Greeter") {}

const context = Context.make(Config, { prefix: "Hello" }).pipe(
  Context.add(Greeter, { greet: (name: string) => `${name}!` })
)

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const config = yield* Effect.service(Config)
    const greeter = yield* Effect.service(Greeter)
    return greeter.greet(config.prefix)
  })
)

const program = Stream.runCollect(Stream.provideContext(stream, context))
// => ["Hello!"]

Stream.provideService

Provides a single service value directly by its key.

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

class Greeter extends Context.Service<Greeter, {
  greet: (name: string) => string
}>()("Greeter") {}

const stream = Stream.fromEffect(
  Effect.map(Effect.service(Greeter), (g) => g.greet("Ada"))
)

const program = stream.pipe(
  Stream.provideService(Greeter, { greet: (name) => `Hello, ${name}` }),
  Stream.runCollect
)
// => ["Hello, Ada"]

Stream.provideServiceEffect

Provides a service whose value is produced by an effect, adding that effect’s error and requirements to the stream.

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

class ApiConfig extends Context.Service<ApiConfig, { readonly baseUrl: string }>()("ApiConfig") {}

const stream = Stream.fromEffect(
  Effect.map(Effect.service(ApiConfig), (c) => c.baseUrl)
)

const program = stream.pipe(
  Stream.provideServiceEffect(
    ApiConfig,
    Effect.succeed({ baseUrl: "https://example.com" }).pipe(
      Effect.tap(() => Console.log("Loading config..."))
    )
  ),
  Stream.runCollect
)
// logs: Loading config...
// => ["https://example.com"]

Stream.updateContext

Transforms the stream’s required context by mapping the current Context to a new one — handy for adding or replacing services without fully providing them.

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

class Logger extends Context.Service<Logger, { prefix: string }>()("Logger") {}
class Config extends Context.Service<Config, { name: string }>()("Config") {}

const stream = Stream.fromEffect(
  Effect.gen(function*() {
    const logger = yield* Effect.service(Logger)
    const config = yield* Effect.service(Config)
    return `${logger.prefix}${config.name}`
  })
)

const updated = stream.pipe(
  Stream.updateContext((context: Context.Context<Logger>) =>
    Context.add(context, Config, { name: "World" })
  )
)
// requires only Logger now; Config is supplied by updateContext

Stream.updateService

Updates a single service in the environment by applying a function to its current value (the requirement remains).

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

class Counter extends Context.Service<Counter, { count: number }>()("Counter") {}

const stream = Stream.fromEffect(Effect.service(Counter)).pipe(
  Stream.updateService(Counter, (counter) => ({ count: counter.count + 1 }))
)

const program = Stream.runCollect(stream).pipe(
  Effect.provideService(Counter, { count: 0 })
)
// => [{ count: 1 }]

Stream.withSpan

Wraps the stream in a tracing span, eliminating the ParentSpan requirement.

import { Effect, Stream } from "effect"

const program = Stream.fromArray([1, 2, 3]).pipe(
  Stream.withSpan("numbers"),
  Stream.runCollect
)
// => [1, 2, 3]   (emitted under a "numbers" span)

Note

Two related operators stop a stream from the outside rather than recovering from an error: Stream.interruptWhen interrupts an in-progress pull as soon as an effect completes, while Stream.haltWhen stops cleanly after the current element. Both are covered on the Concurrency page.