merge: pull from upstream only when downstream has demand

[istreeter]

What

Since #3610, merge pulls a chunk from each side as soon as the previous chunk has been handed downstream, rather than waiting for downstream to ask for the next one. So both sides read one chunk ahead of demand.

This moves the guard release to after the chunk, so each side pulls in response to downstream's next uncons. The change is one line — the stream sent through merge's internal channel goes from Stream.exec(release) ++ chunk to chunk ++ Stream.exec(release).

Why

fs2 is pull-based. map, filter, evalMap, flatMap and through all pull from upstream only when downstream wants an element. merge is now the exception — and since merge is used internally by either, mergeHaltBoth/L/R, observe, zipWithLatest and pull.timed, the read-ahead is inherited by combinators whose call sites give no hint of it.

That costs real memory in our applications, where a single chunk can hold a substantial amount of data. Buffering an extra chunk per side, invisibly, takes away control we depend on.

The asymmetry is what bothers me most. Anyone can opt into read-ahead with .prefetch, on exactly the stream they choose. There is no way to opt out of eagerness baked into merge — you can't call mergeAndAwaitDownstream from inside either. Being pull-based is a strength of fs2, and I'd rather see the library lean into it than relax it combinator by combinator.

This does mean dropping the documented equivalence with Stream(this, that).parJoinUnbounded. I think that's the right trade, and I'd rather merge matched the rest of fs2 than matched parJoin.

This keeps the #3610 fix

#3610 fixed a genuine deadlock and this PR does not revert it. Previously the next pull waited for the scope of the previous chunk to close. Scope#lease covers ancestor scopes, and parJoin holds a lease for as long as each inner stream runs — so a non-terminating inner stream meant the release never ran.

Appending Stream.exec(release) avoids that: it is not a finalizer and opens no scope, just a suspended pull step, and the only thing that resumes it is another uncons — which parJoin and parEvalMap both do eagerly. Both tests added by #3610 (merge produces when concurrently handled and issue #3598) are unchanged and still pass.

Evidence

The read-ahead was visible in assertions. merge not emit ahead expected every value twice — List(v, v, v+1, v+1, ...) — because the prefetch happened before the downstream evalMap had updated its Ref. It now expects List(v, v+1, v+2). TimedPullsSuite's "After the first uncons, timeouts start immediately" reverts to its pre-#3610 form for the same reason.

Full coreJVM/test passes (1903 tests), as do scalafmtCheckAll and mimaReportBinaryIssues.

API

mergeAndAwaitDownstream becomes an alias for merge. With merge no longer reading ahead, the only thing that distinguished them was waiting on scope closure — the deadlock. The signature is unchanged so this is binary compatible, and I have deliberately not deprecated it; that seems like a maintainer's call.

That also makes #3663 unnecessary: those three mergeAndAwaitDownstreamHalt* helpers exist to escape the eagerness, and there is nothing left to escape. #3655 may be related, but I should be upfront that I could not reproduce it either before or after this change.