(pekko) 01/06: fix(stream): force-start supervisor on UnstartedCell race in StageActor.localCell

[he...@apache.org]

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: 虎鸣 <he...@alibaba-inc.com>
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: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org