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-persistence-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 9a3893f6 Stop a timed-out migrator test from leaking into the next one
(#610)
9a3893f6 is described below
commit 9a3893f6a1c1380fd1239ae9860f35ed8f18cefb
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 8 08:58:55 2026 +0100
Stop a timed-out migrator test from leaking into the next one (#610)
Motivation:
The SqlServer migrator integration tests failed on main when the runner was
slow. `should migrate the event journal preserving the order of events` hit
the
10 second PatienceConfig while migrating 3000 events, and
`should migrate the event journal preserving tags` then failed with
`262 was not equal to 0` on its pre-migration count.
The second failure is fallout from the first. `withActorSystem` called
`system.terminate()` after the test body instead of in a `finally`, so a
failing
body left the actor system running. Its in-flight migration stream kept
writing
while the next test recreated the tables in `beforeEach`, and those 262 rows
landed in the fresh table.
The 10 second budget was also too tight. The last green run of the same job
spent 7.3 seconds in that phase, so a moderately slower runner is enough to
cross it.
Wrapping the migration in `eventually` does not help either: the first
`futureValue` consumes the whole patience window, so there is never a second
attempt, and a retry would race with the migration that is still in flight.
Modification:
- Terminate the actor system in a `finally` block in both `withActorSystem`
and
`withLegacyActorSystem`.
- Raise the migrator `PatienceConfig` timeout from 10 to 30 seconds.
- Call the migration once in `JournalMigratorTest` and
`SnapshotMigratorTest`,
keeping `eventually` only around the row-count assertion that follows it.
- Add `MigratorSpecFixtureTest` covering the fixture lifecycle contract.
Result:
A slow migration no longer corrupts the following test, and a migration is
never run concurrently with itself.
Tests:
- sbt "migratorIntegration/Test/testOnly
org.apache.pekko.persistence.jdbc.migrator.H2JournalMigratorTest
org.apache.pekko.persistence.jdbc.migrator.H2SnapshotMigratorTest
org.apache.pekko.persistence.jdbc.migrator.MigratorSpecFixtureTest" - 6
succeeded, 0 failed
- MigratorSpecFixtureTest with the `finally` reverted - both tests fail,
confirming the fixture leak
- sbt "migratorIntegration/scalafmtAll" and
"migratorIntegration/headerCreateAll" - run, no further changes
- The SqlServer, Postgres, MySQL and Oracle migrator tests need running
database servers and were left to CI
References:
Refs
https://github.com/apache/pekko-persistence-jdbc/actions/runs/34093026922/job/101650378069,
Refs #192
---
.../jdbc/migrator/JournalMigratorTest.scala | 9 ++--
.../persistence/jdbc/migrator/MigratorSpec.scala | 12 ++---
.../jdbc/migrator/MigratorSpecFixtureTest.scala | 51 ++++++++++++++++++++++
.../jdbc/migrator/SnapshotMigratorTest.scala | 5 ++-
4 files changed, 67 insertions(+), 10 deletions(-)
diff --git
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/JournalMigratorTest.scala
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/JournalMigratorTest.scala
index ebe6e154..baaedeb1 100644
---
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/JournalMigratorTest.scala
+++
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/JournalMigratorTest.scala
@@ -43,8 +43,9 @@ abstract class JournalMigratorTest(configName: String)
extends MigratorSpec(conf
withActorSystem { implicit systemNew =>
withReadJournal { implicit readJournal =>
countJournal().futureValue shouldBe 0 // before migration
+ // migrate exactly once: retrying while a migration is still in flight
duplicates rows
+ JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
eventually {
- JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
countJournal().futureValue shouldBe 9 // after migration
}
withTestActors() { (actorB1, actorB2, actorB3) =>
@@ -79,8 +80,9 @@ abstract class JournalMigratorTest(configName: String)
extends MigratorSpec(conf
withActorSystem { implicit systemNew =>
withReadJournal { implicit readJournal =>
countJournal().futureValue shouldBe 0 // before migration
+ // migrate exactly once: retrying while a migration is still in flight
duplicates rows
+ JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
eventually {
- JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
countJournal().futureValue shouldBe 3000 // after migration
}
val allEvents: Seq[Seq[AccountEvent]] = events().futureValue
@@ -117,8 +119,9 @@ abstract class JournalMigratorTest(configName: String)
extends MigratorSpec(conf
withActorSystem { implicit systemNew =>
withReadJournal { implicit readJournal =>
countJournal().futureValue shouldBe 0 // before migration
+ // migrate exactly once: retrying while a migration is still in flight
duplicates rows
+ JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
eventually {
- JournalMigrator(SlickDatabase.profile(config,
"slick")).migrate().futureValue shouldBe Done
countJournal().futureValue shouldBe 3000 // after migration
}
val evenEvents: Seq[AccountEvent] =
eventsByTag(MigratorSpec.Even).futureValue
diff --git
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpec.scala
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpec.scala
index 6af87016..91a9d2b1 100644
---
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpec.scala
+++
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpec.scala
@@ -40,7 +40,7 @@ abstract class MigratorSpec(val config: Config) extends
SimpleSpec with BeforeAn
// The db is initialized in the before and after each bocks
var dbOpt: Option[Database] = None
- implicit val pc: PatienceConfig = PatienceConfig(timeout = 10.seconds)
+ implicit val pc: PatienceConfig = PatienceConfig(timeout = 30.seconds)
implicit val timeout: Timeout = Timeout(1.minute)
private val logger: Logger = LoggerFactory.getLogger(this.getClass)
@@ -118,8 +118,10 @@ abstract class MigratorSpec(val config: Config) extends
SimpleSpec with BeforeAn
def withActorSystem(f: ActorSystem => Unit): Unit = {
implicit val system: ActorSystem = ActorSystem("migrator-test", config)
- f(system)
- system.terminate().futureValue
+ // terminate in a finally block: a failing test must not leak a running
migration
+ // stream into the next test, which recreates the tables in its beforeEach
+ try f(system)
+ finally system.terminate().futureValue
}
def withLegacyActorSystem(f: ActorSystem => Unit): Unit = {
@@ -137,8 +139,8 @@ abstract class MigratorSpec(val config: Config) extends
SimpleSpec with BeforeAn
}
implicit val system: ActorSystem = ActorSystem("migrator-test",
legacyDAOConfig)
- f(system)
- system.terminate().futureValue
+ try f(system)
+ finally system.terminate().futureValue
}
def withReadJournal(f: JdbcReadJournal => Unit)(implicit system:
ActorSystem): Unit = {
diff --git
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpecFixtureTest.scala
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpecFixtureTest.scala
new file mode 100644
index 00000000..01995c92
--- /dev/null
+++
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/MigratorSpecFixtureTest.scala
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.persistence.jdbc.migrator
+
+import org.apache.pekko.actor.ActorSystem
+
+/**
+ * Checks the actor system lifecycle contract of the [[MigratorSpec]]
fixtures. A failing test body must not leak a
+ * running actor system, because the migration stream it still owns keeps
writing into the journal tables that the next
+ * test recreates in its `beforeEach`.
+ *
+ * No database is required: the fixtures are exercised without starting any
persistent actor.
+ */
+class MigratorSpecFixtureTest extends MigratorSpec("h2-application.conf") {
+
+ private def assertTerminatesWhenBodyFails(fixture: (ActorSystem => Unit) =>
Unit): Unit = {
+ var captured: Option[ActorSystem] = None
+ val thrown = intercept[RuntimeException] {
+ fixture { system =>
+ captured = Some(system)
+ throw new RuntimeException("boom")
+ }
+ }
+ thrown.getMessage shouldBe "boom"
+ val system = captured.getOrElse(fail("the fixture never ran the test
body"))
+ system.whenTerminated.futureValue
+ }
+
+ it should "terminate the actor system when the test body fails" in {
+ assertTerminatesWhenBodyFails(withActorSystem)
+ }
+
+ it should "terminate the legacy actor system when the test body fails" in {
+ assertTerminatesWhenBodyFails(withLegacyActorSystem)
+ }
+}
diff --git
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/SnapshotMigratorTest.scala
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/SnapshotMigratorTest.scala
index c94e8636..a358ce56 100644
---
a/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/SnapshotMigratorTest.scala
+++
b/migrator-integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/migrator/SnapshotMigratorTest.scala
@@ -40,9 +40,10 @@ abstract class SnapshotMigratorTest(configName: String)
extends MigratorSpec(con
} // legacy persistence
withActorSystem { implicit systemNew =>
withReadJournal { implicit readJournal =>
+ countJournal().futureValue shouldBe 0 // before migration
+ // migrate exactly once: a retry would race with a migration that is
still in flight
+ SnapshotMigrator(SlickDatabase.profile(config,
"slick")).migrateAll().futureValue shouldBe Done
eventually {
- countJournal().futureValue shouldBe 0 // before migration
- SnapshotMigrator(SlickDatabase.profile(config,
"slick")).migrateAll().futureValue shouldBe Done
countJournal().futureValue shouldBe 0 // after migration
}
withTestActors() { (actorB1, actorB2, actorB3) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
