Postgresql support

eclair-core/pom.xml

@@ -194,6 +194,16 @@
             <artifactId>sqlite-jdbc</artifactId>
             <version>3.27.2.1</version>
         </dependency>
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <version>42.2.8</version>
+            <artifactId>postgresql</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP-java7</artifactId>
+            <version>2.4.13</version>
+        </dependency>
         <dependency>
             <!-- This is to get rid of '[WARNING] warning: Class javax.annotation.Nonnull not found - continuing with a stub.' compile errors -->
             <groupId>com.google.code.findbugs</groupId>

eclair-core/src/main/resources/reference.conf

@@ -164,6 +164,28 @@ eclair {
     port = 9051
     private-key-file = "tor.dat"
   }
+
+  db {
+    driver = "sqlite" // sqlite, psql
+    psql {
+      database = "eclair"
+      host = "localhost"
+      port = 5432
+      username = ""
+      password = ""
+      pool {
+        max-size = 10
+        connection-timeout = 30 seconds
+        idle-timeout = 10 minutes
+        max-life-time = 30 minutes
+      }
+    }
+    lock {
+      lease-interval = 5 minutes
+      lease-renew-interval = 1 minute
+      lock-timeout = 5 seconds
+    }
+  }
 }
 
 // do not edit or move this section

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

@@ -16,6 +16,7 @@
 
 package fr.acinq.eclair
 
+import java.lang.management.ManagementFactory
 import java.util.UUID
 
 import akka.actor.ActorRef
@@ -41,7 +42,7 @@ import scodec.bits.ByteVector
 import scala.concurrent.duration._
 import scala.concurrent.{ExecutionContext, Future}
 
-case class GetInfoResponse(nodeId: PublicKey, alias: String, chainHash: ByteVector32, blockHeight: Int, publicAddresses: Seq[NodeAddress])
+case class GetInfoResponse(nodeId: PublicKey, alias: String, chainHash: ByteVector32, blockHeight: Int, publicAddresses: Seq[NodeAddress], instanceId: String)
 
 case class AuditResponse(sent: Seq[PaymentSent], received: Seq[PaymentReceived], relayed: Seq[PaymentRelayed])
 
@@ -306,7 +307,8 @@ class EclairImpl(appKit: Kit) extends Eclair {
       alias = appKit.nodeParams.alias,
       chainHash = appKit.nodeParams.chainHash,
       blockHeight = appKit.nodeParams.currentBlockHeight.toInt,
-      publicAddresses = appKit.nodeParams.publicAddresses)
+      publicAddresses = appKit.nodeParams.publicAddresses,
+      instanceId = ManagementFactory.getRuntimeMXBean().getName())
   )
 
   override def usableBalances()(implicit timeout: Timeout): Future[Iterable[UsableBalance]] =

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

@@ -43,10 +43,10 @@ import fr.acinq.eclair.channel.Register
 import fr.acinq.eclair.crypto.LocalKeyManager
 import fr.acinq.eclair.db.{BackupHandler, Databases}
 import fr.acinq.eclair.io.{Authenticator, Server, Switchboard}
-import fr.acinq.eclair.payment.receive.PaymentHandler
-import fr.acinq.eclair.payment.send.{Autoprobe, PaymentInitiator}
 import fr.acinq.eclair.payment.Auditor
+import fr.acinq.eclair.payment.receive.PaymentHandler
 import fr.acinq.eclair.payment.relay.{CommandBuffer, Relayer}
+import fr.acinq.eclair.payment.send.{Autoprobe, PaymentInitiator}
 import fr.acinq.eclair.router._
 import fr.acinq.eclair.tor.TorProtocolHandler.OnionServiceVersion
 import fr.acinq.eclair.tor.{Controller, TorProtocolHandler}
