Avoid substream subscription timeout for empty prefixAndTail tails

[He-Pin]

Motivation

prefixAndTail emits (prefix, tailSource) and then waits for the tailSource to be materialized before pulling upstream again. When a downstream consumer intentionally discards the tail Source — a common pattern in protocol parsers using splitWhen + prefixAndTail + mapAsync where a known-empty substream is skipped — the stage holds the substream open until the configured pekko.stream.materializer.subscription-timeout fires (default 5 s), silently stalling the rest of the pipeline.

Reproduces the long-standing akka/akka#20008:

Source(List("2", "a", "b", "0", "3", "a", "b", "c", "0", "1", "a"))
  .splitWhen(_.matches("\\d+"))
  .prefixAndTail(1)
  .mapAsync(1) {
    case (Seq("0"), _)             => Future.successful("")   // intentionally drop tail
    case (Seq(_),   elements)      => elements.runWith(Sink.seq).map(_.mkString)
  }
  .concatSubstreams
  .runWith(Sink.seq)

Without the fix every discarded empty tail adds a 5 s wait before the next group is emitted.

Modification

PrefixAndTail now issues a single speculative pull(in) from openSubstream() as soon as the tail Source has been emitted:

• If upstream finishes before any further element arrives, the stage completes immediately instead of waiting for the substream subscription timer.
• If upstream pushes an element before the tail Source has been materialized, the element is buffered in a one-element slot and delivered as the first element on the first downstream pull of the tail; upstream finish that arrives while the buffer is occupied is deferred and replayed together with the buffered element when the tail finally subscribes.
• Subscription-timeout behaviour for tails that are not subscribed but still have unread upstream elements is unchanged.

Operator docs updated to describe the new pre-pull / one-element-buffer semantics.

Result

Discarding a tail Source from prefixAndTail no longer blocks the parent flow for the subscription-timeout window when the tail is empty. The directional test for the issue scenario now completes in ~60 ms instead of ~15 s.

Tests

• sbt \"stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowPrefixAndTailSpec\" – 18/18 passed locally (incl. new "complete promptly on empty tail without waiting for subscription timeout (akka #20008)" and "deliver buffered tail element when tail is materialized after upstream emitted one").
• Verified the new directional test fails against the pre-fix implementation by stashing the impl change and re-running the same testOnly.
• scalafmt --list – clean for the changed files.
• Full stream-tests/Test/test and +mimaReportBinaryIssues – Not run locally, deferred to CI (Check / Binary Compatibility should cover MiMa).

References

Refs akka/akka#20008