typelevel / typelevel/fs2

hold1 + concurrently releases bracket immediately

Open
#3,567 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

bug
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

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.