(pekko) branch main updated: test: fix flaky FusingSpec activeStage assertion (#3496)

[fa...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git


The following commit(s) were added to refs/heads/main by this push:
     new 8181d87721 test: fix flaky FusingSpec activeStage assertion (#3496)
8181d87721 is described below

commit 8181d877216f8135d00c3eca6a2f1e9ad0d861df
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Aug 31 17:51:43 2026 +0100

    test: fix flaky FusingSpec activeStage assertion (#3496)
    
    Motivation:
    FusingSpec "leave activeStage cleared when a stage actor callback completes its
    stage" (added in #3461) intermittently failed observing the downstream IgnoreSink
    as activeStage. The stage the test guards against leaking - the async source - was
    cleared correctly; the sink was still being finalized. The completion future the
    test awaits (Sink.ignore's materialized value) is completed from within the sink's
    own onUpstreamFinish, while activeStage still references the sink, before the
    interpreter runs releaseStage to clear it. The test thread then read the
    non-volatile activeStage in a race with that clear.
    
    Modification:
    Poll the final assertion with awaitAssert so it settles once the interpreter goes
    idle, instead of reading activeStage once and racing the clear. Rename the local
    to avoid shadowing GraphStageLogic.interpreter in the inner stage.
    
    Result:
    The test asserts the same invariant without the timing race.
    
    Tests:
    - sbt "stream-tests/testOnly org.apache.pekko.stream.FusingSpec" - 24 passed
    
    References:
    Fixes #3494
---
 .../src/test/scala/org/apache/pekko/stream/FusingSpec.scala         | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
index e5ada5fedf..d54d09cd5b 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/FusingSpec.scala
@@ -377,7 +377,11 @@ class FusingSpec extends StreamSpec {
       Await.result(ready.future, 3.seconds)
       Await.result(stageActorFuture, 3.seconds) ! "complete"
       Await.result(done, 3.seconds) should ===(Done)
-      Await.result(interpreterPromise.future, 3.seconds).activeStage should be(null)
+      val completedInterpreter = Await.result(interpreterPromise.future, 3.seconds)
+      // `done` completes from within the downstream sink's own processing, so the interpreter
+      // may still be finalizing that stage (and briefly hold it as `activeStage`) when this
+      // thread wakes. Poll until it has settled rather than reading once and racing the clear.
+      awaitAssert(completedInterpreter.activeStage should be(null))
     }
 
     "stop a lazy stage actor dispatch after its handler fails" in {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org