Add onDiscard callback to Flow.fromPublisher

[adamw]

Flow.fromPublisher buffers elements in a channel. Some of them could be received from the publisher but never emitted, and nothing released them. That leaks elements which hold resources, such as Netty's ref-counted ByteBufs. Tapir currently tracks these elements itself (softwaremill/tapir#5573).

• New overload fromPublisher(p, onDiscard: T => Unit), in core and in FlowReactiveStreams. Each element received from the publisher is either emitted, or passed to onDiscard exactly once. Elements are discarded when the flow fails, is interrupted, or stops early (e.g. take), and also when they arrive after the subscription is cancelled. onDiscard may run on publisher threads and must not throw.
• Behaviour change, also for the one-argument variant: when the publisher fails, elements already buffered are now emitted first, and then the error is rethrown. Before, they were dropped.
• The existing one-argument overloads are unchanged and pass a no-op callback. The new method is an overload, not a default parameter, so binary compatibility is kept.
• Added tests for publisher error, consumer exception, interruption, take, and elements arriving after cancellation. Also added a stress test that checks each element is either emitted or discarded, never both or neither. Updated doc/streaming/flows.md.

Closes #528

Open review findings

What the run's review left open, each with the reason recorded — including findings the fixer declined:

• sbt Test/compile fails with JGit NoWorkTreeException — This is a problem with the environment, not the code. git sees the worktree as non-bare (git rev-parse --is-bare-repository prints false; the git dir is /home/adamw/ox/.git/worktrees/zippy-astrodon-94aa2fb6). The exception comes from the sbt-git plugin, which uses JGit by default and reads this linked worktree as bare while the build loads. Nothing in project/ or the build configures this; zz-tmp-git.sbt was my temporary workaround, not the cause. To check the code, I recreated it with useReadableConsoleGit, which makes sbt-git use command-line git: sbt core/Test/compile flowReactiveStreams/Test/compile succeeded (warnings only, in older test files, no errors) and flowReactiveStreams tests passed. I then deleted the file again. Changing the shared build's git plugin setup is outside the scope of this change.
• sbt Test/compile fails with no diagnostic — Environmental. Full output shows sbt-git's JGit throws NoWorkTreeException ("Bare Repository has neither a working tree, nor an index") while loading the build, because this checkout is a git worktree; no source error is involved. With a temporary ThisBuild / git.gitUncommittedChanges := false override, sbt Test/compile succeeds across all modules (only pre-existing unrelated warnings). The temporary file was removed; no codebase change is needed.
• No test covers the drain's deferred-interrupt path in discardBuffered (core/src/main/scala/ox/flow/FlowCompanionReactiveOps.scala:141) — A deterministic test can't be written. In jox 1.1.2, receiveOrClosed only throws InterruptedException when it parks (Continuation.await). After doneOrClosed() has finished on the draining thread, a receive never parks: buffered or in-flight sends are returned or retried, and later cells are already CLOSED. The catch is reached only in a narrow race, when a concurrent onComplete/onError is still closing cells. I tried the suggested test (interrupt plus throw in the consumer, then assert the discards and the interrupt flag). It passed both with the catch removed and with the re-interrupt removed, so it did not cover the branch, and I reverted it. The suggested capacity-1 variant also doesn't help, because a blocked sender's Continuation is resumed without parking.
• sbt Test/compile exited 1 with no compile error in the output — Environmental. sbt-git's JGit fails in this git worktree with NoWorkTreeException: Bare Repository has neither a working tree, nor an index while evaluating gitUncommittedChanges, so the build fails before compiling. It is not a code problem. With a temporary zz-tmp-git.sbt that sets ThisBuild / git.gitUncommittedChanges := false, Test/compile and the core reactive and flowReactiveStreams tests all pass. I deleted the temporary file afterwards.
• sbt Test/compile exits 1 with no compiler diagnostics — Environmental, not a code problem. In this git worktree, sbt-git's JGit throws NoWorkTreeException: Bare Repository has neither a working tree, nor an index while evaluating gitUncommittedChanges during build load. The build fails before any compilation, and the full output shows that stack trace above the timing line. With a temporary zz-tmp-git.sbt setting ThisBuild / git.gitUncommittedChanges := false, Test/compile succeeds. With the same file, core/testOnly ox.flow.reactive.* (44 passed, 7 ignored) and flowReactiveStreams/test (2 passed) also pass. I deleted the file afterwards, so it is not left in the tree.