This is an automated email from the ASF dual-hosted git repository.
He-Pin 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 1ba7efbe86 fix: initialize timer stage deadlines in preStart instead
of at construction (#3111)
1ba7efbe86 is described below
commit 1ba7efbe86161efaba4faac82fcab0e8b247fc2b
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 23 23:18:45 2026 +0800
fix: initialize timer stage deadlines in preStart instead of at
construction (#3111)
* fix: initialize timer stage deadlines in preStart instead of at
construction
Motivation:
IdleInject, Idle, BackpressureTimeout, and IdleTimeoutBidi all initialized
`nextDeadline` at GraphStageLogic construction time
(`System.nanoTime + timeout.toNanos`), not in preStart(). If materialization
takes longer than the configured timeout, the first timer check triggers a
spurious timeout even though no actual idle period has occurred.
Modification:
Move `nextDeadline` initialization from construction time to preStart() in
all
four timer stages: IdleInject, Idle, BackpressureTimeout, and
IdleTimeoutBidi.
The deadline is now set to `System.nanoTime + timeout.toNanos` inside
preStart(),
ensuring the timeout window starts when the stage actually begins
processing.
Result:
Timer stages no longer risk spurious timeouts from delayed materialization.
The deadline is always measured from the actual stream start time.
Tests:
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.scaladsl.FlowIdleInjectSpec" — 10/10 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.impl.TimeoutsSpec" — 25/25 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.javadsl.SourceTest" — 109/109 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.javadsl.FlowTest" — 88/88 passed
References:
Fixes #3105
* fix: improve timer stage fix per review feedback
- Use Long.MaxValue instead of 0L as fail-safe initial value for
nextDeadline, preventing spurious timeouts if preStart() is skipped
- Add concurrent stress tests for IdleInject and IdleTimeout that
exercise delayed materialization scenarios
- Increase FlowIdleInjectSpec timing margin from 100ms to 200ms to
reduce CI flakiness
References: Fixes #3105
* style: apply scalafmt to TimeoutsSpec
---
.../apache/pekko/stream/impl/TimeoutsSpec.scala | 93 ++++++++++++++++++++++
.../pekko/stream/scaladsl/FlowIdleInjectSpec.scala | 42 ++++++++++
.../org/apache/pekko/stream/impl/Timers.scala | 25 ++++--
3 files changed, 152 insertions(+), 8 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TimeoutsSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TimeoutsSpec.scala
index a536c83199..3e2ac42cc9 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TimeoutsSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/TimeoutsSpec.scala
@@ -131,6 +131,49 @@ class TimeoutsSpec extends StreamSpec {
ex.getMessage should ===("No elements passed in the last 1 second.")
}
+ "not fail before timeout elapses from stream start" in {
+ val upstreamProbe = TestPublisher.probe[Int]()
+ val downstreamProbe = TestSubscriber.probe[Int]()
+ val timeout = 1.second
+
+
Source.fromPublisher(upstreamProbe).idleTimeout(timeout).runWith(Sink.fromSubscriber(downstreamProbe))
+
+ // After materialization, the deadline should be set relative to
preStart,
+ // not construction time. No timeout error should occur within the
timeout period.
+ downstreamProbe.request(1)
+ downstreamProbe.expectNoMessage(timeout - 200.millis)
+
+ // Send an element to keep the stream alive, then verify no premature
timeout
+ upstreamProbe.sendNext(42)
+ downstreamProbe.expectNext(42)
+
+ upstreamProbe.sendComplete()
+ downstreamProbe.expectComplete()
+ }
+
+ "not fail prematurely when materialization is delayed under load" in {
+ import system.dispatcher
+ val timeout = 300.millis
+ val streamCount = 30
+
+ // Run many streams concurrently to create materializer scheduling
pressure.
+ // Source.maybe never emits, so the idle timeout should fire after
~timeout.
+ val futures = (1 to streamCount).map { _ =>
+ Source.maybe[Int]
+ .idleTimeout(timeout)
+ .recover { case _: StreamIdleTimeoutException => -1 }
+ .takeWithin(timeout * 4)
+ .runWith(Sink.headOption)
+ }
+
+ val results = Await.result(Future.sequence(futures), 10.seconds)
+ // Each stream should get -1 (idle timeout after proper wait) or None
(takeWithin)
+ // It should NOT fail immediately upon materialization
+ results.foreach { result =>
+ result should (be(Some(-1)).or(be(None)))
+ }
+ }
+
}
"BackpressureTimeout" must {
@@ -232,6 +275,23 @@ class TimeoutsSpec extends StreamSpec {
subscriber.expectComplete()
}
+ "not fail before timeout elapses from stream start" in {
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+ val timeout = 1.second
+
+
Source.fromPublisher(publisher).backpressureTimeout(timeout).runWith(Sink.fromSubscriber(subscriber))
+
+ // After materialization, the deadline should be set relative to
preStart,
+ // not construction time. No timeout error should occur within the
timeout period
+ // even without any demand from downstream.
+ subscriber.ensureSubscription()
+ subscriber.expectNoMessage(timeout - 200.millis)
+
+ publisher.sendComplete()
+ subscriber.expectComplete()
+ }
+
}
"IdleTimeoutBidi" must {
@@ -348,6 +408,39 @@ class TimeoutsSpec extends StreamSpec {
downWrite.expectCancellation()
}
+ "not fail before timeout elapses from stream start" in {
+ val upWrite = TestPublisher.probe[String]()
+ val upRead = TestSubscriber.probe[Int]()
+ val downWrite = TestPublisher.probe[Int]()
+ val downRead = TestSubscriber.probe[String]()
+ val timeout = 1.second
+
+ RunnableGraph
+ .fromGraph(GraphDSL.create() { implicit b =>
+ import GraphDSL.Implicits._
+ val timeoutStage = b.add(BidiFlow.bidirectionalIdleTimeout[String,
Int](timeout))
+ Source.fromPublisher(upWrite) ~> timeoutStage.in1
+ timeoutStage.out1 ~> Sink.fromSubscriber(downRead)
+ Sink.fromSubscriber(upRead) <~ timeoutStage.out2
+ timeoutStage.in2 <~ Source.fromPublisher(downWrite)
+ ClosedShape
+ })
+ .run()
+
+ upRead.request(1)
+ downRead.request(1)
+
+ // After materialization, the deadline should be set relative to
preStart,
+ // not construction time. No timeout error should occur within the
timeout period.
+ upRead.expectNoMessage(timeout - 200.millis)
+ downRead.expectNoMessage(0.millis)
+
+ upWrite.sendComplete()
+ downWrite.sendComplete()
+ upRead.expectComplete()
+ downRead.expectComplete()
+ }
+
}
"Subscription timeouts" must {
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIdleInjectSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIdleInjectSpec.scala
index ac100265b5..5a3b937e99 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIdleInjectSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowIdleInjectSpec.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.stream.scaladsl
import scala.concurrent.Await
+import scala.concurrent.Future
import scala.concurrent.duration._
import org.apache.pekko
@@ -161,6 +162,47 @@ class FlowIdleInjectSpec extends StreamSpec("""
downstream.expectNext(0)
}
+ "not inject idle element before timeout elapses from stream start" in {
+ val upstream = TestPublisher.probe[Int]()
+ val downstream = TestSubscriber.probe[Int]()
+ val timeout = 500.millis
+
+ Source.fromPublisher(upstream).keepAlive(timeout, () =>
0).runWith(Sink.fromSubscriber(downstream))
+
+ downstream.ensureSubscription()
+ // No idle element should appear before the timeout period, even though
+ // the stage is constructed before preStart runs.
+ downstream.expectNoMessage(timeout - 200.millis)
+
+ // After timeout, the injected element should arrive
+ downstream.request(1)
+ downstream.expectNext(0)
+
+ upstream.sendComplete()
+ downstream.expectComplete()
+ }
+
+ "not inject prematurely when materialization is delayed under load" in {
+ import system.dispatcher
+ val timeout = 300.millis
+ val streamCount = 30
+
+ // Run many streams concurrently to create materializer scheduling
pressure,
+ // increasing the chance of delay between createLogic and preStart.
+ val futures = (1 to streamCount).map { _ =>
+ Source.maybe[Int]
+ .keepAlive(timeout, () => 0)
+ .takeWithin(timeout * 4)
+ .runWith(Sink.headOption)
+ }
+
+ val results = Await.result(Future.sequence(futures), 10.seconds)
+ // Each stream should receive the injected element 0 after ~timeout
+ results.foreach { result =>
+ result shouldBe Some(0)
+ }
+ }
+
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Timers.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/Timers.scala
index cdbc42828f..f644a958e7 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/Timers.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Timers.scala
@@ -105,7 +105,7 @@ import pekko.stream.stage._
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
=
new TimerGraphStageLogic(shape) with InHandler with OutHandler {
- private var nextDeadline: Long = System.nanoTime + timeout.toNanos
+ private var nextDeadline: Long = Long.MaxValue
setHandlers(in, out, this)
@@ -120,8 +120,10 @@ import pekko.stream.stage._
if (nextDeadline - System.nanoTime < 0)
failStage(new StreamIdleTimeoutException(s"No elements passed in
the last ${timeout.toCoarsest}."))
- override def preStart(): Unit =
+ override def preStart(): Unit = {
+ nextDeadline = System.nanoTime + timeout.toNanos
scheduleWithFixedDelay(GraphStageLogicTimer,
timeoutCheckInterval(timeout), timeoutCheckInterval(timeout))
+ }
}
override def toString = "IdleTimeout"
@@ -133,7 +135,7 @@ import pekko.stream.stage._
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
=
new TimerGraphStageLogic(shape) with InHandler with OutHandler {
- private var nextDeadline: Long = System.nanoTime + timeout.toNanos
+ private var nextDeadline: Long = Long.MaxValue
private var waitingDemand: Boolean = true
setHandlers(in, out, this)
@@ -153,8 +155,10 @@ import pekko.stream.stage._
if (waitingDemand && (nextDeadline - System.nanoTime < 0))
failStage(new BackpressureTimeoutException(s"No demand signalled
in the last ${timeout.toCoarsest}."))
- override def preStart(): Unit =
+ override def preStart(): Unit = {
+ nextDeadline = System.nanoTime + timeout.toNanos
scheduleWithFixedDelay(GraphStageLogicTimer,
timeoutCheckInterval(timeout), timeoutCheckInterval(timeout))
+ }
}
override def toString = "BackpressureTimeout"
@@ -171,7 +175,7 @@ import pekko.stream.stage._
override def initialAttributes = DefaultAttributes.idleTimeoutBidi
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
= new TimerGraphStageLogic(shape) {
- private var nextDeadline: Long = System.nanoTime + timeout.toNanos
+ private var nextDeadline: Long = Long.MaxValue
setHandlers(in1, out1, new IdleBidiHandler(in1, out1))
setHandlers(in2, out2, new IdleBidiHandler(in2, out2))
@@ -182,8 +186,10 @@ import pekko.stream.stage._
if (nextDeadline - System.nanoTime < 0)
failStage(new StreamIdleTimeoutException(s"No elements passed in the
last ${timeout.toCoarsest}."))
- override def preStart(): Unit =
+ override def preStart(): Unit = {
+ nextDeadline = System.nanoTime + timeout.toNanos
scheduleWithFixedDelay(GraphStageLogicTimer,
timeoutCheckInterval(timeout), timeoutCheckInterval(timeout))
+ }
class IdleBidiHandler[P](in: Inlet[P], out: Outlet[P]) extends InHandler
with OutHandler {
override def onPush(): Unit = {
@@ -240,13 +246,16 @@ import pekko.stream.stage._
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
=
new TimerGraphStageLogic(shape) with StageLogging with InHandler with
OutHandler {
- private var nextDeadline: Long = System.nanoTime + timeout.toNanos
+ private var nextDeadline: Long = Long.MaxValue
private val contextPropagation = ContextPropagation()
setHandlers(in, out, this)
// Prefetching to ensure priority of actual upstream elements
- override def preStart(): Unit = pull(in)
+ override def preStart(): Unit = {
+ nextDeadline = System.nanoTime + timeout.toNanos
+ pull(in)
+ }
override def onPush(): Unit = {
nextDeadline = System.nanoTime + timeout.toNanos
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
