Peer storage, but not persistent

Author: thomash-acinq

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/io/Peer.scala

@@ -44,7 +44,8 @@ import fr.acinq.eclair.remote.EclairInternalsSerializer.RemoteTypes
 import fr.acinq.eclair.router.Router
 import fr.acinq.eclair.wire.protocol
 import fr.acinq.eclair.wire.protocol.FailureMessageCodecs.createBadOnionFailure
-import fr.acinq.eclair.wire.protocol.{AddFeeCredit, ChannelTlv, CurrentFeeCredit, Error, HasChannelId, HasTemporaryChannelId, LightningMessage, LiquidityAds, NodeAddress, OnTheFlyFundingFailureMessage, OnionMessage, OnionRoutingPacket, RecommendedFeerates, RoutingMessage, SpliceInit, TlvStream, TxAbort, UnknownMessage, Warning, WillAddHtlc, WillFailHtlc, WillFailMalformedHtlc}
+import fr.acinq.eclair.wire.protocol.{AddFeeCredit, ChannelTlv, CurrentFeeCredit, Error, HasChannelId, HasTemporaryChannelId, LightningMessage, LiquidityAds, NodeAddress, OnTheFlyFundingFailureMessage, OnionMessage, OnionRoutingPacket, PeerStorageRetrieval, PeerStorageStore, RecommendedFeerates, RoutingMessage, SpliceInit, TlvStream, TxAbort, UnknownMessage, Warning, WillAddHtlc, WillFailHtlc, WillFailMalformedHtlc}
+import scodec.bits.ByteVector
 
 /**
  * This actor represents a logical peer. There is one [[Peer]] per unique remote node id at all time.
@@ -84,7 +85,7 @@ class Peer(val nodeParams: NodeParams,
         FinalChannelId(state.channelId) -> channel
       }.toMap
       context.system.eventStream.publish(PeerCreated(self, remoteNodeId))
-      goto(DISCONNECTED) using DisconnectedData(channels) // when we restart, we will attempt to reconnect right away, but then we'll wait
+      goto(DISCONNECTED) using DisconnectedData(channels, None) // when we restart, we will attempt to reconnect right away, but then we'll wait
   }
 
   when(DISCONNECTED) {
@@ -93,7 +94,7 @@ class Peer(val nodeParams: NodeParams,
       stay()
 
     case Event(connectionReady: PeerConnection.ConnectionReady, d: DisconnectedData) =>
-      gotoConnected(connectionReady, d.channels.map { case (k: ChannelId, v) => (k, v) })
+      gotoConnected(connectionReady, d.channels.map { case (k: ChannelId, v) => (k, v) }, d.peerStorage)
 
     case Event(Terminated(actor), d: DisconnectedData) if d.channels.values.toSet.contains(actor) =>
       // we have at most 2 ids: a TemporaryChannelId and a FinalChannelId
@@ -461,7 +462,7 @@ class Peer(val nodeParams: NodeParams,
           stopPeer()
         } else {
           d.channels.values.toSet[ActorRef].foreach(_ ! INPUT_DISCONNECTED) // we deduplicate with toSet because there might be two entries per channel (tmp id and final id)
-          goto(DISCONNECTED) using DisconnectedData(d.channels.collect { case (k: FinalChannelId, v) => (k, v) })
+          goto(DISCONNECTED) using DisconnectedData(d.channels.collect { case (k: FinalChannelId, v) => (k, v) }, d.peerStorage)
         }
 
       case Event(Terminated(actor), d: ConnectedData) if d.channels.values.toSet.contains(actor) =>
@@ -480,7 +481,7 @@ class Peer(val nodeParams: NodeParams,
         log.debug(s"got new connection, killing current one and switching")
         d.peerConnection ! PeerConnection.Kill(KillReason.ConnectionReplaced)
         d.channels.values.toSet[ActorRef].foreach(_ ! INPUT_DISCONNECTED) // we deduplicate with toSet because there might be two entries per channel (tmp id and final id)
-        gotoConnected(connectionReady, d.channels)
+        gotoConnected(connectionReady, d.channels, d.peerStorage)
 
       case Event(msg: OnionMessage, _: ConnectedData) =>
         OnionMessages.process(nodeParams.privateKey, msg) match {
@@ -513,6 +514,9 @@ class Peer(val nodeParams: NodeParams,
         d.peerConnection forward unknownMsg
         stay()
 
+      case Event(store: PeerStorageStore, d: ConnectedData) if nodeParams.features.hasFeature(Features.ProvideStorage) && d.channels.nonEmpty =>
+        stay() using d.copy(peerStorage = Some(store.blob))
+
       case Event(unhandledMsg: LightningMessage, _) =>
         log.warning("ignoring message {}", unhandledMsg)
         stay()
@@ -744,7 +748,7 @@ class Peer(val nodeParams: NodeParams,
       context.system.eventStream.publish(PeerDisconnected(self, remoteNodeId))
   }
 
-  private def gotoConnected(connectionReady: PeerConnection.ConnectionReady, channels: Map[ChannelId, ActorRef]): State = {
+  private def gotoConnected(connectionReady: PeerConnection.ConnectionReady, channels: Map[ChannelId, ActorRef], peerStorage: Option[ByteVector]): State = {
     require(remoteNodeId == connectionReady.remoteNodeId, s"invalid nodeId: $remoteNodeId != ${connectionReady.remoteNodeId}")
     log.debug("got authenticated connection to address {}", connectionReady.address)
 
@@ -754,6 +758,9 @@ class Peer(val nodeParams: NodeParams,
       nodeParams.db.peers.addOrUpdatePeer(remoteNodeId, connectionReady.address)
     }
 
+    // If we have some data stored from our peer, we send it to them before doing anything else.
+    peerStorage.foreach(connectionReady.peerConnection ! PeerStorageRetrieval(_))
+
     // let's bring existing/requested channels online
     channels.values.toSet[ActorRef].foreach(_ ! INPUT_RECONNECTED(connectionReady.peerConnection, connectionReady.localInit, connectionReady.remoteInit)) // we deduplicate with toSet because there might be two entries per channel (tmp id and final id)
 
@@ -771,7 +778,7 @@ class Peer(val nodeParams: NodeParams,
       connectionReady.peerConnection ! CurrentFeeCredit(nodeParams.chainHash, feeCredit.getOrElse(0 msat))
     }
 
-    goto(CONNECTED) using ConnectedData(connectionReady.address, connectionReady.peerConnection, connectionReady.localInit, connectionReady.remoteInit, channels, feerates, None)
+    goto(CONNECTED) using ConnectedData(connectionReady.address, connectionReady.peerConnection, connectionReady.localInit, connectionReady.remoteInit, channels, feerates, None, peerStorage)
   }
 
   /**
@@ -908,10 +915,14 @@ object Peer {
 
   sealed trait Data {
     def channels: Map[_ <: ChannelId, ActorRef] // will be overridden by Map[FinalChannelId, ActorRef] or Map[ChannelId, ActorRef]
+    def peerStorage: Option[ByteVector]
+  }
+  case object Nothing extends Data {
+    override def channels = Map.empty
+    override def peerStorage: Option[ByteVector] = None
   }
-  case object Nothing extends Data { override def channels = Map.empty }
-  case class DisconnectedData(channels: Map[FinalChannelId, ActorRef]) extends Data
-  case class ConnectedData(address: NodeAddress, peerConnection: ActorRef, localInit: protocol.Init, remoteInit: protocol.Init, channels: Map[ChannelId, ActorRef], currentFeerates: RecommendedFeerates, previousFeerates_opt: Option[RecommendedFeerates]) extends Data {
+  case class DisconnectedData(channels: Map[FinalChannelId, ActorRef], peerStorage: Option[ByteVector]) extends Data
+  case class ConnectedData(address: NodeAddress, peerConnection: ActorRef, localInit: protocol.Init, remoteInit: protocol.Init, channels: Map[ChannelId, ActorRef], currentFeerates: RecommendedFeerates, previousFeerates_opt: Option[RecommendedFeerates], peerStorage: Option[ByteVector]) extends Data {
     val connectionInfo: ConnectionInfo = ConnectionInfo(address, peerConnection, localInit, remoteInit)
     def localFeatures: Features[InitFeature] = localInit.features
     def remoteFeatures: Features[InitFeature] = remoteInit.features

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageCodecs.scala

@@ -22,7 +22,7 @@ import fr.acinq.eclair.wire.protocol.CommonCodecs._
 import fr.acinq.eclair.{Features, InitFeature, KamonExt}
 import scodec.bits.{BinStringSyntax, BitVector, ByteVector}
 import scodec.codecs._
-import scodec.{Attempt, Codec}
+import scodec.{Attempt, Codec, Err}
 
 /**
  * Created by PM on 15/11/2016.
@@ -389,6 +389,17 @@ object LightningMessageCodecs {
       ("onionPacket" | MessageOnionCodecs.messageOnionPacketCodec) ::
       ("tlvStream" | OnionMessageTlv.onionMessageTlvCodec)).as[OnionMessage]
 
+  private def isAcceptableBlobLength(length: Int) =
+    if (length <= 65531) Attempt.Successful(length) else Attempt.failure(Err(s"length $length is larger than 65531"))
+
+  val peerStorageStore: Codec[PeerStorageStore] = (
+    ("blob" | variableSizeBytes(uint16.exmap(isAcceptableBlobLength, isAcceptableBlobLength), bytes)) ::
+      ("tlvStream" | PeerStorageTlv.peerStorageTlvCodec)).as[PeerStorageStore]
+
+  val peerStorageRetrieval: Codec[PeerStorageRetrieval] = (
+    ("blob" | variableSizeBytes(uint16.exmap(isAcceptableBlobLength, isAcceptableBlobLength), bytes)) ::
+      ("tlvStream" | PeerStorageTlv.peerStorageTlvCodec)).as[PeerStorageRetrieval]
+
   // NB: blank lines to minimize merge conflicts
 
   //
@@ -476,6 +487,8 @@ object LightningMessageCodecs {
   val lightningMessageCodec = discriminated[LightningMessage].by(uint16)
     .typecase(1, warningCodec)
     .typecase(2, stfuCodec)
+    .typecase(7, peerStorageStore)
+    .typecase(9, peerStorageRetrieval)
     .typecase(16, initCodec)
     .typecase(17, errorCodec)
     .typecase(18, pingCodec)

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageTypes.scala

@@ -602,6 +602,10 @@ case class GossipTimestampFilter(chainHash: BlockHash, firstTimestamp: Timestamp
 
 case class OnionMessage(blindingKey: PublicKey, onionRoutingPacket: OnionRoutingPacket, tlvStream: TlvStream[OnionMessageTlv] = TlvStream.empty) extends LightningMessage
 
+case class PeerStorageStore(blob: ByteVector, tlvStream: TlvStream[PeerStorageTlv] = TlvStream.empty) extends LightningMessage
+
+case class PeerStorageRetrieval(blob: ByteVector, tlvStream: TlvStream[PeerStorageTlv] = TlvStream.empty) extends LightningMessage
+
 // NB: blank lines to minimize merge conflicts
 
 //

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/PeerStorageTlv.scala

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021 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.wire.protocol
+
+import fr.acinq.eclair.wire.protocol.CommonCodecs.varint
+import fr.acinq.eclair.wire.protocol.TlvCodecs.tlvStream
+import scodec.Codec
+import scodec.codecs.discriminated
+
+/**
+ * Created by thomash on July 2024.
+ */
+
+sealed trait PeerStorageTlv extends Tlv
+
+object PeerStorageTlv {
+  val peerStorageTlvCodec: Codec[TlvStream[PeerStorageTlv]] = tlvStream(discriminated[PeerStorageTlv].by(varint))
+}

