ZStream.tapSink, either a flaky test or flaky implementation

[eyalfa]

I've encountered this failure on my local env (saw it once, wasn't able to reproduce):

 - ZStreamSpec / Combinators / tapSink / does not read ahead - 5 ms
        ✗ 1 was not equal to 6
        result == 6
        result = 1
        at /Users/efarago/dev/github/zio/streams-tests/shared/src/test/scala/zio/stream/ZStreamSpec.scala:3727 

I suspect the issue depends on ZStream.tapSink 's guarantees in face of downstream cancellation;
If it guarantees that all emitted values were processed by the sink as well this is a real bug,
on the other hand if this is not guaranteed then this test is flaky as it cannot expect all values to have been processed by the sink (it can be relaxed to actually asset what its name suggests).

looking at ZStream.tapSink 's implementation it seems to me that once downstream completes the merge will also be halted, this effectively means interrupting both sides of the merge (if still running) which is done by fiber interruption. since there's no guarantee which one is interrupted first the ensuring block in tapSink can't guarantee proper completion of the sink.

I was able to show this with an awkward test:

test("early termination with a slow sink") {
            for {
              ref    <- Ref.make(0)
              stream  = ZStream(1, 2, 3, 4, 5).rechunk(1).forever
              sink    = ZSink.foreach((n: Int) => ZIO.sleep(2.second) *> ref.update(_ + n))
              fib     <- stream.tapSink(sink).take(1).mapZIO(_ => ZIO.sleep(1.second)).runDrain.fork
              _       <- TestClock.adjust(1.second)
              _       <- fib.join
              result <- ref.get
              _       <- ZIO.fail(s"fiber shouldn't have completed! (result=$result)")
            } yield {
              assertTrue(result == 0)
            }
          }

the test is awkward since it's expected to halt (if tapSink's guarantee is strong enough), instead it explicitly fails when the stream execution completes.

[jdegoes]

/bounty $100 to fix the test or implementation

[algora-pbc]

💎 $100 bounty • ZIO

Steps to solve:

1. Start working: Comment /attempt #8792 with your implementation plan
2. Submit work: Create a pull request including /claim #8792 in the PR body to claim the bounty
3. Receive payment: 100% of the bounty is received 2-5 days post-reward. Make sure you are eligible for payouts

Thank you for contributing to zio/zio!

Add a bounty • Share on socials

Attempt	Started (GMT+0)	Solution
🟢 @kpritam	May 20, 2024, 6:41:30 AM	#8880
🟢 @varshith257	May 20, 2024, 1:08:13 PM	#8879

[kpritam]

/attempt #8792

Algora profile	Completed bounties	Tech	Active attempts	Options
@kpritam	2 ZIO bounties	Scala, Makefile, Kotlin & more		Cancel attempt

[varshith257]

/attempt #8792

Algora profile	Completed bounties	Tech	Active attempts	Options
@varshith257	1 bounty from 1 project	TypeScript, Go		Cancel attempt

[algora-pbc]

Note

The user @kpritam is already attempting to complete issue #8792 and claim the bounty. We recommend checking in on @kpritam's progress, and potentially collaborating, before starting a new solution.

[algora-pbc]

💡 @varshith257 submitted a pull request that claims the bounty. You can visit your bounty board to reward.