Reduce pg transaction isolation (#1860)

I was able to reproduce #1856 by replaying the "concurrent channel updates" test with hardcoded additional delays in the database code. It was due to a conflict between `addOrUpdateChannel` and `updateChannelMetaTimestampColumn`. The two calls run in parallel and the latter completed before the former, causing it to fail. Reducing the isolation level makes the problem disappear. We reduce the transaction isolation level from `SERIALIZABLE` to `READ_COMMITTED`. Note that [*]: > Read Committed is the default isolation level in PostgreSQL. I'm not sure why we were using a stricter isolation level than the default one, since we only use very basic queries. Doc does say that: > This behavior makes Read Committed mode unsuitable for commands that involve complex search conditions; however, it is just right for simpler cases To make sure this didn't cause regression withe the locking mechanism, I wrote an additional test specifically on the `withLock` method. Here is what the doc says on the `INSERT ON CONFLICT DO UPDATE` statement, which we use for `addOrUpdateChannel`: > INSERT with an ON CONFLICT DO UPDATE clause behaves similarly. In Read Committed mode, each row proposed for insertion will either insert or update. Unless there are unrelated errors, one of those two outcomes is guaranteed. If a conflict originates in another transaction whose effects are not yet visible to the INSERT, the UPDATE clause will affect that row, even though possibly no version of that row is conventionally visible to the command. In the scenario described above, the `addOrUpdate` will update the row which timestamp was updated in parallel by the `updateChannelMetaTimestampColumn`, and it's exactly what we want. Fixes #1856. [*] https://www.postgresql.org/docs/13/transaction-iso.html
ACINQ · Jul 8, 2021 · 95fffe3 · 95fffe3
1 parent cea3fc0
commit 95fffe3
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 6 deletions.
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgUtils.scala b/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgUtils.scala
@@ -268,7 +268,7 @@ object PgUtils extends JdbcUtils {
 
   def inTransaction[T](f: Connection => T)(implicit dataSource: DataSource): T = {
     withConnection { connection =>
-      inTransactionInternal(IsolationLevel.TRANSACTION_SERIALIZABLE)(connection)(f)
+      inTransactionInternal(IsolationLevel.TRANSACTION_READ_COMMITTED)(connection)(f)
     }
   }
 

diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/db/PgUtilsSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/db/PgUtilsSpec.scala
@@ -2,19 +2,19 @@ package fr.acinq.eclair.db
 
 import com.opentable.db.postgres.embedded.EmbeddedPostgres
 import com.typesafe.config.{Config, ConfigFactory}
-import fr.acinq.eclair.db.pg.PgUtils.{JdbcUrlChanged, migrateTable, using}
+import fr.acinq.eclair.db.pg.PgUtils.ExtendedResultSet._
 import fr.acinq.eclair.db.pg.PgUtils.PgLock.{LockFailure, LockFailureHandler}
+import fr.acinq.eclair.db.pg.PgUtils.{JdbcUrlChanged, migrateTable, using}
 import fr.acinq.eclair.{TestKitBaseClass, TestUtils}
-import grizzled.slf4j.Logging
-import org.postgresql.PGConnection
-import org.postgresql.jdbc.PgConnection
 import grizzled.slf4j.{Logger, Logging}
+import org.postgresql.jdbc.PgConnection
+import org.postgresql.util.PGInterval
 import org.scalatest.concurrent.Eventually
 import org.scalatest.funsuite.AnyFunSuiteLike
-import fr.acinq.eclair.db.pg.PgUtils.ExtendedResultSet._
 
 import java.io.File
 import java.util.UUID
+import javax.sql.DataSource
 
 class PgUtilsSpec extends TestKitBaseClass with AnyFunSuiteLike with Eventually {
 
@@ -67,6 +67,38 @@ class PgUtilsSpec extends TestKitBaseClass with AnyFunSuiteLike with Eventually
     pg.close()
   }
 
+  test("withLock utility method") {
+    val pg = EmbeddedPostgres.start()
+    val config = PgUtilsSpec.testConfig(pg.getPort)
+    val datadir = new File(TestUtils.BUILD_DIRECTORY, s"pg_test_${UUID.randomUUID()}")
+    datadir.mkdirs()
+    val instanceId1 = UUID.randomUUID()
+    // this will lock the database for this instance id
+    val db = Databases.postgres(config, instanceId1, datadir, LockFailureHandler.logAndThrow)
+    implicit val ds: DataSource = db.dataSource
+
+    // dummy query works
+    db.lock.withLock { conn =>
+      conn.createStatement().executeQuery("SELECT 1")
+    }
+
+    intercept[LockFailureHandler.LockException] {
+      db.lock.withLock { conn =>
+        // we start with a dummy query
+        conn.createStatement().executeQuery("SELECT 1")
+        // but before we complete the query, a separate connection takes the lock
+        using(pg.getPostgresDatabase.getConnection.prepareStatement(s"UPDATE lease SET expires_at = now() + ?, instance = ? WHERE id = 1")) {
+          statement =>
+            statement.setObject(1, new PGInterval("60 seconds"))
+            statement.setString(2, UUID.randomUUID().toString)
+            statement.executeUpdate()
+        }
+      }
+    }
+
+    pg.close()
+  }
+
   test("jdbc url check") {
     val pg = EmbeddedPostgres.start()
     val config = PgUtilsSpec.testConfig(pg.getPort)