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

[He-Pin]

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

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

// 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:

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