hold1 + concurrently releases bracket immediately
Nobody has claimed this yet.
- Dominant language
- Scala
- Stars
- 2.5k
- Forks
- 636
- Avg merge
- 2d 4h
- Merged PRs (30d)
- 7
Description
Using hold1 appears to break concurrently and/or bracket.
Here's a minimized version of some code we were using to keep a token fresh:
import fs2.Stream
import cats.effect._
import cats.syntax.all._
import scala.concurrent.duration.DurationDouble
import cats.effect.unsafe.implicits.global
(Ref[IO].of(none[Boolean]), Ref[IO].of(0))
.flatMapN { (flag, count) =>
val acquire = println("acquire") *> flag.set(true.some)
val release: Unit => IO[Unit] = _ => println("release") *> flag.set(none)
val increment = count.updateAndGet(_ + 1).flatTap(c => println(s"increment: $c"))
Stream
.bracket(acquire)(release)
.flatMap { _ =>
Stream.emit('e').concurrently {
Stream.repeatEval(increment).metered(0.1.seconds)
}
}
.evalTap(v => flag.get.flatMap(f => println(s"emitted: $v $f")))
.hold1
.evalTap { valueIO =>
(valueIO.get, flag.get)
.flatMapN((v, f) => println(s"check: $v $f"))
.delayBy(.3.seconds)
.replicateA(2) *> println("Done")
}
.compile
.drain
}
.unsafeRunTimed(5.seconds)
Output:
acquire
emitted: e Some(true)
release
check: e None
check: e None
Done
Notably, increment never runs, release is called immediately, and stream is only consumed after.
The behavior I expected would have been:
acquire
emitted: e Some(true)
increment: 1
increment: 2
increment: 3
check: e Some(true)
increment: 4
increment: 5
increment: 6
check: e Some(true)
Done
release
I don't know enough about the internals to figure out why this is happening, though it seems like it may be related to #3123, and possibly #2936.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the minimized Scala reproduction in the issue and run it to compare the observed and expected output. Trace the interaction between Stream.hold1, bracket, and concurrently, then review the related issues #3123 and #2936. Done means resource release and background increments follow the expected ordering while the held value is consumed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100