This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch perf/actorref-source-optimization in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 22f32e86fc13ff32b2c13dd28c1f517a1d82669a Author: 虎鸣 <[email protected]> AuthorDate: Fri Jun 19 17:58:40 2026 +0800 fix(stream): force-start supervisor on UnstartedCell race in StageActor.localCell Motivation: The first Source.actorRef (or any StageActor-backed stage) materialized on a fresh ActorSystem can race the async supervisor startup. When the RepointableActorRef still holds an UnstartedCell, localCell throws IllegalStateException instead of waiting for the real ActorCell. Modification: Add an UnstartedCell case to StageActor.localCell that force-starts the supervisor via Ask(Identify) — the Identify round-trip ensures the Supervise message has been processed and the cell has been swapped. Result: First materialization on a fresh system no longer throws. The fix is global — all stages using getStageActor/getEagerStageActor benefit. Tests: - sbt "stream-tests / Test / testOnly ...ActorRefSourceSpec" — 14/14 pass - New regression test creates 20 fresh ActorSystems and materializes Source.actorRef as the first operation on each. References: Refs akkadotnet/akka.net#8263 --- .../pekko/stream/scaladsl/ActorRefSourceSpec.scala | 23 ++++++++++++++++++++- .../org/apache/pekko/stream/stage/GraphStage.scala | 24 ++++++++++++++++------ 2 files changed, 40 insertions(+), 7 deletions(-) diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala index 255f41e3ad..c7538c7539 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/ActorRefSourceSpec.scala @@ -14,11 +14,12 @@ package org.apache.pekko.stream.scaladsl import scala.annotation.nowarn +import scala.concurrent.Await import scala.concurrent.duration._ import org.apache.pekko import pekko.Done -import pekko.actor.{ ActorRef, Status } +import pekko.actor.{ ActorRef, ActorSystem, Status } import pekko.stream.{ OverflowStrategy, _ } import pekko.stream.testkit._ import pekko.stream.testkit.Utils._ @@ -227,5 +228,25 @@ class ActorRefSourceSpec extends StreamSpec { mat.shutdown() } } + + "materialize even when the stream supervisor is not yet started" in { + // Regression: the first Source.actorRef materialization on a fresh ActorSystem + // can race the async supervisor startup, hitting an UnstartedCell in StageActor.localCell. + // The fix force-starts the supervisor via Ask(Identify) before returning the cell. + (1 to 20).foreach { i => + val sys = ActorSystem(s"ActorRefSource-startup-race-$i") + try { + val mat = Materializer(sys) + val ref = Source + .actorRef[Int]({ case "ok" => CompletionStrategy.draining }, PartialFunction.empty, 8, OverflowStrategy.fail) + .to(Sink.ignore) + .run()(mat) + ref should not be null + ref ! "ok" + } finally { + Await.result(sys.terminate(), 10.seconds) + } + } + } } } diff --git a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala index dcd6faab5f..8325403b25 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala @@ -21,13 +21,15 @@ import java.util.concurrent.atomic.AtomicReference import scala.annotation.nowarn import scala.annotation.tailrec import scala.collection.{ immutable, mutable } -import scala.concurrent.{ Future, Promise } +import scala.concurrent.{ Await, Future, Promise } import scala.concurrent.duration.FiniteDuration import org.apache.pekko import pekko.{ Done, NotUsed } import pekko.actor._ import pekko.annotation.InternalApi +import pekko.pattern.ask +import pekko.util.Timeout import pekko.dispatch.AbstractNodeQueue import pekko.japi.function.{ Effect, Procedure } import pekko.stream._ @@ -312,12 +314,22 @@ object GraphStageLogic { private object StageActor { def localCell(ref: ActorRef, description: String): ActorCell = ref match { - case ref: LocalActorRef => ref.underlying - case ref: RepointableActorRef => - ref.underlying match { + case ref: LocalActorRef => ref.underlying + case r: RepointableActorRef => + r.underlying match { case cell: ActorCell => cell - case unknown => - throw new IllegalStateException(s"$description must be a local actor, was [${unknown.getClass.getName}]") + case _: UnstartedCell => + implicit val timeout: Timeout = r.system.settings.CreationTimeout + Await.result(r ? Identify(null), timeout.duration) + r.underlying match { + case cell: ActorCell => cell + case unknown => + throw new IllegalStateException( + s"$description must be a local actor, was [${unknown.getClass.getName}]") + } + case unknown => + throw new IllegalStateException( + s"$description must be a local actor, was [${unknown.getClass.getName}]") } case unknown => throw new IllegalStateException(s"$description must be a local actor, was [${unknown.getClass.getName}]") --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
