ptrdom commented on code in PR #488:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/488#discussion_r4370602211
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala:
##########
@@ -292,11 +292,22 @@ class R2dbcExecutor(
def withConnection[A](logPrefix: String)(fun: Connection => Future[A]):
Future[A] = {
getConnection(logPrefix).flatMap { connection =>
val startTime = nanoTime()
- connection.beginTransaction().asFutureDone().flatMap { _ =>
- val timeoutTask = closeCallsExceeding.map { timeout =>
- system.scheduler.scheduleOnce(timeout, () =>
closeAfterTimeout(connection))
- }
+ // scheduled before the first call on the connection, so that the
connection is released
+ // also when that call fails or never completes
+ val timeoutTask = closeCallsExceeding.map { timeout =>
+ system.scheduler.scheduleOnce(timeout, () =>
closeAfterTimeout(connection))
+ }
+
+ val transactionBegun = connection.beginTransaction().asFutureDone()
Review Comment:
If a driver or wrapper throws here synchronously instead of returning a
failed Publisher (`PooledConnection.assertNotClosed` does this, for example),
the exception escapes the `flatMap` lambda before the `failed.foreach` below is
registered. The connection is then leaked, and the watchdog only closes the
physical connection, not the pool slot. A fresh pooled connection can't be in
that state, but guarding it costs little and makes "released whatever the first
call does" hold in full:
```scala
val transactionBegun =
try connection.beginTransaction().asFutureDone()
catch { case NonFatal(exc) => Future.failed(exc) }
```
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala:
##########
@@ -340,7 +351,15 @@ class R2dbcExecutor(
system.scheduler.scheduleOnce(timeout, () =>
closeAfterTimeout(connection))
}
- connection.setAutoCommit(true).asFutureDone().flatMap { _ =>
+ val autoCommitSet = connection.setAutoCommit(true).asFutureDone()
Review Comment:
Same as above for `setAutoCommit(true)`.
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutorConnectionReleaseSpec.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.r2dbc.internal
+
+import java.time.{ Duration => JDuration }
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.function.BiFunction
+import java.util.function.Predicate
+
+import scala.collection.immutable
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorSystem
+import io.r2dbc.spi.Batch
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
+import io.r2dbc.spi.ConnectionFactoryMetadata
+import io.r2dbc.spi.ConnectionMetadata
+import io.r2dbc.spi.IsolationLevel
+import io.r2dbc.spi.R2dbcNonTransientResourceException
+import io.r2dbc.spi.Result
+import io.r2dbc.spi.Row
+import io.r2dbc.spi.RowMetadata
+import io.r2dbc.spi.Statement
+import io.r2dbc.spi.TransactionDefinition
+import io.r2dbc.spi.ValidationDepth
+import org.reactivestreams.Publisher
+import org.scalatest.wordspec.AnyWordSpecLike
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+object R2dbcExecutorConnectionReleaseSpec {
+
+ private def notImplemented(): Nothing =
+ throw new UnsupportedOperationException("not used by these tests")
+
+ /**
+ * Connection that can be told to fail the call that `R2dbcExecutor` makes
before handing the connection to the
+ * caller, and to mimic the `IllegalStateException` that
`io.r2dbc.pool.PooledConnection` throws from
+ * `assertNotClosed` when the connection has already been closed.
+ */
+ final class FakeConnection(
+ beginTransactionFailure: Option[Throwable] = None,
+ setAutoCommitFailure: Option[Throwable] = None,
+ closeThrows: Boolean = false,
+ rows: immutable.IndexedSeq[String] = Vector.empty)
+ extends Connection {
+
+ private val closeAttempts = new AtomicInteger
+
+ def closeCount: Int = closeAttempts.get
+
+ private def emptyOrFailed(failure: Option[Throwable]): Publisher[Void] =
+ failure match {
+ case Some(exc) => Mono.error[Void](exc)
+ case None => Mono.empty[Void]()
+ }
+
+ override def close(): Publisher[Void] = {
+ closeAttempts.incrementAndGet()
+ if (closeThrows) throw new IllegalStateException("Connection is closed")
+ else Mono.empty[Void]()
+ }
+
+ override def beginTransaction(): Publisher[Void] =
emptyOrFailed(beginTransactionFailure)
+
+ override def setAutoCommit(autoCommit: Boolean): Publisher[Void] =
emptyOrFailed(setAutoCommitFailure)
+
+ override def commitTransaction(): Publisher[Void] = Mono.empty[Void]()
+
+ override def rollbackTransaction(): Publisher[Void] = Mono.empty[Void]()
+
+ override def createStatement(sql: String): Statement = new
FakeStatement(rows)
+
+ override def beginTransaction(definition: TransactionDefinition):
Publisher[Void] = notImplemented()
+ override def createBatch(): Batch = notImplemented()
+ override def createSavepoint(name: String): Publisher[Void] =
notImplemented()
+ override def isAutoCommit: Boolean = notImplemented()
+ override def getMetadata: ConnectionMetadata = notImplemented()
+ override def getTransactionIsolationLevel: IsolationLevel =
notImplemented()
+ override def releaseSavepoint(name: String): Publisher[Void] =
notImplemented()
+ override def rollbackTransactionToSavepoint(name: String): Publisher[Void]
= notImplemented()
+ override def setLockWaitTimeout(timeout: JDuration): Publisher[Void] =
notImplemented()
+ override def setStatementTimeout(timeout: JDuration): Publisher[Void] =
notImplemented()
+ override def setTransactionIsolationLevel(isolationLevel: IsolationLevel):
Publisher[Void] = notImplemented()
+ override def validate(depth: ValidationDepth):
Publisher[java.lang.Boolean] = notImplemented()
+ }
+
+ final class FakeStatement(rows: immutable.IndexedSeq[String]) extends
Statement {
+ override def execute(): Publisher[? <: Result] = Mono.just(new
FakeResult(rows))
+
+ override def add(): Statement = notImplemented()
+ override def bind(index: Int, value: Any): Statement = notImplemented()
+ override def bind(name: String, value: Any): Statement = notImplemented()
+ override def bindNull(index: Int, `type`: Class[?]): Statement =
notImplemented()
+ override def bindNull(name: String, `type`: Class[?]): Statement =
notImplemented()
+ }
+
+ final class FakeResult(rows: immutable.IndexedSeq[String]) extends Result {
+ override def map[T](mappingFunction: BiFunction[Row, RowMetadata, ? <:
T]): Publisher[T] =
+ Flux.fromIterable(rows.map(row => mappingFunction.apply(new
FakeRow(row), null): T).asJava)
+
+ override def getRowsUpdated: Publisher[java.lang.Long] =
Mono.just(java.lang.Long.valueOf(rows.size.toLong))
+
+ override def filter(filter: Predicate[Result.Segment]): Result =
notImplemented()
+ override def flatMap[T](
+ mappingFunction: java.util.function.Function[Result.Segment, ? <:
Publisher[? <: T]]): Publisher[T] =
+ notImplemented()
+ }
+
+ final class FakeRow(value: String) extends Row {
+ override def get[T](name: String, `type`: Class[T]): T = `type`.cast(value)
+
+ override def get[T](index: Int, `type`: Class[T]): T = `type`.cast(value)
+ override def getMetadata: RowMetadata = notImplemented()
+ }
+
+ final class FakeConnectionFactory(connection: Connection) extends
ConnectionFactory {
+ override def create(): Publisher[? <: Connection] = Mono.just(connection)
+
+ override def getMetadata: ConnectionFactoryMetadata = new
ConnectionFactoryMetadata {
+ override def getName: String = "fake"
+ }
+ }
+}
+
+class R2dbcExecutorConnectionReleaseSpec extends ScalaTestWithActorTestKit
with AnyWordSpecLike with LogCapturing {
+ import R2dbcExecutorConnectionReleaseSpec._
+
+ private val log =
LoggerFactory.getLogger(classOf[R2dbcExecutorConnectionReleaseSpec])
+
+ private implicit val ec: ExecutionContext = testKit.system.executionContext
+ private implicit val typedSystem: ActorSystem[?] = testKit.system
+
+ // no close-calls-exceeding watchdog, so a connection that is not released
by the executor itself
+ // stays leaked for the duration of the test
+ private def executorFor(connection: Connection): R2dbcExecutor =
Review Comment:
Suggestion: add a test for the watchdog reordering in `withConnection`. On
`main` it fails (no timer is scheduled until `BEGIN` completes); on this branch
it passes:
```scala
"close the connection when beginTransaction never completes" in {
val connection = new FakeConnection(beginTransactionNever = true) //
returns Mono.never()
val executor = new R2dbcExecutor(new FakeConnectionFactory(connection),
log,
logDbCallsExceeding = -1.millis, closeCallsExceeding = Some(100.millis))
executor.withConnection("test") { _ => fail("should not be called") }
eventually { connection.closeCount should be >= 1 }
}
```
`FakeConnection` isn't `Wrapped`, so `closeAfterTimeout` closes it directly.
##########
core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala:
##########
@@ -375,21 +392,20 @@ class R2dbcExecutor(
}
}
+ private def closeConnection(connection: Connection): Future[Done] = {
Review Comment:
Nit: the catcher now swallows `IllegalStateException` silently at five call
sites. A `log.debug` in `ignoreConnectionClosedException` would make a real
double close easier to diagnose. The catcher's existing comment ("if connection
closed after timeout") also doesn't match how `closeAfterTimeout` works with
the pool (it closes the unwrapped connection). It predates this PR, but it
might be worth correcting while you're here.
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutorConnectionReleaseSpec.scala:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.r2dbc.internal
+
+import java.time.{ Duration => JDuration }
+import java.util.concurrent.atomic.AtomicInteger
+import java.util.function.BiFunction
+import java.util.function.Predicate
+
+import scala.collection.immutable
+import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorSystem
+import io.r2dbc.spi.Batch
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
+import io.r2dbc.spi.ConnectionFactoryMetadata
+import io.r2dbc.spi.ConnectionMetadata
+import io.r2dbc.spi.IsolationLevel
+import io.r2dbc.spi.R2dbcNonTransientResourceException
+import io.r2dbc.spi.Result
+import io.r2dbc.spi.Row
+import io.r2dbc.spi.RowMetadata
+import io.r2dbc.spi.Statement
+import io.r2dbc.spi.TransactionDefinition
+import io.r2dbc.spi.ValidationDepth
+import org.reactivestreams.Publisher
+import org.scalatest.wordspec.AnyWordSpecLike
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+object R2dbcExecutorConnectionReleaseSpec {
+
+ private def notImplemented(): Nothing =
+ throw new UnsupportedOperationException("not used by these tests")
+
+ /**
+ * Connection that can be told to fail the call that `R2dbcExecutor` makes
before handing the connection to the
+ * caller, and to mimic the `IllegalStateException` that
`io.r2dbc.pool.PooledConnection` throws from
+ * `assertNotClosed` when the connection has already been closed.
+ */
+ final class FakeConnection(
+ beginTransactionFailure: Option[Throwable] = None,
+ setAutoCommitFailure: Option[Throwable] = None,
+ closeThrows: Boolean = false,
+ rows: immutable.IndexedSeq[String] = Vector.empty)
+ extends Connection {
+
+ private val closeAttempts = new AtomicInteger
+
+ def closeCount: Int = closeAttempts.get
+
+ private def emptyOrFailed(failure: Option[Throwable]): Publisher[Void] =
+ failure match {
+ case Some(exc) => Mono.error[Void](exc)
+ case None => Mono.empty[Void]()
+ }
+
+ override def close(): Publisher[Void] = {
+ closeAttempts.incrementAndGet()
+ if (closeThrows) throw new IllegalStateException("Connection is closed")
+ else Mono.empty[Void]()
+ }
+
+ override def beginTransaction(): Publisher[Void] =
emptyOrFailed(beginTransactionFailure)
+
+ override def setAutoCommit(autoCommit: Boolean): Publisher[Void] =
emptyOrFailed(setAutoCommitFailure)
+
+ override def commitTransaction(): Publisher[Void] = Mono.empty[Void]()
+
+ override def rollbackTransaction(): Publisher[Void] = Mono.empty[Void]()
+
+ override def createStatement(sql: String): Statement = new
FakeStatement(rows)
+
+ override def beginTransaction(definition: TransactionDefinition):
Publisher[Void] = notImplemented()
+ override def createBatch(): Batch = notImplemented()
+ override def createSavepoint(name: String): Publisher[Void] =
notImplemented()
+ override def isAutoCommit: Boolean = notImplemented()
+ override def getMetadata: ConnectionMetadata = notImplemented()
+ override def getTransactionIsolationLevel: IsolationLevel =
notImplemented()
+ override def releaseSavepoint(name: String): Publisher[Void] =
notImplemented()
+ override def rollbackTransactionToSavepoint(name: String): Publisher[Void]
= notImplemented()
+ override def setLockWaitTimeout(timeout: JDuration): Publisher[Void] =
notImplemented()
+ override def setStatementTimeout(timeout: JDuration): Publisher[Void] =
notImplemented()
+ override def setTransactionIsolationLevel(isolationLevel: IsolationLevel):
Publisher[Void] = notImplemented()
+ override def validate(depth: ValidationDepth):
Publisher[java.lang.Boolean] = notImplemented()
+ }
+
+ final class FakeStatement(rows: immutable.IndexedSeq[String]) extends
Statement {
+ override def execute(): Publisher[? <: Result] = Mono.just(new
FakeResult(rows))
+
+ override def add(): Statement = notImplemented()
+ override def bind(index: Int, value: Any): Statement = notImplemented()
+ override def bind(name: String, value: Any): Statement = notImplemented()
+ override def bindNull(index: Int, `type`: Class[?]): Statement =
notImplemented()
+ override def bindNull(name: String, `type`: Class[?]): Statement =
notImplemented()
+ }
+
+ final class FakeResult(rows: immutable.IndexedSeq[String]) extends Result {
+ override def map[T](mappingFunction: BiFunction[Row, RowMetadata, ? <:
T]): Publisher[T] =
+ Flux.fromIterable(rows.map(row => mappingFunction.apply(new
FakeRow(row), null): T).asJava)
+
+ override def getRowsUpdated: Publisher[java.lang.Long] =
Mono.just(java.lang.Long.valueOf(rows.size.toLong))
+
+ override def filter(filter: Predicate[Result.Segment]): Result =
notImplemented()
+ override def flatMap[T](
+ mappingFunction: java.util.function.Function[Result.Segment, ? <:
Publisher[? <: T]]): Publisher[T] =
+ notImplemented()
+ }
+
+ final class FakeRow(value: String) extends Row {
+ override def get[T](name: String, `type`: Class[T]): T = `type`.cast(value)
+
+ override def get[T](index: Int, `type`: Class[T]): T = `type`.cast(value)
+ override def getMetadata: RowMetadata = notImplemented()
+ }
+
+ final class FakeConnectionFactory(connection: Connection) extends
ConnectionFactory {
+ override def create(): Publisher[? <: Connection] = Mono.just(connection)
+
+ override def getMetadata: ConnectionFactoryMetadata = new
ConnectionFactoryMetadata {
+ override def getName: String = "fake"
+ }
+ }
+}
+
+class R2dbcExecutorConnectionReleaseSpec extends ScalaTestWithActorTestKit
with AnyWordSpecLike with LogCapturing {
+ import R2dbcExecutorConnectionReleaseSpec._
+
+ private val log =
LoggerFactory.getLogger(classOf[R2dbcExecutorConnectionReleaseSpec])
+
+ private implicit val ec: ExecutionContext = testKit.system.executionContext
+ private implicit val typedSystem: ActorSystem[?] = testKit.system
+
+ // no close-calls-exceeding watchdog, so a connection that is not released
by the executor itself
+ // stays leaked for the duration of the test
+ private def executorFor(connection: Connection): R2dbcExecutor =
+ new R2dbcExecutor(new FakeConnectionFactory(connection), log,
logDbCallsExceeding = -1.millis,
+ closeCallsExceeding = None)
+
+ "R2dbcExecutor" should {
+
+ "close the connection when beginTransaction fails" in {
+ val failure = new R2dbcNonTransientResourceException("connection closed
by the server")
+ val connection = new FakeConnection(beginTransactionFailure =
Some(failure))
+
+ executorFor(connection)
+ .withConnection("test") { _ => fail("the function should not be called
when beginTransaction fails") }
+ .failed
+ .futureValue shouldBe failure
+
+ eventually {
+ connection.closeCount shouldBe 1
+ }
+ }
+
+ "close the connection when setting auto-commit fails" in {
+ val failure = new R2dbcNonTransientResourceException("connection closed
by the server")
+ val connection = new FakeConnection(setAutoCommitFailure = Some(failure))
+
+ executorFor(connection)
+ .withAutoCommitConnection("test") { _ => fail("the function should not
be called when setAutoCommit fails") }
+ .failed
+ .futureValue shouldBe failure
+
+ eventually {
+ connection.closeCount shouldBe 1
+ }
+ }
+
+ "return the selected rows when the connection was already closed by the
timeout task" in {
Review Comment:
This test runs with `closeCallsExceeding = None`, and the fake throws on its
first `close()`. So it checks "tolerates `close()` throwing
`IllegalStateException`", not a race with the timeout task. With r2dbc-pool the
timeout task doesn't put the `PooledConnection` in the closed state. Maybe
rename it to something like `"return the selected rows when close throws
IllegalStateException"`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
