Ensure

ensure.sc

//> using scala 3.3.1
//> using dep co.fs2::fs2-core::3.9.4

import cats.syntax.all.*
import cats.effect.IO
import fs2.Stream
import cats.effect.unsafe.implicits.global

Stream(1, 2)
  .covary[IO]
  .debugChunks(formatter = (ch) => s"Before: $ch")
  .ensure(new Error("!"))(_ => true)
  .debugChunks(formatter = (ch) => s"After: $ch")
  .compile
  .drain
  .unsafeRunSync()

// Before: Chunk(1, 2)
// After: Chunk(1)
// After: Chunk(2)