eclair-core/src/test/scala/fr/acinq/eclair/io/ReconnectionTaskSpec.scala

@@ -38,8 +38,8 @@ class ReconnectionTaskSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike
   private val recommendedFeerates = RecommendedFeerates(Block.RegtestGenesisBlock.hash, TestConstants.feeratePerKw, TestConstants.anchorOutputsFeeratePerKw)
 
   private val PeerNothingData = Peer.Nothing
-  private val PeerDisconnectedData = Peer.DisconnectedData(channels)
-  private val PeerConnectedData = Peer.ConnectedData(fakeIPAddress, system.deadLetters, null, null, channels.map { case (k: ChannelId, v) => (k, v) }, recommendedFeerates, None)
+  private val PeerDisconnectedData = Peer.DisconnectedData(channels, None)
+  private val PeerConnectedData = Peer.ConnectedData(fakeIPAddress, system.deadLetters, null, null, channels.map { case (k: ChannelId, v) => (k, v) }, recommendedFeerates, None, None)
 
   case class FixtureParam(nodeParams: NodeParams, remoteNodeId: PublicKey, reconnectionTask: TestFSMRef[ReconnectionTask.State, ReconnectionTask.Data, ReconnectionTask], monitor: TestProbe)
 
@@ -82,7 +82,7 @@ class ReconnectionTaskSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike
     import f._
 
     val peer = TestProbe()
-    peer.send(reconnectionTask, Peer.Transition(PeerNothingData, Peer.DisconnectedData(Map.empty)))
+    peer.send(reconnectionTask, Peer.Transition(PeerNothingData, Peer.DisconnectedData(Map.empty, None)))
     monitor.expectNoMessage()
   }