[He-Pin]

Motivation

After #3730 removes repeated oldest-cursor scans, Sink.asPublisher(fanout = true) still scans every subscription to calculate upstream demand even when outstanding requests already reserve all available buffer space, or one subscriber's demand already covers that space.

This is stacked on #3730 at 9bed9a0e523da8c7356f46a5928306459584c018. The temporary base branch stack/pr-3730-fanout-publisher-cursor-scan points to that exact commit so this PR shows only the additional change. Merge #3730 first, then rebase/retarget this PR to main before merging it.

Modification

• Skip the demand scan when outstanding upstream requests cover the available buffer capacity.
• Stop scanning once a subscriber's demand reaches the full available capacity. Keep the existing request arithmetic and subscription lifecycle; no cached demand or request batching is introduced.
• Add eight focused tests for outstanding requests, capacity one, finite and overflowing demand, cancellation, joining readers, and completion with retained elements.
• Extend the existing publisher benchmark with request batches of 1, 16, and Long.MaxValue.

Result

The scan can visit zero subscriptions when capacity is already reserved, or stop at the first subscriber whose demand covers the capacity. Small requests can still require a full scan; this does not remove the worst-case repeated scans for many subscribers requesting one element at a time.

Exploratory JMH results, comparing the parent commit with this change using identical benchmark sources on macOS arm64 / Temurin 17.0.21 / Scala 2.13.18. One JMH thread, two forks, 2 x 1 s warmup and 3 x 1 s measurement per fork. Throughput counts input elements delivered to every subscriber, not total subscriber deliveries.

Subscribers	Request batch	Parent elements/s	This change elements/s	Ratio
1	1	2,089,613	1,999,753	0.957
32	1	245,838	243,633	0.991
1	16	2,291,339	2,289,714	0.999
32	16	1,243,455	1,436,254	1.155
1	unbounded	2,381,075	2,329,985	0.979
32	unbounded	1,590,420	1,710,606	1.076

All corresponding 99.9% confidence intervals overlap. To investigate the lower single-subscriber, batch-1 mean, a separate longer comparison reversed the execution order (this change, then parent), with three forks, 5 x 1 s warmup and 5 x 2 s measurement. It measured 1,988,046 +/- 123,156 elements/s for the parent and 1,950,412 +/- 89,989 for this change, a ratio of 0.981. These intervals also overlap. A small performance loss in that workload remains possible; these measurements do not establish a universal throughput improvement or absence of regression.

Tests

• Scala 2.13.18: SubscriberManagementSpec (8 tests), plus FanoutPublisherBehaviorSpec, ResizableMultiReaderRingBufferSpec, and FlowSpec (64 tests): passed. The new demand tests also pass against the parent implementation.
• stream-tests-tck / Test / testOnly org.apache.pekko.stream.tck.FanoutPublisherTest: 31 passed; 7 standard untested TCK cases skipped.
• Scala 3.3.9: SubscriberManagementSpec and FanoutPublisherBehaviorSpec: 22 passed.
• +bench-jmh/compile and +stream/mimaReportBinaryIssues: passed on Scala 2.13.18 and 3.3.9.
• sbt headerCreateAll, scoped header checks for stream, stream-tests / Test, and bench-jmh, native scalafmt, and git diff --check: passed.
• Import sorting: sortImports is a root command alias, so the module-scoped invocation is unavailable. The scoped equivalent (scalafixEnable; stream-tests/scalafixAll SortImports) fails with the existing rule's NoSuchMethodError: scala.meta.package$.Input(). Incidental edits were discarded; native scalafmt, including its SortImports rewrite, passed.
• Independent adversarial review and Qoder CLI stdout review: no blocker or must-fix findings. Qoder's first attempt ended with a service error and a second attachment-only attempt stalled and was terminated; the final review with full source context included directly in the prompt completed successfully. All attempts retained stdout logs.
• Full local validation was intentionally not run at the requester's direction; the remaining build, style, header, documentation, and test matrix validation is delegated to remote CI.

References

Refs #3730. This PR is the second change in the fanout publisher optimization stack.