He-Pin opened a new issue, #3163:
URL: https://github.com/apache/pekko/issues/3163
### Summary
Three Pekko Streams operators — `Sink.lazySink`, `Source.futureSource`, and
`Sink.last` / `Sink.lastOption` / `Sink.takeLast` — can leave their
materialized `Future` uncompleted forever when the stream is terminated
abruptly (actor-system shutdown, materializer shutdown, async boundary
isolation) before the operator reaches its normal completion path.
Other operators with the same materialized-value shape (`Source.lazySource`,
`Flow.lazyFlow`, `Sink.ignore`, `Sink.head` / `Sink.headOption`, `Sink.seq`,
`Sink.eagerFutureSink`, etc.) all complete their `Promise` inside a `postStop`
override using `AbruptStageTerminationException`. The three operators above are
missing that override, leaving the future hanging.
### Symptom
```scala
val mat: Future[M] = source.toMat(Sink.lazySink(factory))(Keep.right).run()
// Actor system / materializer terminated before first element arrives
Await.result(mat, 5.seconds) // hangs forever, never fails
```
Same shape with `Source.futureSource(longRunningFuture)` and `Sink.last` /
`Sink.lastOption` / `Sink.takeLast`.
### Locations
All three stages lack a `postStop` override that completes the materialized
promise when the stage is finalized without having run through its normal
completion path.
1. **`Sink.lazySink`** —
`stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala`
- Materialized `Promise[M]` is completed in `onPush`, `onUpstreamFinish`,
`onUpstreamFailure`.
- No `postStop` override.
- Neighbouring `EagerFutureSink` in the same file has a correct
`postStop` override that fails the promise with
`AbruptStageTerminationException` — `LazySink` does not.
2. **`Source.futureSource` / `FutureFlattenSource`** —
`stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphStages.scala`
- Materialized `Promise[M]` is completed in `onFutureSourceCompleted` and
in the initial handler's `onDownstreamFinish` (with `StreamDetachedException`).
- No `postStop` override.
- Source stages are the most vulnerable: with an async boundary, the
source interpreter may have no boundary inputs to receive the abort error at
all, so the completion paths may never fire.
- Neighbouring stages (`LazySource`, `FutureFlow`, `MaybeSource`,
`InputStreamSource`) all have `postStop` overrides.
3. **`Sink.last` / `Sink.lastOption` / `Sink.takeLast` (shared
`TakeLastStage`)** —
`stream/src/main/scala/org/apache/pekko/stream/impl/Sinks.scala`
- Materialized `Promise[immutable.Seq[T]]` is completed in
`onUpstreamFinish` / `onUpstreamFailure`.
- No `postStop` override.
- Neighbouring `HeadOptionStage` (`Sink.head` / `Sink.headOption`) and
`SeqStage` (`Sink.seq`) in the same file have a correct `postStop` override —
`TakeLastStage` does not.
### Why the existing tests miss it
The existing abrupt-termination tests use simple topologies with boundary
sources (e.g. `Source.fromPublisher(probe)`). In those topologies, the abort
error reliably propagates to the stage via `onUpstreamFailure`, which completes
the promise through the normal failure path. The tests therefore exercise the
**error-propagation** path, not the `postStop` fallback.
When the error propagation path fails to reach the stage — async boundaries,
complex topologies, or the source interpreter having no boundary inputs —
`postStop` is the only remaining opportunity to complete the materialized
promise. The interpreter always calls `postStop` during stage finalization (via
`ActorGraphInterpreter.postStop` → `tryAbort` → `GraphInterpreter.finish` →
`finalizeStage`). Without a `postStop` override, that opportunity is wasted and
the promise hangs.
### Reproduction sketch
```scala
// lazySink variant
val mat = Source.empty
.delay(10.seconds)
.toMat(Sink.lazySink(_ => Sink.ignore))(Keep.right)
.run()
// Terminate the materializer before the first element.
// mat should fail with AbruptStageTerminationException but instead hangs
forever.
```
Similar sketches apply for `Source.futureSource` (pass an uncompleted
future, then terminate) and `Sink.last` (terminate before upstream finishes).
### Proposed fix
For each of the three stages, add a `postStop` override completing the
materialized promise when not already completed, matching the pattern used in
neighboring operators:
```scala
override def postStop(): Unit = {
if (!promise.isCompleted)
promise.failure(new AbruptStageTerminationException(this))
}
```
Apply this to:
- `LazySink`'s `stageLogic` in `Sinks.scala`
- `FutureFlattenSource`'s `GraphStageLogic` in `GraphStages.scala`
- `TakeLastStage` in `Sinks.scala`
Directional tests should cover abrupt termination with an async boundary
between the stage under test and its upstream/downstream, so that the
error-propagation path does NOT reach the stage and `postStop` is the only
fallback. Existing abrupt-termination tests in `EagerFutureSinkSpec`,
`HeadSinkSpec`, and `TakeLastSinkSpec` provide templates; new tests should
assert the `AbruptStageTerminationException` via the `postStop` fallback.
### Environment
- Apache Pekko streams, current `main`
- Affects: `Sink.lazySink`, `Sink.lazySinkAsync` variants,
`Source.futureSource`, `Sink.last`, `Sink.lastOption`, `Sink.takeLast`
--
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]
