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 5732f223f0 Cover ordered unstashing in durable state behavior (#3659)
5732f223f0 is described below
commit 5732f223f0fd4933466eede39b15341ec5d83c15
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Oct 23 21:21:20 2026 +0800
Cover ordered unstashing in durable state behavior (#3659)
* Cover ordered unstashing in durable state behavior
Motivation:
The durable state suite does not cover a large user stash followed by a new
command after unstashing completes.
Modification:
Stash a deterministic command sequence, observe a processing barrier,
unstash and assert every reply in order, then verify the next command succeeds.
Result:
Coverage protects ordered draining and continued processing without
sleep-based coordination.
Tests:
- Passed: `persistence-typed-tests / Test / testOnly
org.apache.pekko.persistence.typed.state.scaladsl.DurableStateRevisionSpec`.
- Passed: `checkCodeStyle`.
- Passed: `+headerCheckAll`.
- Full `validatePullRequest` is left to remote CI.
References:
- Refs [akka/akka-core#32137](https://github.com/akka/akka-core/pull/32137).
- Upstream commit:
[a95c177fe9d3dce2e9f83eb3c4cd02e963101658](https://github.com/akka/akka-core/commit/a95c177fe9d3dce2e9f83eb3c4cd02e963101658).
- The user-stash regression in `DurableStateRevisionSpec.scala` adapts the
5000-event recovery/stash/unstash scenario from Akka's
`DurableStateStashOverflowSpec.scala` in akka/akka-core#32137, retaining that
source's exact 2020-2023 copyright notice alongside the existing Pekko file
notice. CountStashed synchronization, exact sequence ordering and the
subsequent-command assertions are Pekko additions.
- Source snapshot: Akka `v2.9.0`
(`bf20e967571574994ccb3cae411401f5a9021ab9`). Its
[license](https://github.com/akka/akka-core/blob/v2.9.0/LICENSE) names
2026-10-23 as the Change Date and Apache License 2.0 as the Change License.
`LICENSE` and `NOTICE` record the source, and original copyright notices are
retained. Headers were generated/checked with the project sbt tasks.
* Keep licensing metadata outside the migration change
Motivation:
Keep PR #3659 scoped to its code or documentation change.
Modification:
Restore the root LICENSE and NOTICE files to main.
Result:
The migration diff retains its source headers without repository-level
licensing edits.
Tests:
- git diff --check passed.
- Verified every other file is unchanged; prior focused validation still
applies.
References:
Refs #3659.
* Name durable unstashing coverage for its contract #3659
Motivation:
The test demonstrates ordered unstashing and later-command progress rather
than a proven stack-overflow regression.
Modification:
Name the test for those assertions and consolidate the Lightbend notices to
cover 2009-2023.
Result:
The test states the behavior it verifies and retains full copyright-year
coverage.
Tests:
Validation used a local integration checkout with current main and the
other review fixes.
- persistence-typed-tests / Test / testOnly
org.apache.pekko.persistence.typed.state.scaladsl.DurableStateRevisionSpec: 5
tests passed.
- headerCreateAll, +headerCheckAll, native scalafmt, and git diff --check:
passed.
- Full validation is delegated to remote CI.
References:
Refs #3659.
---
.../state/scaladsl/DurableStateRevisionSpec.scala | 54 +++++++++++++++++++++-
1 file changed, 53 insertions(+), 1 deletion(-)
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/state/scaladsl/DurableStateRevisionSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/state/scaladsl/DurableStateRevisionSpec.scala
index f02d907114..99fce196bb 100644
---
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/state/scaladsl/DurableStateRevisionSpec.scala
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/state/scaladsl/DurableStateRevisionSpec.scala
@@ -8,12 +8,15 @@
*/
/*
- * Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com>
+ * Copyright (C) 2009-2023 Lightbend Inc. <https://www.lightbend.com>
*/
package org.apache.pekko.persistence.typed.state.scaladsl
+import scala.concurrent.duration._
+
import org.apache.pekko
+import pekko.Done
import pekko.actor.Dropped
import pekko.actor.testkit.typed.scaladsl.LogCapturing
import pekko.actor.testkit.typed.scaladsl.LoggingTestKit
@@ -34,6 +37,12 @@ import com.typesafe.config.ConfigFactory
object DurableStateRevisionSpec {
+ sealed trait UserStashCommand
+ case object Ready extends UserStashCommand
+ final case class Read(index: Int) extends UserStashCommand
+ case object CountStashed extends UserStashCommand
+ case object Unstash extends UserStashCommand
+
def conf: Config =
PersistenceTestKitDurableStateStorePlugin.config.withFallback(ConfigFactory.parseString(s"""
pekko.loglevel = INFO
"""))
@@ -180,4 +189,47 @@ class DurableStateRevisionSpec
}
}
+
+ "The durable state user stash" must {
+ "unstash a large read-only batch in order and process subsequent commands"
in {
+ import DurableStateRevisionSpec._
+
+ val size = 5000
+ val ready = TestProbe[Done]()
+ val counted = TestProbe[Int]()
+ val replies = TestProbe[Int]()
+
system.settings.config.getBoolean("pekko.persistence.typed.recurse-when-unstashing-read-only-commands")
shouldBe
+ false
+
+ val ref = spawn(Behaviors.setup[UserStashCommand] { _ =>
+ var stashed = 0
+ DurableStateBehavior[UserStashCommand, Boolean](
+ PersistenceId.ofUniqueId("read-only-user-stash"),
+ false,
+ (readEnabled, command) =>
+ command match {
+ case Ready => Effect.reply(ready.ref)(Done)
+ case Read(_) if !readEnabled =>
+ stashed += 1
+ Effect.stash()
+ case Read(index) => Effect.reply(replies.ref)(index)
+ case CountStashed => Effect.reply(counted.ref)(stashed)
+ case Unstash => Effect.persist(true).thenUnstashAll()
+ }).withStashCapacity(size)
+ })
+
+ // This reply proves recovery has completed before commands enter the
user stash.
+ ref ! Ready
+ ready.expectMessage(Done)
+ (1 to size).foreach(index => ref ! Read(index))
+ ref ! CountStashed
+ counted.expectMessage(size)
+
+ ref ! Unstash
+ replies.receiveMessages(size) shouldBe (1 to size)
+ ref ! Read(size + 1)
+ replies.expectMessage(size + 1)
+ replies.expectNoMessage(100.millis)
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
