He-Pin opened a new pull request, #3016:
URL: https://github.com/apache/pekko/pull/3016
## Motivation
`StreamRefsSpec` *"pass cancellation upstream across remoting before
elements has been emitted"* is flaky (#610, upstream akka#30844).
When the consuming `SourceRef` cancels gracefully **before any element**, it
sends a `RemoteStreamCompleted` to the origin `SinkRef` and then terminates. On
transports such as Artery, system messages (the consumer's `Terminated` /
deathwatch) travel on a **dedicated lane** and can overtake the ordinary-lane
`RemoteStreamCompleted`. The `SinkRef` then observed `Terminated` first and,
with no completion recorded yet, immediately failed with
`RemoteStreamRefActorTerminatedException` ("message loss may have happened") —
so the origin's `watchTermination` yielded `Failure` instead of `Done`.
The `SinkRef` already guards its *own* termination against this lane race
(`finishedWithAwaitingPartnerTermination` + `setKeepGoing`), and the
`SourceRef` already waits a `finalTerminationSignalDeadline` grace period when
*its* partner terminates (`SourceRefImpl` `TerminationDeadlineTimerKey`). The
**symmetric** case on the `SinkRef` side was missing.
## Modification
When the `SinkRef` observes its partner `Terminated` with no recorded
completion/failure, schedule a `finalTerminationSignalDeadline` grace timer
instead of failing immediately. If the in-flight `RemoteStreamCompleted` /
`RemoteStreamFailure` arrives within the deadline, it completes/fails cleanly
via the existing handlers; otherwise the timer fires and message loss is
declared as before. The change is confined to the `case _ =>` branch (partner
terminated before any completion) — the `Some(Success)` / `Some(Failure)`
branches are unchanged.
Removed the corresponding `// FIXME akka#30844` from the test.
## Result
```
[info] Tests: succeeded 24, failed 0 (full StreamRefsSpec)
cancellation cases: 5/5 on repeat
stream/mimaReportBinaryIssues: [success] no binary issues
```
## Tests
`stream-tests/testOnly org.apache.pekko.stream.scaladsl.StreamRefsSpec` — 24
pass; cancellation subset 5/5 on repeat. `scalafmt` run;
`stream/mimaReportBinaryIssues` clean. Change is internal to `SinkRefStageImpl`.
## References
Fixes #610
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