@@ -92,32 +92,30 @@ class Setup(datadir: File,
   val chaindir = new File(datadir, chain)
   val keyManager = new LocalKeyManager(seed, NodeParams.makeChainHash(chain))
 
-  val database = db match {
-    case Some(d) => d
-    case None => Databases.sqliteJDBC(chaindir)
-  }
+  val database = initDatabase(config.getConfig("db"))
 
   /**
-   * This counter holds the current blockchain height.
-   * It is mainly used to calculate htlc expiries.
-   * The value is read by all actors, hence it needs to be thread-safe.
-   */
+    * This counter holds the current blockchain height.
+    * It is mainly used to calculate htlc expiries.
+    * The value is read by all actors, hence it needs to be thread-safe.
+    */
   val blockCount = new AtomicLong(0)
 
   /**
-   * This holds the current feerates, in satoshi-per-kilobytes.
-   * The value is read by all actors, hence it needs to be thread-safe.
-   */
+    * This holds the current feerates, in satoshi-per-kilobytes.
+    * The value is read by all actors, hence it needs to be thread-safe.
+    */
   val feeratesPerKB = new AtomicReference[FeeratesPerKB](null)
 
   /**
-   * This holds the current feerates, in satoshi-per-kw.
-   * The value is read by all actors, hence it needs to be thread-safe.
-   */
+    * This holds the current feerates, in satoshi-per-kw.
+    * The value is read by all actors, hence it needs to be thread-safe.
+    */
   val feeratesPerKw = new AtomicReference[FeeratesPerKw](null)
 
   val feeEstimator = new FeeEstimator {
     override def getFeeratePerKb(target: Int): Long = feeratesPerKB.get().feePerBlock(target)
+
     override def getFeeratePerKw(target: Int): Long = feeratesPerKw.get().feePerBlock(target)
   }
 
@@ -349,12 +347,44 @@ class Setup(datadir: File,
       None
     }
   }
+
+  private def initDatabase(dbConfig: Config): Databases = {
+    try {
+      val database = db match {
+        case Some(d) => d
+        case None =>
+          dbConfig.getString("driver") match {
+            case "sqlite" => Databases.sqliteJDBC(chaindir)
+            case "psql" => Databases.setupPsqlDatabases(dbConfig)
+            case _ => throw new RuntimeException(s"Unknown database driver `${dbConfig.getString("driver")}`")
+          }
+      }
+      val dbLockLeaseRenewInterval = dbConfig.getDuration("lock.lease-renew-interval").toSeconds.seconds
+      system.scheduler.schedule(dbLockLeaseRenewInterval, dbLockLeaseRenewInterval) {
+        try {
+          database.obtainExclusiveLock()
+        } catch {
+          case e: Throwable =>
+            logger.error("Cannot lock database. Exiting...", e)
+            sys.exit(-2)
+        }
+      }
+      database
+    } catch {
+      case e: Throwable =>
+        logger.error("Cannot initialize database. Exiting...", e)
+        sys.exit(-1)
+    }
+  }
 }
 
 // @formatter:off
 sealed trait Bitcoin
+
 case class Bitcoind(bitcoinClient: BasicBitcoinJsonRPCClient) extends Bitcoin
+
 case class Electrum(electrumClient: ActorRef) extends Bitcoin
