Skip to content
LogoLogo

Error Handling in Streams

Stream recovery replaces a failed source with another source. Use stream operators to control recovery during consumption, and ordinary try/catch when the whole consuming function should handle a typed failure. Elements emitted before a failure are not rolled back.

Recovering typed failures and defects

// run: Console.log(JSON.stringify([readRecovered(), tagged.runCollect().map((result) => result._tag)]))
error Disconnected
 
const primary = Stream.make("live-a", "live-b").concat(Stream.fail(new Disconnected()))
const recovered = Stream.catch(primary, () => Stream.succeed("cached"))
 
function readRecovered() {
  return recovered.runCollect() // ["live-a", "live-b", "cached"]
}
 
const tagged = Stream.catch(
  primary.map((value) => Result.succeed(value)),
  () => Stream.make("cached").map((value) => Result.fail(value))
)
// Output: [["live-a","live-b","cached"],["Success","Success","Failure"]]

Stream.catch receives a typed failure and returns the replacement Stream. Recovery abandons the rest of the failed source; it does not resume after its failing step. Tagging each source with Result is useful when downstream code needs to distinguish live data from fallback data.

// run: Console.log(JSON.stringify(boundary.runCollect()))
const boundary = Stream.die(new Error("decoder invariant failed")).catchDefect(() => Stream.succeed("unreadable"))
// Output: ["unreadable"]

catchDefect handles defects at a deliberate boundary. catchCause receives the full Cause, including failure, defect, and interruption reasons; use it only when that full information is needed. Ordinary VibeLang catch handles typed failures only, and propagates a mixed cause containing defects or interruption.

Selective recovery

// run: Console.log(JSON.stringify([recoverKnown(Stream.fail(new NotReady())).runCollect(), selected.runCollect(), defectOnly.runCollect()])); try { recoverKnown(Stream.fail(new AccessDenied())).runCollect() } catch (error) { Console.log(error._tag) }
error NotReady
error AccessDenied
 
function recoverKnown(source: Stream.Stream<string, NotReady | AccessDenied>) {
  return source.catchTag("NotReady", () => Stream.succeed("pending"))
}
 
const selected = Stream.fail("busy" as "busy" | "forbidden").catchFilter(Filter.fromPredicate((error: "busy" | "forbidden") => error === "busy"), () => Stream.succeed("try later"))
 
const defectOnly = Stream.die("corrupt frame").catchCauseFilter(Cause.findDie, () => Stream.empty)
// Output: [["pending"],["try later"],[]]
// Output: AccessDenied

catchTag removes the handled tagged error from the output failure type. catchFilter handles values selected by a Filter and preserves failures it rejects. catchCauseFilter selects from the full Cause; Cause.findDie selects a defect reason rather than treating every failure as recoverable.

Cleanup after failure

// run: try { consumeChecked() } catch (error) { Console.log(error) }
function consumeChecked() {
  const source = Stream.succeed("accepted")
    .concat(Stream.fail("read failed"))
    .onError(() => Console.log("input failed; closing reader"))
  try {
    return source.runCollect()
  } finally {
    Console.log("consumer finished")
  }
}
// Output: input failed; closing reader
// Output: consumer finished
// Output: read failed

onError runs its callback on an unsuccessful stream exit and preserves the failure if that callback succeeds. It is useful for diagnostics tied to a source. Use Stream.ensuring for all exits or finally for cleanup belonging to the consuming function; cleanup failure is never silently discarded.

Retrying a source

// run: (() => { let attempts = 0; provide Sensor = { read: () => { attempts++; if (attempts < 4) throw new Offline(); return 42 } } { Console.log(JSON.stringify(readings.runCollect()), attempts) } })()
error Offline
 
service Sensor {
  read(): number!Offline
}
 
const readings = Stream.fromEffect(() => Sensor.read()).retry(Schedule.recurs(3))
// Output: [42] 4

retry starts the wrapped source again after a typed failure, according to its schedule. Retrying a multi-element source can repeat elements already delivered. Retry the smallest source that represents the intended operation, and make downstream side effects safe to repeat. An exponential schedule adds backoff; a finite recurrence schedule bounds the number of retries.

Narrowing error types

// run: try { keepSyntaxErrors(Stream.fail(new SyntaxError("bad syntax"))).runCollect() } catch (error) { Console.log(error.message) }; Console.log(exit(keepSyntaxErrors(Stream.fail(new Error("defect"))).runCollect()).isFailure())
function keepSyntaxErrors(source: Stream.Stream<string, Error>) {
  return Stream.catch(source, (error) => {
    if (error instanceof SyntaxError) return Stream.fail(error)
    return Stream.die(error)
  })
}
// The returned Stream has E = SyntaxError.
// Output: bad syntax
// Output: true

Refail the errors callers are expected to handle and turn impossible failures into defects. Stream.orDie(source) makes every typed failure a defect. These operators change the failure channel; they do not make a faulty source succeed.

Timeout failures and replacement sources

// run: Console.log(JSON.stringify(all([() => endWhenIdle.runCollect(), () => { const result = Effect.result(() => failWhenIdle.runCollect()); return result._tag }, () => { const outcome = exit(defectWhenIdle.runCollect()); return outcome.isFailure() }, () => fallbackWhenIdle.runCollect()], { concurrency: 4 })))
error Idle
 
const endWhenIdle = Stream.never.timeout("3 seconds")
const failWhenIdle = Stream.never.timeoutOrElse({
  duration: "3 seconds",
  orElse: () => Stream.fail(new Idle())
})
const defectWhenIdle = Stream.never.timeoutOrElse({
  duration: "3 seconds",
  orElse: () => Stream.failCause(Cause.die("idle invariant"))
})
const fallbackWhenIdle = Stream.never.timeoutOrElse({
  duration: "3 seconds",
  orElse: () => Stream.make("backup-a", "backup-b")
})
// Output: [[],"Failure",true,["backup-a","backup-b"]]

timeout completes normally when an upstream pull waits too long. timeoutOrElse switches to the Stream returned by orElse, so that Stream can fail, defect, or emit fallback data. The timeout applies to each upstream pull, not to total consumption time, and does not time the replacement stream.

The matching Effect page: effect.website/docs/v4/stream/error-handling

Suggest changes to this page