[I] LazySink, Source.futureSource, and Sink.last never complete their materialized Future on abrupt termination [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

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: notifications-unsubscribe@pekko.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org

Re: [I] LazySink, Source.futureSource, and Sink.last never complete their materialized Future on abrupt termination [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin closed issue #3163: LazySink, Source.futureSource, and Sink.last never complete their materialized Future on abrupt termination
URL: https://github.com/apache/pekko/issues/3163


-- 
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: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org