+
 // @formatter:on
 
 case class Kit(nodeParams: NodeParams,

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

@@ -53,28 +53,26 @@ class BackupHandler private(databases: Databases, backupFile: File, backupScript
 
   def receive = {
     case persisted: ChannelPersisted =>
-      val start = System.currentTimeMillis()
-      val tmpFile = new File(backupFile.getAbsolutePath.concat(".tmp"))
-      databases.backup(tmpFile)
-      // this will throw an exception if it fails, which is possible if the backup file is not on the same filesystem
-      // as the temporary file
-      Files.move(tmpFile.toPath, backupFile.toPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE)
-      val end = System.currentTimeMillis()
+      if (databases.isBackupSupported) {
+        val start = System.currentTimeMillis()
+        databases.backup(backupFile)
+        val end = System.currentTimeMillis()
 
-      // publish a notification that we have updated our backup
-      context.system.eventStream.publish(BackupCompleted)
+        // publish a notification that we have updated our backup
+        context.system.eventStream.publish(BackupCompleted)
 
-      log.info(s"database backup triggered by channelId=${persisted.channelId} took ${end - start}ms")
+        log.info(s"database backup triggered by channelId=${persisted.channelId} took ${end - start}ms")
 
-      backupScript_opt.foreach(backupScript => {
-        Try {
-          // run the script in the current thread and wait until it terminates
-          Process(backupScript).!
-        } match {
-          case Success(exitCode) => log.info(s"backup notify script $backupScript returned $exitCode")
-          case Failure(cause) => log.warning(s"cannot start backup notify script $backupScript:  $cause")
-        }
-      })
+        backupScript_opt.foreach(backupScript => {
+          Try {
+            // run the script in the current thread and wait until it terminates
+            Process(backupScript).!
+          } match {
+            case Success(exitCode) => log.info(s"backup notify script $backupScript returned $exitCode")
+            case Failure(cause) => log.warning(s"cannot start backup notify script $backupScript:  $cause")
+          }
+        })
+      }
   }
 }
 

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

@@ -17,11 +17,18 @@
 package fr.acinq.eclair.db
 
 import java.io.File
+import java.lang.management.ManagementFactory
+import java.nio.file.{Files, StandardCopyOption}
 import java.sql.{Connection, DriverManager}
+import java.util.concurrent.TimeUnit
 
+import com.typesafe.config.Config
+import fr.acinq.eclair.db.psql._
 import fr.acinq.eclair.db.sqlite._
 import grizzled.slf4j.Logging
-import org.sqlite.SQLiteException
+import javax.sql.DataSource
+
+import scala.concurrent.duration._
 
 trait Databases {
 
@@ -38,6 +45,10 @@ trait Databases {
   val pendingRelay: PendingRelayDb
 
   def backup(file: File): Unit
+
+  val isBackupSupported: Boolean
+
+  def obtainExclusiveLock(): Unit
 }
 
 object Databases extends Logging {
@@ -57,9 +68,8 @@ object Databases extends Logging {
       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")
-      databaseByConnections(sqliteAudit, sqliteNetwork, sqliteEclair)
+      sqliteDatabaseByConnections(sqliteAudit, sqliteNetwork, sqliteEclair)
     } catch {
       case t: Throwable => {
         logger.error("could not create connection to sqlite databases: ", t)
@@ -69,23 +79,106 @@ object Databases extends Logging {
         throw t
       }
     }
+  }
 
+  def postgresJDBC(database: String = "eclair", host: String = "localhost", port: Int = 5432,
+                   username: Option[String] = None, password: Option[String] = None,
+                   poolProperties: Map[String, Long] = Map(),
+                   instanceId: String = ManagementFactory.getRuntimeMXBean().getName(),
+                   lockLeaseInterval: FiniteDuration = 5.minutes,
+                   lockTimeout: FiniteDuration = 5.seconds): Databases = {
+    try {
+      val url = s"jdbc:postgresql://${host}:${port}/${database}"
+
+      import com.zaxxer.hikari.{HikariConfig, HikariDataSource}
+
+      val config = new HikariConfig()
+      config.setJdbcUrl(url)
+      username.foreach(config.setUsername)
+      password.foreach(config.setPassword)
+      poolProperties.get("max-size").foreach(x => config.setMaximumPoolSize(x.toInt))
+      poolProperties.get("connection-timeout").foreach(config.setConnectionTimeout)
+      poolProperties.get("idle-timeout").foreach(config.setIdleTimeout)
+      poolProperties.get("max-life-time").foreach(config.setMaxLifetime)
+
+      implicit val ds: DataSource = new HikariDataSource(config)
+
+      PsqlUtils.obtainExclusiveLock(instanceId, lockLeaseInterval, lockTimeout)
+
+      new Databases {
+        override val network = new PsqlNetworkDb()
+        override val audit = new PsqlAuditDb()
+        override val channels = new PsqlChannelsDb()
+        override val peers = new PsqlPeersDb()
+        override val payments = new PsqlPaymentsDb()
+        override val pendingRelay = new PsqlPendingRelayDb()
+        override def backup(file: File): Unit = throw new RuntimeException("psql driver does not support channels backup")
+        override val isBackupSupported: Boolean = false
+        override def obtainExclusiveLock(): Unit = PsqlUtils.obtainExclusiveLock(instanceId, lockLeaseInterval, lockTimeout)
+      }
+    } catch {
+      case t: Throwable => {
+        logger.error("could not create connection to psql database: ", t)
+        throw t
+      }
+    }
   }
 
-  def databaseByConnections(auditJdbc: Connection, networkJdbc: Connection, eclairJdbc: Connection) = new Databases {
+  def sqliteDatabaseByConnections(auditJdbc: Connection, networkJdbc: Connection, eclairJdbc: Connection): Databases = new Databases {
     override val network = new SqliteNetworkDb(networkJdbc)
     override val audit = new SqliteAuditDb(auditJdbc)
     override val channels = new SqliteChannelsDb(eclairJdbc)
     override val peers = new SqlitePeersDb(eclairJdbc)
     override val payments = new SqlitePaymentsDb(eclairJdbc)
     override val pendingRelay = new SqlitePendingRelayDb(eclairJdbc)
+    override def backup(backupFile: File): Unit = {
+      val tmpFile = new File(backupFile.getAbsolutePath.concat(".tmp"))
 
-    override def backup(file: File): Unit = {
       SqliteUtils.using(eclairJdbc.createStatement()) {
-        statement => {
-          statement.executeUpdate(s"backup to ${file.getAbsolutePath}")
+        statement =>  {
+          statement.executeUpdate(s"backup to ${tmpFile.getAbsolutePath}")
         }
       }
+
+      // this will throw an exception if it fails, which is possible if the backup file is not on the same filesystem
+      // as the temporary file
+      Files.move(tmpFile.toPath, backupFile.toPath, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE)
     }
+    override val isBackupSupported: Boolean = true
+
+    override def obtainExclusiveLock(): Unit = ()
   }
+
+  def setupPsqlDatabases(dbConfig: Config): Databases = {
+    val database = dbConfig.getString("psql.database")
+    val host = dbConfig.getString("psql.host")
+    val port = dbConfig.getInt("psql.port")
+    val username = if (dbConfig.getIsNull("psql.username") || dbConfig.getString("psql.username").isBlank)
+      None
+    else
+      Some(dbConfig.getString("psql.username"))
+    val password = if (dbConfig.getIsNull("psql.password") || dbConfig.getString("psql.password").isBlank)
+      None
+    else
+      Some(dbConfig.getString("psql.password"))
+    val properties = {
+      val poolConfig = dbConfig.getConfig("psql.pool")
+      Map.empty
+        .updated("max-size", poolConfig.getInt("max-size").toLong)
+        .updated("connection-timeout", poolConfig.getDuration("connection-timeout").toMillis)
+        .updated("idle-timeout", poolConfig.getDuration("idle-timeout").toMillis)
+        .updated("max-life-time", poolConfig.getDuration("max-life-time").toMillis)
+
+    }
+    val lockLeaseInterval = dbConfig.getDuration("lock.lease-interval").toSeconds.seconds
+    val lockTimeout = dbConfig.getDuration("lock.lock-timeout").toSeconds.seconds
+
+    Databases.postgresJDBC(
+      database = database, host = host, port = port,
+      username = username, password = password,
+      poolProperties = properties,
+      lockLeaseInterval = lockLeaseInterval, lockTimeout = lockTimeout
+    )
+  }
+
 }