Peer storage (#2888)

Implements lightning/bolts#1110 to allow storing a small amount of data for our peers.

---------

Co-authored-by: t-bast <[email protected]>

Author: thomash-acinq

docs/release-notes/eclair-vnext.md

@@ -6,6 +6,13 @@
 
 <insert changes>
 
+### Peer storage
+
+With this release, eclair supports the `option_provide_storage` feature introduced in <https://github.com/lightning/bolts/pull/1110>.
+When `option_provide_storage` is enabled, eclair will store a small encrypted backup for peers that request it.
+This backup is limited to 65kB and node operators should customize the `eclair.peer-storage` configuration section to match their desired SLAs.
+This is mostly intended for LSPs that serve mobile wallets to allow users to restore their channels when they switch phones.
+
 ### API changes
 
 <insert changes>

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

@@ -73,6 +73,9 @@ eclair {
     option_dual_fund = optional
     option_quiesce = optional
     option_onion_messages = optional
+    // This feature should only be enabled when acting as an LSP for mobile wallets.
+    // When activating this feature, the peer-storage section should be customized to match desired SLAs.
+    option_provide_storage = disabled
     option_channel_type = optional
     option_scid_alias = optional
     option_payment_metadata = optional
@@ -596,6 +599,19 @@ eclair {
     enabled = true // enable automatic purges of expired invoices from the database
     interval = 24 hours // interval between expired invoice purges
   }
+
+  peer-storage {
+    // Peer storage is persisted only after this delay to reduce the number of writes when updating it multiple times in a row.
+    // A small delay may result in a lot of IO write operations, which can have a negative performance impact on the node.
+    // But using a large delay increases the risk of not storing the latest peer data if you restart your node while writes are pending.
+    write-delay = 1 minute
+    // Peer storage is kept this long after the last channel with that peer has been closed.
+    // A long delay here guarantees that peers who are offline while their channels are closed will be able to get their funds
+    // back if they restore from seed on a different device after the channels have been closed.
+    removal-delay = 30 days
+    // Frequency at which we clean our DB to remove peer storage from nodes with whom we don't have channels anymore.
+    cleanup-frequency = 1 day
+  }
 }
 
 akka {

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

@@ -275,6 +275,11 @@ object Features {
     val mandatory = 38
   }
 
+  case object ProvideStorage extends Feature with InitFeature with NodeFeature {
+    val rfcName = "option_provide_storage"
+    val mandatory = 42
+  }
+
   case object ChannelType extends Feature with InitFeature with NodeFeature {
     val rfcName = "option_channel_type"
     val mandatory = 44
@@ -358,6 +363,7 @@ object Features {
     DualFunding,
     Quiescence,
     OnionMessages,
+    ProvideStorage,
     ChannelType,
     ScidAlias,
     PaymentMetadata,

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

@@ -92,7 +92,8 @@ case class NodeParams(nodeKeyManager: NodeKeyManager,
                       revokedHtlcInfoCleanerConfig: RevokedHtlcInfoCleaner.Config,
                       willFundRates_opt: Option[LiquidityAds.WillFundRates],
                       peerWakeUpConfig: PeerReadyNotifier.WakeUpConfig,
-                      onTheFlyFundingConfig: OnTheFlyFunding.Config) {
+                      onTheFlyFundingConfig: OnTheFlyFunding.Config,
+                      peerStorageConfig: PeerStorageConfig) {
   val privateKey: Crypto.PrivateKey = nodeKeyManager.nodeKey.privateKey
 
   val nodeId: PublicKey = nodeKeyManager.nodeId
@@ -156,6 +157,13 @@ case class PaymentFinalExpiryConf(min: CltvExpiryDelta, max: CltvExpiryDelta) {
   }
 }
 
+/**
+ * @param writeDelay       delay before writing the peer's data to disk, which avoids doing multiple writes during bursts of storage updates.
+ * @param removalDelay     we keep our peer's data in our DB even after closing all of our channels with them, up to this duration.
+ * @param cleanUpFrequency frequency at which we go through the DB to remove unused storage.                    
+ */
+case class PeerStorageConfig(writeDelay: FiniteDuration, removalDelay: FiniteDuration, cleanUpFrequency: FiniteDuration)
+
 object NodeParams extends Logging {
 
   /**
@@ -680,6 +688,11 @@ object NodeParams extends Logging {
       onTheFlyFundingConfig = OnTheFlyFunding.Config(
         proposalTimeout = FiniteDuration(config.getDuration("on-the-fly-funding.proposal-timeout").getSeconds, TimeUnit.SECONDS),
       ),
+      peerStorageConfig = PeerStorageConfig(
+        writeDelay = FiniteDuration(config.getDuration("peer-storage.write-delay").getSeconds, TimeUnit.SECONDS),
+        removalDelay = FiniteDuration(config.getDuration("peer-storage.removal-delay").getSeconds, TimeUnit.SECONDS),
+        cleanUpFrequency = FiniteDuration(config.getDuration("peer-storage.cleanup-frequency").getSeconds, TimeUnit.SECONDS),
+      )
     )
   }
 }

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

@@ -37,7 +37,7 @@ import fr.acinq.eclair.crypto.WeakEntropyPool
 import fr.acinq.eclair.crypto.keymanager.{LocalChannelKeyManager, LocalNodeKeyManager, LocalOnChainKeyManager}
 import fr.acinq.eclair.db.Databases.FileBackup
 import fr.acinq.eclair.db.FileBackupHandler.FileBackupParams
-import fr.acinq.eclair.db.{Databases, DbEventHandler, FileBackupHandler}
+import fr.acinq.eclair.db.{Databases, DbEventHandler, FileBackupHandler, PeerStorageCleaner}
 import fr.acinq.eclair.io._
 import fr.acinq.eclair.message.Postman
 import fr.acinq.eclair.payment.offer.OfferManager
@@ -356,6 +356,9 @@ class Setup(val datadir: File,
         logger.warn("database backup is disabled")
         system.deadLetters
       }
+      _ = if (nodeParams.features.hasFeature(Features.ProvideStorage)) {
+        system.spawn(Behaviors.supervise(PeerStorageCleaner(nodeParams.db.peers, nodeParams.peerStorageConfig)).onFailure(typed.SupervisorStrategy.restart), name = "peer-storage-cleaner")
+      }
       dbEventHandler = system.actorOf(SimpleSupervisor.props(DbEventHandler.props(nodeParams), "db-event-handler", SupervisorStrategy.Resume))
       register = system.actorOf(SimpleSupervisor.props(Register.props(), "register", SupervisorStrategy.Resume))
       offerManager = system.spawn(Behaviors.supervise(OfferManager(nodeParams, router, paymentTimeout = 1 minute)).onFailure(typed.SupervisorStrategy.resume), name = "offer-manager")

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

@@ -13,8 +13,9 @@ import fr.acinq.eclair.payment.relay.OnTheFlyFunding
 import fr.acinq.eclair.payment.relay.Relayer.RelayFees
 import fr.acinq.eclair.router.Router
 import fr.acinq.eclair.wire.protocol.{ChannelAnnouncement, ChannelUpdate, NodeAddress, NodeAnnouncement}
-import fr.acinq.eclair.{CltvExpiry, MilliSatoshi, Paginated, RealShortChannelId, ShortChannelId, TimestampMilli}
+import fr.acinq.eclair.{CltvExpiry, MilliSatoshi, Paginated, RealShortChannelId, ShortChannelId, TimestampMilli, TimestampSecond}
 import grizzled.slf4j.Logging
+import scodec.bits.ByteVector
 
 import java.io.File
 import java.util.UUID
@@ -292,6 +293,21 @@ case class DualPeersDb(primary: PeersDb, secondary: PeersDb) extends PeersDb {
     runAsync(secondary.getRelayFees(nodeId))
     primary.getRelayFees(nodeId)
   }
+
+  override def updateStorage(nodeId: PublicKey, data: ByteVector): Unit = {
+    runAsync(secondary.updateStorage(nodeId, data))
+    primary.updateStorage(nodeId, data)
+  }
+
+  override def getStorage(nodeId: PublicKey): Option[ByteVector] = {
+    runAsync(secondary.getStorage(nodeId))
+    primary.getStorage(nodeId)
+  }
+
+  override def removePeerStorage(peerRemovedBefore: TimestampSecond): Unit = {
+    runAsync(secondary.removePeerStorage(peerRemovedBefore))
+    primary.removePeerStorage(peerRemovedBefore)
+  }
 }
 
 case class DualPaymentsDb(primary: PaymentsDb, secondary: PaymentsDb) extends PaymentsDb {

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

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2024 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.db
+
+import akka.actor.typed.Behavior
+import akka.actor.typed.scaladsl.Behaviors
+import fr.acinq.eclair.{PeerStorageConfig, TimestampSecond}
+
+/**
+ * This actor frequently deletes from our DB peer storage from nodes with whom we don't have channels anymore, after a
+ * grace period.
+ */
+object PeerStorageCleaner {
+  // @formatter:off
+  sealed trait Command
+  private case object CleanPeerStorage extends Command
+  // @formatter:on
+
+  def apply(db: PeersDb, config: PeerStorageConfig): Behavior[Command] = {
+    Behaviors.withTimers { timers =>
+      timers.startTimerWithFixedDelay(CleanPeerStorage, config.cleanUpFrequency)
+      Behaviors.receiveMessage {
+        case CleanPeerStorage =>
+          db.removePeerStorage(TimestampSecond.now() - config.removalDelay)
+          Behaviors.same
+      }
+    }
+  }
+
+}

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

@@ -17,8 +17,10 @@
 package fr.acinq.eclair.db
 
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
+import fr.acinq.eclair.TimestampSecond
 import fr.acinq.eclair.payment.relay.Relayer.RelayFees
 import fr.acinq.eclair.wire.protocol.NodeAddress
+import scodec.bits.ByteVector
 
 trait PeersDb {
 
@@ -34,4 +36,13 @@ trait PeersDb {
 
   def getRelayFees(nodeId: PublicKey): Option[RelayFees]
 
+  /** Update our peer's blob data when [[fr.acinq.eclair.Features.ProvideStorage]] is enabled. */
+  def updateStorage(nodeId: PublicKey, data: ByteVector): Unit
+
+  /** Get the last blob of data we stored for that peer, if [[fr.acinq.eclair.Features.ProvideStorage]] is enabled. */
+  def getStorage(nodeId: PublicKey): Option[ByteVector]
+
+  /** Remove storage from peers that have had no active channel with us for a while. */
+  def removePeerStorage(peerRemovedBefore: TimestampSecond): Unit
+
 }

eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgPeersDb.scala

@@ -18,22 +18,22 @@ package fr.acinq.eclair.db.pg
 
 import fr.acinq.bitcoin.scalacompat.Crypto
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
-import fr.acinq.eclair.MilliSatoshi
+import fr.acinq.eclair.{MilliSatoshi, TimestampSecond}
 import fr.acinq.eclair.db.Monitoring.Metrics.withMetrics
 import fr.acinq.eclair.db.Monitoring.Tags.DbBackends
 import fr.acinq.eclair.db.PeersDb
 import fr.acinq.eclair.db.pg.PgUtils.PgLock
 import fr.acinq.eclair.payment.relay.Relayer.RelayFees
 import fr.acinq.eclair.wire.protocol._
 import grizzled.slf4j.Logging
-import scodec.bits.BitVector
+import scodec.bits.{BitVector, ByteVector}
 
 import java.sql.Statement
 import javax.sql.DataSource
 
 object PgPeersDb {
   val DB_NAME = "peers"
-  val CURRENT_VERSION = 3
+  val CURRENT_VERSION = 4
 }
 
 class PgPeersDb(implicit ds: DataSource, lock: PgLock) extends PeersDb with Logging {
@@ -54,20 +54,31 @@ class PgPeersDb(implicit ds: DataSource, lock: PgLock) extends PeersDb with Logg
       statement.executeUpdate("CREATE TABLE local.relay_fees (node_id TEXT NOT NULL PRIMARY KEY, fee_base_msat BIGINT NOT NULL, fee_proportional_millionths BIGINT NOT NULL)")
     }
 
+    def migration34(statement: Statement): Unit = {
+      statement.executeUpdate("CREATE TABLE local.peer_storage (node_id TEXT NOT NULL PRIMARY KEY, data BYTEA NOT NULL, last_updated_at TIMESTAMP WITH TIME ZONE NOT NULL, removed_peer_at TIMESTAMP WITH TIME ZONE)")
+      statement.executeUpdate("CREATE INDEX removed_peer_at_idx ON local.peer_storage(removed_peer_at)")
+    }
+
     using(pg.createStatement()) { statement =>
       getVersion(statement, DB_NAME) match {
         case None =>
           statement.executeUpdate("CREATE SCHEMA IF NOT EXISTS local")
           statement.executeUpdate("CREATE TABLE local.peers (node_id TEXT NOT NULL PRIMARY KEY, data BYTEA NOT NULL)")
           statement.executeUpdate("CREATE TABLE local.relay_fees (node_id TEXT NOT NULL PRIMARY KEY, fee_base_msat BIGINT NOT NULL, fee_proportional_millionths BIGINT NOT NULL)")
-        case Some(v@(1 | 2)) =>
+          statement.executeUpdate("CREATE TABLE local.peer_storage (node_id TEXT NOT NULL PRIMARY KEY, data BYTEA NOT NULL, last_updated_at TIMESTAMP WITH TIME ZONE NOT NULL, removed_peer_at TIMESTAMP WITH TIME ZONE)")
+
+          statement.executeUpdate("CREATE INDEX removed_peer_at_idx ON local.peer_storage(removed_peer_at)")
+        case Some(v@(1 | 2 | 3)) =>
           logger.warn(s"migrating db $DB_NAME, found version=$v current=$CURRENT_VERSION")
           if (v < 2) {
             migration12(statement)
           }
           if (v < 3) {
             migration23(statement)
           }
+          if (v < 4) {
+            migration34(statement)
+          }
         case Some(CURRENT_VERSION) => () // table is up-to-date, nothing to do
         case Some(unknownVersion) => throw new RuntimeException(s"Unknown version of DB $DB_NAME found, version=$unknownVersion")
       }
@@ -98,6 +109,20 @@ class PgPeersDb(implicit ds: DataSource, lock: PgLock) extends PeersDb with Logg
         statement.setString(1, nodeId.value.toHex)
         statement.executeUpdate()
       }
+      using(pg.prepareStatement("UPDATE local.peer_storage SET removed_peer_at = ? WHERE node_id = ?")) { statement =>
+        statement.setTimestamp(1, TimestampSecond.now().toSqlTimestamp)
+        statement.setString(2, nodeId.value.toHex)
+        statement.executeUpdate()
+      }
+    }
+  }
+
+  override def removePeerStorage(peerRemovedBefore: TimestampSecond): Unit = withMetrics("peers/remove-storage", DbBackends.Postgres) {
+    withLock { pg =>
+      using(pg.prepareStatement("DELETE FROM local.peer_storage WHERE removed_peer_at < ?")) { statement =>
+        statement.setTimestamp(1, peerRemovedBefore.toSqlTimestamp)
+        statement.executeUpdate()
+      }
     }
   }
 
@@ -155,4 +180,32 @@ class PgPeersDb(implicit ds: DataSource, lock: PgLock) extends PeersDb with Logg
       }
     }
   }
+
+  override def updateStorage(nodeId: PublicKey, data: ByteVector): Unit = withMetrics("peers/update-storage", DbBackends.Postgres) {
+    withLock { pg =>
+      using(pg.prepareStatement(
+        """
+        INSERT INTO local.peer_storage (node_id, data, last_updated_at, removed_peer_at)
+        VALUES (?, ?, ?, NULL)
+        ON CONFLICT (node_id)
+        DO UPDATE SET data = EXCLUDED.data, last_updated_at = EXCLUDED.last_updated_at, removed_peer_at = NULL
+        """)) { statement =>
+        statement.setString(1, nodeId.value.toHex)
+        statement.setBytes(2, data.toArray)
+        statement.setTimestamp(3, TimestampSecond.now().toSqlTimestamp)
+        statement.executeUpdate()
+      }
+    }
+  }
+
+  override def getStorage(nodeId: PublicKey): Option[ByteVector] = withMetrics("peers/get-storage", DbBackends.Postgres) {
+    withLock { pg =>
+      using(pg.prepareStatement("SELECT data FROM local.peer_storage WHERE node_id = ?")) { statement =>
+        statement.setString(1, nodeId.value.toHex)
+        statement.executeQuery()
+          .headOption
+          .map(rs => ByteVector(rs.getBytes("data")))
+      }
+    }
+  }
 }