Enable WAL mode on Sqlite

eclair-core/pom.xml

@@ -224,7 +224,7 @@
         <dependency>
             <groupId>org.xerial</groupId>
             <artifactId>sqlite-jdbc</artifactId>
-            <version>3.27.2.1</version>
+            <version>3.34.0</version>
         </dependency>
         <dependency>
             <groupId>org.postgresql</groupId>

eclair-core/src/main/scala/fr/acinq/eclair/Setup.scala

@@ -92,6 +92,7 @@ class Setup(val datadir: File,
   val Seeds(nodeSeed, channelSeed) = seeds_opt.getOrElse(NodeParams.getSeeds(datadir))
   val chain = config.getString("chain")
   val chaindir = new File(datadir, chain)
+  chaindir.mkdirs()
   val nodeKeyManager = new LocalNodeKeyManager(nodeSeed, NodeParams.hashFromChain(chain))
   val channelKeyManager = new LocalChannelKeyManager(channelSeed, NodeParams.hashFromChain(chain))
   val instanceId = UUID.randomUUID()

eclair-core/src/main/scala/fr/acinq/eclair/db/Databases.scala

@@ -27,7 +27,7 @@ import grizzled.slf4j.Logging
 
 import java.io.File
 import java.nio.file._
-import java.sql.{Connection, DriverManager}
+import java.sql.Connection
 import java.util.UUID
 import scala.concurrent.duration._
 
@@ -183,28 +183,16 @@ object Databases extends Logging {
    * Given a parent folder it creates or loads all the databases from a JDBC connection
    */
   def sqlite(dbdir: File): SqliteDatabases = {
-    dbdir.mkdir()
-    var sqliteEclair: Connection = null
-    var sqliteNetwork: Connection = null
-    var sqliteAudit: Connection = null
-    try {
-      sqliteEclair = DriverManager.getConnection(s"jdbc:sqlite:${new File(dbdir, "eclair.sqlite")}")
-      sqliteNetwork = DriverManager.getConnection(s"jdbc:sqlite:${new File(dbdir, "network.sqlite")}")
-      sqliteAudit = DriverManager.getConnection(s"jdbc:sqlite:${new File(dbdir, "audit.sqlite")}")
-      SqliteUtils.obtainExclusiveLock(sqliteEclair) // there should only be one process writing to this file
-      logger.info("successful lock on eclair.sqlite")
-      SqliteDatabases(sqliteAudit, sqliteNetwork, sqliteEclair)
-    } catch {
-      case t: Throwable =>
-        logger.error("could not create connection to sqlite databases: ", t)
-        if (sqliteEclair != null) sqliteEclair.close()
-        if (sqliteNetwork != null) sqliteNetwork.close()
-        if (sqliteAudit != null) sqliteAudit.close()
-        throw t
-    }
+    dbdir.mkdirs()
+    SqliteDatabases(
+      eclairJdbc = SqliteUtils.openSqliteFile(dbdir, "eclair.sqlite", exclusiveLock = true, journalMode = "wal", syncFlag = "full"), // there should only be one process writing to this file
+      networkJdbc = SqliteUtils.openSqliteFile(dbdir, "network.sqlite", exclusiveLock = false, journalMode = "wal", syncFlag = "normal"), // we don't need strong durability guarantees on the network db
+      auditJdbc = SqliteUtils.openSqliteFile(dbdir, "audit.sqlite", exclusiveLock = false, journalMode = "wal", syncFlag = "full")
+    )
   }
 
   def postgres(dbConfig: Config, instanceId: UUID, dbdir: File, lockExceptionHandler: LockFailureHandler = LockFailureHandler.logAndStop)(implicit system: ActorSystem): PostgresDatabases = {
+    dbdir.mkdirs()
     val database = dbConfig.getString("postgres.database")
     val host = dbConfig.getString("postgres.host")
     val port = dbConfig.getInt("postgres.port")

eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteUtils.scala

@@ -17,10 +17,32 @@
 package fr.acinq.eclair.db.sqlite
 
 import fr.acinq.eclair.db.jdbc.JdbcUtils
+import grizzled.slf4j.Logging
 
-import java.sql.Connection
+import java.io.File
+import java.sql.{Connection, DriverManager}
 
-object SqliteUtils extends JdbcUtils {
+object SqliteUtils extends JdbcUtils with Logging {
+
+  def openSqliteFile(directory: File, filename: String, exclusiveLock: Boolean, journalMode: String, syncFlag: String): Connection = {
+    var sqlite: Connection = null
+    try {
+      sqlite = DriverManager.getConnection(s"jdbc:sqlite:${new File(directory, filename)}")
+      if (exclusiveLock) {
+        obtainExclusiveLock(sqlite)
+      }
+      setJournalMode(sqlite, journalMode)
+      setSynchronousFlag(sqlite, syncFlag)
+      sqlite
+    } catch {
+      case t: Throwable =>
+        logger.error("could not create connection to sqlite databases: ", t)
+        if (sqlite != null) {
+          sqlite.close()
+        }
+        throw t
+    }
+  }
 
   /**
    * Obtain an exclusive lock on a sqlite database. This is useful when we want to make sure that only one process
@@ -29,11 +51,33 @@ object SqliteUtils extends JdbcUtils {
    * The lock will be kept until the database is closed, or if the locking mode is explicitly reset.
    */
   def obtainExclusiveLock(sqlite: Connection): Unit = synchronized {
-    val statement = sqlite.createStatement()
-    statement.execute("PRAGMA locking_mode = EXCLUSIVE")
-    // we have to make a write to actually obtain the lock
-    statement.executeUpdate("CREATE TABLE IF NOT EXISTS dummy_table_for_locking (a INTEGER NOT NULL)")
-    statement.executeUpdate("INSERT INTO dummy_table_for_locking VALUES (42)")
+    using(sqlite.createStatement()) { statement =>
+      statement.execute("PRAGMA locking_mode = EXCLUSIVE")
+      // we have to make a write to actually obtain the lock
+      statement.executeUpdate("CREATE TABLE IF NOT EXISTS dummy_table_for_locking (a INTEGER NOT NULL)")
+      statement.executeUpdate("INSERT INTO dummy_table_for_locking VALUES (42)")
+    }
+  }
+
+  /**
+   * See https://www.sqlite.org/pragma.html#pragma_journal_mode
+   */
+  def setJournalMode(sqlite: Connection, mode: String): Unit = {
+    using(sqlite.createStatement()) { statement =>
+      val res = statement.executeQuery(s"PRAGMA journal_mode=$mode")
+      res.next()
+      val currentMode = res.getString(1)
+      assert(currentMode == mode, s"couldn't activate mode=$mode")
+    }
+  }
+
+  /**
+   * See https://www.sqlite.org/pragma.html#pragma_synchronous
+   */
+  def setSynchronousFlag(sqlite: Connection, flag: String): Unit = {
+    using(sqlite.createStatement()) { statement =>
+      statement.executeUpdate(s"PRAGMA synchronous=$flag")
+    }
   }
 
 }

eclair-core/src/test/scala/fr/acinq/eclair/db/ChannelsDbSpec.scala

@@ -105,7 +105,7 @@ class ChannelsDbSpec extends AnyFunSuite {
         Future(db.updateChannelMeta(channelId, ChannelEvent.EventType.PaymentSent))
       }
       val res = Future.sequence(futures)
-      Await.result(res, 60 seconds)
+      Await.result(res, 5 minutes)
     }
   }
 

eclair-core/src/test/scala/fr/acinq/eclair/integration/PerformanceIntegrationSpec.scala

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2019 ACINQ SAS
+ *
+ * Licensed 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 fr.acinq.eclair.integration
+
+import akka.testkit.TestProbe
+import com.typesafe.config.ConfigFactory
+import fr.acinq.bitcoin.SatoshiLong
+import fr.acinq.eclair.MilliSatoshiLong
+import fr.acinq.eclair.channel._
+import fr.acinq.eclair.payment._
+import fr.acinq.eclair.payment.receive.MultiPartHandler.ReceivePayment
+import fr.acinq.eclair.payment.send.MultiPartPaymentLifecycle.PreimageReceived
+import fr.acinq.eclair.payment.send.PaymentInitiator
+import fr.acinq.eclair.router.Router
+import org.scalatest.Ignore
+
+import java.util.UUID
+import java.util.concurrent.Executors
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.jdk.CollectionConverters._
+
+/**
+ * Created by PM on 12/07/2021.
+ */
+
+@Ignore
+class PerformanceIntegrationSpec extends IntegrationSpec {
+
+  test("start eclair nodes") {
+    val commonPerfTestConfig = ConfigFactory.parseMap(Map(
+      "eclair.max-funding-satoshis" -> 100_000_000,
+      "eclair.max-accepted-htlcs" -> Channel.MAX_ACCEPTED_HTLCS,
+      "eclair.file-backup.enabled" -> false,
+    ).asJava)
+
+    instantiateEclairNode("A", ConfigFactory.parseMap(Map("eclair.node-alias" -> "A", "eclair.server.port" -> 29730).asJava).withFallback(commonPerfTestConfig).withFallback(commonFeatures).withFallback(commonConfig)) // A's channels are private
+    instantiateEclairNode("B", ConfigFactory.parseMap(Map("eclair.node-alias" -> "B", "eclair.server.port" -> 29731).asJava).withFallback(commonPerfTestConfig).withFallback(commonFeatures).withFallback(commonConfig))
+  }
+
+  test("connect nodes") {
+    // A---B
+
+    val eventListener = TestProbe()
+    nodes.values.foreach(_.system.eventStream.subscribe(eventListener.ref, classOf[ChannelStateChanged]))
+
+    connect(nodes("A"), nodes("B"), 100_000_000 sat, 0 msat)
+
+    // confirming the funding tx
+    generateBlocks(6)
+
+    within(60 seconds) {
+      eventListener.expectMsgType[ChannelStateChanged](60 seconds).currentState == NORMAL
+    }
+  }
+
+  test("wait for channels") {
+    // Channels balance should now be available in the router
+    val sender = TestProbe()
+    awaitCond({
+      sender.send(nodes("A").router, Router.GetRoutingState)
+      val routingState = sender.expectMsgType[Router.RoutingState]
+      routingState.channels.nonEmpty
+    }, 60 seconds)
+  }
+
+  def sendPayment()(implicit ec: ExecutionContext): Future[PaymentSent] = Future {
+    val sender = TestProbe()
+    val amountMsat = 100_000.msat
+    // first we retrieve a payment hash from B
+    sender.send(nodes("B").paymentHandler, ReceivePayment(Some(amountMsat), "1 coffee"))
+    val pr = sender.expectMsgType[PaymentRequest]
+    // then we make the actual payment
+    sender.send(nodes("A").paymentInitiator, PaymentInitiator.SendPayment(amountMsat, pr, fallbackFinalExpiryDelta = finalCltvExpiryDelta, routeParams = integrationTestRouteParams, maxAttempts = 1))
+    val paymentId = sender.expectMsgType[UUID]
+    sender.expectMsgType[PreimageReceived]
+    val ps = sender.expectMsgType[PaymentSent]
+    assert(ps.id == paymentId)
+    ps
+  }
+
+  test("send a large number of htlcs A->B") {
+    val SENDERS_COUNT = 16
+    val PAYMENTS_COUNT = 3_000
+    val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(SENDERS_COUNT))
+    val start = System.currentTimeMillis()
+    val futures = (0 until PAYMENTS_COUNT).map(_ => sendPayment()(ec))
+    implicit val dummyEc: ExecutionContext = ExecutionContext.Implicits.global
+    val f = Future.sequence(futures)
+    Await.result(f, 1 hour)
+    val end = System.currentTimeMillis()
+    val duration = end - start
+    println(s"$PAYMENTS_COUNT payments in ${duration}ms ${PAYMENTS_COUNT * 1000 / duration}htlc/s (senders=$SENDERS_COUNT)")
+  }
+
+}