fix: release reference to initial interpreter shell after it is shut down

stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreterSpec.scala

@@ -22,6 +22,7 @@ import scala.concurrent.duration._
 import org.apache.pekko
 import pekko.Done
 import pekko.stream._
+import pekko.stream.impl.{ PhasedFusingActorMaterializer, StreamSupervisor }
 import pekko.stream.impl.ReactiveStreamsCompliance.SpecViolation
 import pekko.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
 import pekko.stream.scaladsl._
@@ -33,8 +34,7 @@ import pekko.stream.testkit.StreamSpec
 import pekko.stream.testkit.TestPublisher
 import pekko.stream.testkit.TestSubscriber
 import pekko.stream.testkit.Utils._
-import pekko.testkit.EventFilter
-import pekko.testkit.TestLatch
+import pekko.testkit.{ EventFilter, TestLatch }
 
 import org.reactivestreams.Publisher
 import org.reactivestreams.Subscriber
@@ -436,5 +436,60 @@ class ActorGraphInterpreterSpec extends StreamSpec {
       done.future.futureValue // would throw on failure
     }
 
+    "terminate actor graph interpreter when initial shell completes without subfused interpreters" in {
+      val mat = Materializer(system).asInstanceOf[PhasedFusingActorMaterializer]
+
+      Source.empty[Int].to(Sink.ignore).run()(mat)
+
+      // Wait briefly for the stream to complete and the actor to stop
+      mat.supervisor.tell(StreamSupervisor.GetChildren, testActor)
+      val children = expectMsgType[StreamSupervisor.Children]
+      // After Source.empty completes, the actor should have stopped already
+      // (preStart calls context.stop(self) when activeInterpreters is empty)
+      // If children is empty, actor already stopped; otherwise watch and expect termination
+      if (children.children.nonEmpty) {
+        val interpreterRef = children.children.head
+        watch(interpreterRef)
+        expectTerminated(interpreterRef, remainingOrDefault)
+      }
+    }
+
+    "continue working with subfused interpreters after initial shell reference is released" in {
+      val mat = Materializer(system).asInstanceOf[PhasedFusingActorMaterializer]
+
+      // flatMapConcat triggers subfusing: the inner Source is materialized via
+      // SubFusingActorMaterializerImpl, which registers additional shells into
+      // the same ActorGraphInterpreter actor via registerShell.
+      val upstream = TestPublisher.probe[Int]()
+      val downstream = TestSubscriber.probe[Int]()
+
+      Source
+        .fromPublisher(upstream)
+        .flatMapConcat(i => Source.single(i * 10))
+        .to(Sink.fromSubscriber(downstream))
+        .run()(mat)
+
+      mat.supervisor.tell(StreamSupervisor.GetChildren, testActor)
+      val children = expectMsgType[StreamSupervisor.Children]
+      val interpreterRef = children.children.head
+      watch(interpreterRef)
+
+      // Process elements through subfused inner sources
+      downstream.request(3)
+      upstream.sendNext(1)
+      downstream.expectNext(10)
+      upstream.sendNext(2)
+      downstream.expectNext(20)
+      upstream.sendNext(3)
+      downstream.expectNext(30)
+
+      // Complete the upstream: initial shell completes, subfused shells finish
+      upstream.sendComplete()
+      downstream.expectComplete()
+
+      // Actor should terminate: all shells (initial + subfused) have completed
+      expectTerminated(interpreterRef, remainingOrDefault)
+    }
+
   }
 }

stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala

@@ -894,7 +894,7 @@ import org.reactivestreams.Subscription
 /**
  * INTERNAL API
  */
-@InternalApi private[pekko] final class ActorGraphInterpreter(_initial: GraphInterpreterShell)
+@InternalApi private[pekko] final class ActorGraphInterpreter(private var _initial: GraphInterpreterShell)
     extends Actor
     with ActorLogging {
   import ActorGraphInterpreter._
@@ -907,7 +907,7 @@ import org.reactivestreams.Subscription
     try {
       currentLimit = shell.init(self, subFusingMaterializerImpl, enqueueToShortCircuit, currentLimit)
       if (GraphInterpreter.Debug)
-        println(s"registering new shell in ${_initial}\n  ${shell.toString.replace("\n", "\n  ")}")
+        println(s"registering new shell in ${shell}\n  ${shell.toString.replace("\n", "\n  ")}")
       if (shell.isTerminated) false
       else {
         activeInterpreters += shell
@@ -955,6 +955,9 @@ import org.reactivestreams.Subscription
 
   override def preStart(): Unit = {
     tryInit(_initial)
+    // Release reference to initial shell to avoid keeping it alive after it is shut down
+    // when the actor is still alive hosting other subfused interpreters
+    _initial = null
     if (activeInterpreters.isEmpty) context.stop(self)
     else if (shortCircuitBuffer ne null) shortCircuitBatch()
   }