[BUG]: Stream.take resource safety

[johnhungerford]

Version

1.0-RC1+14-a287ac73+20250724-1908-SNAPSHOT

Scala Version

3.7.1

Expected Behavior

When a stream is defined using an Emit[Chunk] effect wrapped in Sync.ensure or Scope.ensure, and then the stream is run using .take the finalizer should run.

Actual Behavior

The finalizer does not run

Steps to Reproduce

Run the following tests:

"Sync.ensure" - {
    "take" in run {
        for
            ref <- AtomicRef.init(List.empty[String])
            stream = Stream:
                Sync.ensure(ref.getAndUpdate("finalized" :: _)):
                    Loop(0)(i => ref.getAndUpdate(i.toString :: _).andThen(Emit.valueWith(Chunk(i))(Loop.continue(i + 1))))
            emittedValues <- stream.take(5).run
            refValues     <- ref.get
        yield assert(emittedValues == Chunk(0, 1, 2, 3, 4) && refValues == List("finalized", "4", "3", "2", "1", "0"))
    }
}

"Scope.ensure" - {
    "take" in run {
        for
            ref <- AtomicRef.init(List.empty[String])
            stream = Stream:
                Scope.run:
                    Scope.ensure(ref.getAndUpdate("finalized" :: _)).andThen:
                        Loop(0)(i => ref.getAndUpdate(i.toString :: _).andThen(Emit.valueWith(Chunk(i))(Loop.continue(i + 1))))
            emittedValues <- stream.take(5).run
            refValues     <- ref.get
        yield assert(emittedValues == Chunk(0, 1, 2, 3, 4) && refValues == List("finalized", "4", "3", "2", "1", "0"))
    }
}

Current Workaround

No response

[johnhungerford]

@fwbrasil @hearnadam @ahoy-jon I think this may point to a serious issue with our streaming model that I don't see an obvious solution to.

Treating Emit effects as streams requires running effects only in part to support a lot of expected functionality, such as take and its variants (see example above). ensure methods like Sync.ensure and Scope.ensure apparently require the effect to run to completion for cleanup to be triggered. Is there another way to make this work? How does Abort short-circuit with finalizers but not Emit?

[johnhungerford]

Actually finalizers don't run with Abort either. The following test fails with "called was false"

"abort" in run {
    val ex     = new Exception
    var called = false
    Abort.run[Any](Sync.ensure { called = true } {
        Abort.fail(ex).andThen(Sync.defer(1))
    }).map { result =>
        assert(result == Result.Failure(ex))
        assert(called)
    }
}

[hearnadam]

@johnhungerford I think the short answer is that we can store finalizers in the stream, and only bind them as finalizers once the stream is run.

Something like:

sealed abstract class Stream[A, S]:
  def emit: (Unit < Emit[A] & S, Result[Any, Any] => Unit < S)

That doesn't help with the Scope case, but we can bind finalizers directly - like Stream.ensuring.