fix: preserve StageActor demand across fused streams

[He-Pin]

Motivation

StageActor lazy dispatch coalesces a burst into one async callback. A callback can enqueue interpreter work that replenishes downstream demand; dispatching the next message before that work runs makes it observe stale demand and can drop data, as reported in #3459.

Fixed nested calls such as interpreter.execute(2) or execute(16) are not a general fix: the required work depends on the fused topology depth, and the nested work is not charged to the actor shell's processing quota. Restoring a saved activeStage is also unsafe when interpreter execution may have completed and released that stage.

Modification

• Add an internal event-limit-aware async handler contract so lazy StageActor dispatch and the graph interpreter share the enclosing shell's event budget.
• Drain callback-generated interpreter work before dispatching the next StageActor message. If the budget is exhausted, park the dispatcher and resume it only after the interpreter queue becomes idle; this has no topology-dependent event limit.
• Avoid entering GraphInterpreter.execute when a callback enqueues no interpreter work, preserving the common no-graph-work hot path.
• Set activeStage before every callback without restoring a potentially finalized stage.
• Stop and clear queued dispatch after stage completion or callback failure.
• Add directional tests for per-element demand replenishment, a 256-stage fused topology, sync-processing-limit = 1, stage completion, and callback failure.

Result

Demand is replenished before the next StageActor message independently of fused topology depth. Actor/interpreter fairness remains bounded by the existing shell quota, lazy dispatch still coalesces mailbox traffic, and completed or failed stages are not dispatched again.

The original pekko-connectors UDP reproduction was run from connector commit 1c15f3b6f; only PekkoCoreDependency.currentVersion changed:

Pekko Core	sbt "udp/test"
2.0.0-M4	1/5 passed; all four Scala/Java 100-datagram send/receive tests timed out
2.0.0-M4+14-3f07c449-SNAPSHOT	5/5 passed

This directly verifies the connector regression, not only the reduced Core reproducer.

The direct StageActorRefBenchmark.lazy_stage_actor_ref_tell_10k JMH benchmark was run on the same machine and OpenJDK 25.0.2 with -wi 3 -i 5 -w 2s -r 2s -f 3:

Revision	Throughput (ops/s, JMH 99.9% error)
origin/main (6be39c658f)	26,285,866 ± 6,002,205
This PR	27,987,687 ± 3,124,268

The PR point estimate is 6.5% higher, but the confidence intervals overlap. The supported conclusion is that this run found no measurable throughput regression, not that the change is statistically faster.

Tests

• pekko-connectors: sbt "udp/test" — 2.0.0-M4 failed 4/5; PR snapshot passed 5/5.
• sbt "stream-tests / Test / testOnly org.apache.pekko.stream.FusingSpec org.apache.pekko.stream.scaladsl.StageActorRefSpec org.apache.pekko.stream.scaladsl.ActorRefSourceSpec org.apache.pekko.stream.scaladsl.StreamRefsSpec" — 80/80 passed.
• sbt -Dpekko.stream.fuzzing-mode=on "stream-tests / Test / testOnly org.apache.pekko.stream.FusingSpec" — 24/24 passed with randomized event ordering and chasing disabled.
• sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec" — 13/13 passed.
• sbt validatePullRequest — Streams suite passed 3,418/3,418. The aggregate command failed outside this change: LevelDB JNI is unavailable on this ARM64 host; two Remote tests failed on local bind/TLS timing; IntegrationDocTest failed while instantiating UnboundedMailbox.
• sbt "stream / mimaReportBinaryIssues" "++3.3.8" "stream / mimaReportBinaryIssues" — passed.
• sbt headerCreateAll "+headerCheckAll" checkCodeStyle, native scalafmt, and git diff --check — passed.
• sbt "bench-jmh / Jmh / run -wi 3 -i 5 -w 2s -r 2s -f 3 .*StageActorRefBenchmark.lazy_stage_actor_ref_tell_10k.*" — results shown above.

References

Fixes #3459

[pjfanning]

LGTM as it fixes the issue (or seems likely to).

Claude AI raised some concerns (below).

Two things I'd want addressed

1. activeStage is clobbered by the nested execute().

runAsyncInput sets activeStage = logic (GraphInterpreter.scala:577) and calls afterStageHasRun(logic) at the end (line 592). The new execute(2) runs processEvent, which does activeStage = null (line 602) and then reassigns it to whichever connection's owner it processes (lines 616/636/655/666). So when handler(item) returns for message 2, activeStage is no longer this logic.

This is benign for finalization — afterStageHasRun takes the logic explicitly, so it finalizes the right stage. But reportStageError (line 442) uses the mutable activeStage field, not a parameter. If handler(item) throws for message 3, the error is attributed to, and failStage is called on, whatever stage the previous nested execute last touched. That's a misattributed failure — wrong stage failed, wrong stage named in the log.

I have not constructed a failing case for this, so I'd flag it as a question rather than a defect: is it worth saving and restoring activeStage around the execute(2) call? It's a two-line change if so.

2. The 2 deserves a firmer justification than "sufficient".

The PR says "1 event slot for the push + chase budget for the pull". Reading execute (line 486), eventsRemaining is decremented per dequeued event and chaseCounter = math.min(ChaseLimit, eventsRemaining) — so with a limit of 2, the chase budget after the first dequeue is 1. That's enough for the simple push→pull chain the test exercises.

What I can't tell from the diff is what happens with a longer downstream chain inside the same shell, where the pull needs more than one chase hop to reach back. The events aren't lost — they stay queued and get drained by the runBatch that follows — but demand may still not be replenished by the time message N+1 is handled, which is exactly the bug. That would make the fix correct for the tested topology and partial for others.

A test with two or three map stages between the source and the async boundary would settle it. Worth asking the author whether that was considered.

[He-Pin]

Let's me check this again.

[pjfanning]

@He-Pin this is still ok to merge for me - the new commit does seem to address the highlighted issues