Add optional rate-limit on incoming pre-auth connections (#3356)

### docs/release-notes/eclair-vnext.md
@@ -26,6 +26,28 @@ eclair.features.option_onion_messages = disabled
 eclair.features.option_onion_messages_only_channels = optional
 ```
 
+### Restrict the number of pending unauthenticated incoming connections
+
+Peers that open a connection but never complete the BOLT 8 handshake consume resources on our node (memory, file
+descriptors and CPU). We now bound how many of those we're willing to keep around:
+
+```conf
+eclair.peer-connection.max-pending-incoming-connections = 500
+eclair.peer-connection.pending-connection-min-age = 1 second
+eclair.peer-connection.pending-connection-accept-delay = 100 milliseconds
+```
+
+When we reach `max-pending-incoming-connections`, we drop the oldest connection that hasn't authenticated yet to make
+room for the new one, and then wait for `pending-connection-accept-delay` before accepting the next one, which lets the
+kernel backlog absorb (and discard) the excess connection attempts. We never drop a pending connection that is more
+recent than `pending-connection-min-age`: this guarantees that honest peers always have enough time to complete the
+handshake, even while we're being flooded with connection attempts. When every pending connection is that recent, we
+reject the incoming connection instead.
+
+The default values shouldn't be reached by honest nodes. Note that this is only a last-resort safety net: DDoS
+protection is much more efficiently handled at the network layer (for example by a cloud provider). Setting
+`max-pending-incoming-connections = 0` disables this limit entirely.
+
 ### Configuration changes
 
 #### Gossip queries
@@ -72,6 +94,7 @@ authentication method used.
 - Answering channel range queries is much cheaper: we cache the timestamps and checksums of our channel updates instead of recomputing them for the whole routing table on every incoming query
 - We ignore duplicate `short_channel_id`s in a `query_short_channel_ids`, and reject queries whose query flags don't cover every `short_channel_id`, or that are sent before we've replied to the previous one
 - We ignore `reply_short_channel_ids_end` messages that don't answer one of our queries: a peer could previously send us one to make us drop our synchronization state and ignore the rest of its replies
+- We now disconnect peers that authenticate but whose connection is never initialized: such connections previously stayed around forever
 
 ## Verifying signatures
 

### eclair-core/src/main/resources/reference.conf
@@ -423,6 +423,21 @@ eclair {
     // This should be disabled if your node is behind a load balancer that doesn't preserve source IP addresses.
     send-remote-address-init = true
     max-no-channels = 64 // maximum number of incoming connections from peers that do not have any channels with us
+    // Maximum number of incoming connections that haven't completed the BOLT 8 handshake yet. Peers that connect
+    // without ever authenticating cost us memory, file descriptors and CPU, so we bound how many of them we keep
+    // around: when this limit is reached, we drop the oldest pending connection to make room for the new one.
+    // NB: this is only a last-resort safety net, DDoS protection is much more efficiently handled at the network
+    // layer (e.g. by a cloud provider).
+    max-pending-incoming-connections = 500
+    // We never drop a pending connection before it reaches that age, which guarantees that honest peers always have
+    // enough time to complete the handshake, even while we're being flooded with connection attempts. When every
+    // pending connection is more recent than that, we reject the incoming connection instead.
+    pending-connection-min-age = 1 second
+    // Whenever we had to drop a pending connection, we wait for that duration before accepting the next one. This lets
+    // the kernel backlog absorb (and discard) the excess connection attempts instead of spending our own resources on
+    // them. Note that this only applies while we're at `max-pending-incoming-connections`: it never slows down
+    // connections that we have capacity for.
+    pending-connection-accept-delay = 100 milliseconds
   }
 
   // When relaying payments or messages to mobile peers who are disconnected, we may try to wake them up using a mobile

### eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala
@@ -541,6 +541,14 @@ object NodeParams extends Logging {
     val maxNoChannels = config.getInt("peer-connection.max-no-channels")
     require(maxNoChannels > 0, "peer-connection.max-no-channels must be > 0")
 
+    val maxPendingIncomingConnections = config.getInt("peer-connection.max-pending-incoming-connections")
+    require(maxPendingIncomingConnections >= 0, "peer-connection.max-pending-incoming-connections must be >= 0 (0 disables the limit)")
+    // Those two durations may be set to 0 to disable the corresponding protection.
+    val pendingConnectionMinAge = FiniteDuration(config.getDuration("peer-connection.pending-connection-min-age").toMillis, TimeUnit.MILLISECONDS)
+    require(pendingConnectionMinAge >= Duration.Zero, "peer-connection.pending-connection-min-age must be >= 0")
+    val pendingConnectionAcceptDelay = FiniteDuration(config.getDuration("peer-connection.pending-connection-accept-delay").toMillis, TimeUnit.MILLISECONDS)
+    require(pendingConnectionAcceptDelay >= Duration.Zero, "peer-connection.pending-connection-accept-delay must be >= 0")
+
     val willFundRates_opt = {
       val supportedPaymentTypes = Map(
         LiquidityAds.PaymentType.FromChannelBalance.rfcName -> LiquidityAds.PaymentType.FromChannelBalance,
@@ -687,6 +695,9 @@ object NodeParams extends Logging {
         maxGossipQueriesPerSecond = config.getInt("router.sync.max-queries-per-second"),
         sendRemoteAddressInit = config.getBoolean("peer-connection.send-remote-address-init"),
         maxNoChannels = maxNoChannels,
+        maxPendingIncomingConnections = maxPendingIncomingConnections,
+        pendingConnectionMinAge = pendingConnectionMinAge,
+        pendingConnectionAcceptDelay = pendingConnectionAcceptDelay,
       ),
       routerConf = RouterConf(
         watchSpentWindow = watchSpentWindow,

### eclair-core/src/main/scala/fr/acinq/eclair/io/Monitoring.scala
@@ -39,6 +39,9 @@ object Monitoring {
 
     val IncomingConnectionsNoChannels = Kamon.gauge("incomingconnections.nochannels")
     val IncomingConnectionsDisconnected = Kamon.counter("incomingconnections.disconnected")
+    val IncomingConnectionsPending = Kamon.gauge("incomingconnections.pending")
+    val IncomingConnectionsEvicted = Kamon.counter("incomingconnections.evicted")
+    val IncomingConnectionsRejected = Kamon.counter("incomingconnections.rejected")
 
     val OnTheFlyFunding = Kamon.counter("on-the-fly-funding.attempts")
     val OnTheFlyFundingFees = Kamon.histogram("on-the-fly-funding.fees-msat")

### eclair-core/src/main/scala/fr/acinq/eclair/io/PeerConnection.scala
@@ -86,16 +86,16 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
 
   when(AUTHENTICATING) {
     case Event(TransportHandler.HandshakeCompleted(remoteNodeId), d: AuthenticatingData) =>
-      cancelTimer(AUTH_TIMER)
       Logs.withMdc(diagLog)(Logs.mdc(remoteNodeId_opt = Some(remoteNodeId))) {
         log.info(s"connection authenticated (direction=${if (d.pendingAuth.outgoing) "outgoing" else "incoming"})")
       }
       Metrics.PeerConnectionsConnecting.withTag(Tags.ConnectionState, Tags.ConnectionStates.Authenticated).increment()
+      d.pendingAuth.authTracker_opt.foreach(_ ! Authenticated(self, remoteNodeId, d.pendingAuth.outgoing))
       switchboard ! Authenticated(self, remoteNodeId, d.pendingAuth.outgoing)
       goto(BEFORE_INIT) using BeforeInitData(remoteNodeId, d.pendingAuth, d.transport, d.isPersistent)
 
     case Event(AuthTimeout, d: AuthenticatingData) =>
-      log.warning(s"authentication timed out after ${conf.authTimeout}")
+      log.warning("authentication timed out after {}", conf.authTimeout)
       d.pendingAuth.origin_opt.foreach(_ ! ConnectionResult.AuthenticationFailed("authentication timed out"))
       stop(FSM.Normal)
 
@@ -107,6 +107,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
 
   when(BEFORE_INIT) {
     case Event(InitializeConnection(peer, chainHash, localFeatures, doSync, fundingRates_opt), d: BeforeInitData) =>
+      cancelTimer(AUTH_TIMER)
       d.transport ! TransportHandler.Listener(self)
       Metrics.PeerConnectionsConnecting.withTag(Tags.ConnectionState, Tags.ConnectionStates.Initializing).increment()
       log.debug(s"using features=$localFeatures")
@@ -125,6 +126,11 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
       unstashAll() // unstash remote init if it already arrived
       goto(INITIALIZING) using InitializingData(chainHash, d.pendingAuth, d.remoteNodeId, d.transport, peer, localInit, doSync, d.isPersistent)
 
+    case Event(AuthTimeout, d: BeforeInitData) =>
+      log.warning("connection was not initialized within {}", conf.authTimeout)
+      d.pendingAuth.origin_opt.foreach(_ ! ConnectionResult.InitializationFailed("connection was never initialized"))
+      stop(FSM.Normal)
+
     case Event(_: protocol.Init, _) =>
       log.debug("stashing remote init")
       stash()
@@ -702,7 +708,10 @@ object PeerConnection {
                   maxOnionMessagesPerSecond: Int,
                   maxGossipQueriesPerSecond: Int,
                   sendRemoteAddressInit: Boolean,
-                  maxNoChannels: Int)
+                  maxNoChannels: Int,
+                  maxPendingIncomingConnections: Int,
+                  pendingConnectionMinAge: FiniteDuration,
+                  pendingConnectionAcceptDelay: FiniteDuration)
 
   /**
    * Gossip sync messages, which all carry a chain hash. Unlike gossip announcements, which we validate and may ignore,
@@ -753,7 +762,7 @@ object PeerConnection {
   case object INITIALIZING extends State
   case object CONNECTED extends State
 
-  case class PendingAuth(connection: ActorRef, remoteNodeId_opt: Option[PublicKey], address: NodeAddress, origin_opt: Option[ActorRef], transport_opt: Option[ActorRef] = None, isPersistent: Boolean) {
+  case class PendingAuth(connection: ActorRef, remoteNodeId_opt: Option[PublicKey], address: NodeAddress, origin_opt: Option[ActorRef], transport_opt: Option[ActorRef] = None, isPersistent: Boolean, authTracker_opt: Option[ActorRef] = None) {
     def outgoing: Boolean = remoteNodeId_opt.isDefined // if this is an outgoing connection, we know the node id in advance
   }
   case class Authenticated(peerConnection: ActorRef, remoteNodeId: PublicKey, outgoing: Boolean) extends RemoteTypes
@@ -794,6 +803,7 @@ object PeerConnection {
     case object NoRemainingChannel extends KillReason
     case object AllChannelsFail extends KillReason
     case object ConnectionReplaced extends KillReason
+    case object TooManyPendingConnections extends KillReason
   }
   // @formatter:on
 

### eclair-core/src/main/scala/fr/acinq/eclair/io/Server.scala
@@ -16,24 +16,26 @@
 
 package fr.acinq.eclair.io
 
-import java.net.InetSocketAddress
 import akka.Done
-import akka.actor.{Actor, ActorRef, DiagnosticActorLogging, Props}
+import akka.actor.{Actor, ActorRef, DiagnosticActorLogging, Props, Terminated}
 import akka.event.Logging.MDC
 import akka.io.Tcp.SO.KeepAlive
 import akka.io.{IO, Tcp}
-import fr.acinq.eclair.Logs
 import fr.acinq.eclair.Logs.LogCategory
 import fr.acinq.eclair.crypto.Noise.KeyPair
-import fr.acinq.eclair.wire.protocol.{IPAddress, NodeAddress}
+import fr.acinq.eclair.io.Monitoring.Metrics
+import fr.acinq.eclair.wire.protocol.IPAddress
+import fr.acinq.eclair.{Logs, TimestampMilli}
 
+import java.net.InetSocketAddress
 import scala.concurrent.Promise
 
 /**
  * Created by PM on 27/10/2015.
  */
 class Server(keyPair: KeyPair, peerConnectionConf: PeerConnection.Conf, switchboard: ActorRef, router: ActorRef, address: InetSocketAddress, bound: Option[Promise[Done]] = None) extends Actor with DiagnosticActorLogging {
 
+  import Server._
   import Tcp._
   import context.system
 
@@ -43,27 +45,104 @@ class Server(keyPair: KeyPair, peerConnectionConf: PeerConnection.Conf, switchbo
     case Bound(localAddress) =>
       bound.map(_.success(Done))
       log.info(s"bound on $localAddress")
-      // Accept connections one by one
+      // Accept connections one by one.
       sender() ! ResumeAccepting(batchSize = 1)
-      context.become(listening(sender()))
+      Metrics.IncomingConnectionsPending.withoutTags().update(0)
+      context.become(listening(sender(), Map.empty))
 
     case CommandFailed(_: Bind) =>
       bound.map(_.failure(new RuntimeException("TCP bind failed")))
       context stop self
   }
 
-  def listening(listener: ActorRef): Receive = {
+  /**
+   * @param pending incoming connections that haven't completed the BOLT 8 handshake yet, with the time at which we
+   *                accepted them. Peers that connect without ever authenticating consume resources (memory, file
+   *                descriptors and CPU), so we bound how many of them we're willing to keep around.
+   */
+  def listening(listener: ActorRef, pending: Map[ActorRef, TimestampMilli]): Receive = {
     case Connected(remote, _) =>
-      log.info(s"connected to $remote")
       val connection = sender()
-      val peerConnection = context.actorOf(PeerConnection.props(
-        keyPair = keyPair,
-        conf = peerConnectionConf,
-        switchboard = switchboard,
-        router = router
-      ))
-      peerConnection ! PeerConnection.PendingAuth(connection, remoteNodeId_opt = None, address = IPAddress(remote.getAddress, remote.getPort), origin_opt = None, isPersistent = true)
-      listener ! ResumeAccepting(batchSize = 1)
+      // NB: we check whether we have capacity *before* creating the peer connection: the rejection path must be as
+      // cheap as possible, since it is exercised exactly when we're being flooded.
+      checkLimits(pending) match {
+        case Admission.Accept =>
+          val pending1 = accept(connection, remote, pending)
+          listener ! ResumeAccepting(batchSize = 1)
+          Metrics.IncomingConnectionsPending.withoutTags().update(pending1.size)
+          context.become(listening(listener, pending1))
+        case Admission.Evict(evicted) =>
+          // We make room for that new connection by dropping the oldest one that is pending authentication.
+          Metrics.IncomingConnectionsEvicted.withoutTags().increment()
+          log.debug("dropping pending connection to make room for incoming connection from {}", remote)
+          evicted ! PeerConnection.Kill(PeerConnection.KillReason.TooManyPendingConnections)
+          // We're above our limit: instead of sending ResumeAccepting immediately, we let the kernel backlog absorb
+          // (and discard) the excess connection attempts. Meanwhile, the pending connections may also complete the
+          // authentication handshake, which lets us free up resources before we start accepting new connections.
+          context.system.scheduler.scheduleOnce(peerConnectionConf.pendingConnectionAcceptDelay, self, AcceptNext)(context.dispatcher)
+          // NB: we remove the evicted connection immediately instead of waiting for its Terminated event, otherwise we
+          // could select it again as eviction candidate for the next incoming connection.
+          val pending1 = accept(connection, remote, pending - evicted)
+          Metrics.IncomingConnectionsPending.withoutTags().update(pending1.size)
+          context.become(listening(listener, pending1))
+        case Admission.Reject =>
+          // Every pending connection is still within its grace period: we protect them and reject the new connection.
+          Metrics.IncomingConnectionsRejected.withoutTags().increment()
+          log.debug("rejecting incoming connection from {}: too many pending connections", remote)
+          connection ! Abort
+          listener ! ResumeAccepting(batchSize = 1)
+      }
+
+    case AcceptNext => listener ! ResumeAccepting(batchSize = 1)
+
+    case PeerConnection.Authenticated(peerConnection, _, _) =>
+      // This connection isn't pending authentication anymore: we stop watching it and free up its slot.
+      context.unwatch(peerConnection)
+      val pending1 = pending - peerConnection
+      Metrics.IncomingConnectionsPending.withoutTags().update(pending1.size)
+      context.become(listening(listener, pending1))
+
+    case Terminated(peerConnection) =>
+      // The connection died before completing the BOLT 8 handshake (timeout, disconnection, or our own eviction).
+      val pending1 = pending - peerConnection
+      Metrics.IncomingConnectionsPending.withoutTags().update(pending1.size)
+      context.become(listening(listener, pending1))
+
+      // Confirmation that a connection we didn't have capacity for was indeed aborted: nothing to do.
+    case _: ConnectionClosed => ()
+
+    case GetPendingConnections(replyTo) => replyTo ! PendingConnections(pending.keySet)
+  }
+
+  private def accept(connection: ActorRef, remote: InetSocketAddress, pending: Map[ActorRef, TimestampMilli]): Map[ActorRef, TimestampMilli] = {
+    log.info("connected to {}", remote)
+    val peerConnection = context.actorOf(PeerConnection.props(
+      keyPair = keyPair,
+      conf = peerConnectionConf,
+      switchboard = switchboard,
+      router = router
+    ))
+    peerConnection ! PeerConnection.PendingAuth(connection, remoteNodeId_opt = None, address = IPAddress(remote.getAddress, remote.getPort), origin_opt = None, isPersistent = true, authTracker_opt = Some(self))
+    context.watch(peerConnection)
+    pending + (peerConnection -> TimestampMilli.now())
+  }
+
+  private def checkLimits(pending: Map[ActorRef, TimestampMilli]): Admission = {
+    if (peerConnectionConf.maxPendingIncomingConnections == 0 || pending.size < peerConnectionConf.maxPendingIncomingConnections) {
+      Admission.Accept
+    } else {
+      // NB: pending is guaranteed to be non-empty (otherwise we would be in the case above).
+      val (oldest, acceptedAt) = pending.minBy(_._2)
+      // When we've reached our maximum capacity, we don't immediately evict the oldest pending connection, otherwise
+      // attackers could just spam us with new connections and we would evict pending honest connections before they
+      // have a chance to complete the authentication handshake. This guarantees that honest connections can eventually
+      // be accepted, only degrading the initial latency.
+      if (TimestampMilli.now() - acceptedAt < peerConnectionConf.pendingConnectionMinAge) {
+        Admission.Reject
+      } else {
+        Admission.Evict(oldest)
+      }
+    }
   }
 
   override def mdc(currentMessage: Any): MDC = Logs.mdc(Some(LogCategory.CONNECTION))
@@ -73,5 +152,29 @@ object Server {
 
   def props(keyPair: KeyPair, peerConnectionConf: PeerConnection.Conf, switchboard: ActorRef, router: ActorRef, address: InetSocketAddress, bound: Option[Promise[Done]] = None): Props = Props(new Server(keyPair, peerConnectionConf, switchboard, router: ActorRef, address, bound))
 
+  /**
+   * When we've reached our limits and have too many pending connections, we add a delay before accepting the next
+   * connection, which allows pending connections to complete the authentication handshake. This message is sent after
+   * the delay to resume listening.
+   */
+  private case object AcceptNext
+
+  // @formatter:off
+  private[io] case class GetPendingConnections(replyTo: ActorRef)
+  private[io] case class PendingConnections(peerConnections: Set[ActorRef])
+  // @formatter:on
+
+  // @formatter:off
+  private sealed trait Admission
+  private object Admission {
+    /** We have capacity for another pending connection. */
+    case object Accept extends Admission
+    /** We're at capacity, but the given pending connection is old enough to be dropped to make room. */
+    case class Evict(peerConnection: ActorRef) extends Admission
+    /** We're at capacity, and every pending connection is too recent to be dropped: we protect them instead by rejecting the new connection. */
+    case object Reject extends Admission
+  }
+  // @formatter:on
+
 }
 

### eclair-core/src/main/scala/fr/acinq/eclair/remote/EclairInternalsSerializer.scala
@@ -127,7 +127,10 @@ object EclairInternalsSerializer {
       ("maxOnionMessagesPerSecond" | int32) ::
       ("maxGossipQueriesPerSecond" | int32) ::
       ("sendRemoteAddressInit" | bool(8)) ::
-      ("maxNoChannels" | int32)).as[PeerConnection.Conf]
+      ("maxNoChannels" | int32) ::
+      ("maxPendingIncomingConnections" | int32) ::
+      ("pendingConnectionMinAge" | finiteDurationCodec) ::
+      ("pendingConnectionAcceptDelay" | finiteDurationCodec)).as[PeerConnection.Conf]
 
   val peerConnectionDoSyncCodec: Codec[PeerConnection.DoSync] = bool(8).as[PeerConnection.DoSync]
 
@@ -136,6 +139,7 @@ object EclairInternalsSerializer {
     .typecase(1, provide(PeerConnection.KillReason.NoRemainingChannel))
     .typecase(2, provide(PeerConnection.KillReason.AllChannelsFail))
     .typecase(3, provide(PeerConnection.KillReason.ConnectionReplaced))
+    .typecase(4, provide(PeerConnection.KillReason.TooManyPendingConnections))
 
   val peerConnectionKillCodec: Codec[PeerConnection.Kill] = peerConnectionKillReasonCodec.as[PeerConnection.Kill]
 

### eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala
@@ -210,6 +210,9 @@ object TestConstants {
         maxGossipQueriesPerSecond = 10,
         sendRemoteAddressInit = true,
         maxNoChannels = 250,
+        maxPendingIncomingConnections = 100,
+        pendingConnectionMinAge = 1 second,
+        pendingConnectionAcceptDelay = 100 millis,
       ),
       routerConf = RouterConf(
         watchSpentWindow = 1 second,
@@ -436,6 +439,9 @@ object TestConstants {
         maxGossipQueriesPerSecond = 10,
         sendRemoteAddressInit = true,
         maxNoChannels = 250,
+        maxPendingIncomingConnections = 100,
+        pendingConnectionMinAge = 1 second,
+        pendingConnectionAcceptDelay = 100 millis,
       ),
       routerConf = RouterConf(
         watchSpentWindow = 1 second,

### eclair-core/src/test/scala/fr/acinq/eclair/io/PeerConnectionSpec.scala
@@ -130,6 +130,30 @@ class PeerConnectionSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike wi
     origin.expectMsg(PeerConnection.ConnectionResult.AuthenticationFailed("authentication timed out"))
   }
 
+  test("disconnect if connection is never initialized") { f =>
+    import f._
+    val probe = TestProbe()
+    val origin = TestProbe()
+    probe.watch(peerConnection)
+    probe.send(peerConnection, PeerConnection.PendingAuth(connection.ref, Some(remoteNodeId), address, origin_opt = Some(origin.ref), transport_opt = Some(transport.ref), isPersistent = true))
+    transport.send(peerConnection, TransportHandler.HandshakeCompleted(remoteNodeId))
+    switchboard.expectMsg(PeerConnection.Authenticated(peerConnection, remoteNodeId, outgoing = true))
+    // If we never receive InitializeConnection from the switchboard, we eventually stop ourselves.
+    assert(peerConnection.stateName == PeerConnection.BEFORE_INIT)
+    probe.expectTerminated(peerConnection)
+    origin.expectMsg(PeerConnection.ConnectionResult.InitializationFailed("connection was never initialized"))
+  }
+
+  test("notify auth tracker when the handshake completes") { f =>
+    import f._
+    val probe = TestProbe()
+    val authTracker = TestProbe()
+    probe.send(peerConnection, PeerConnection.PendingAuth(connection.ref, Some(remoteNodeId), address, origin_opt = None, transport_opt = Some(transport.ref), isPersistent = true, authTracker_opt = Some(authTracker.ref)))
+    authTracker.expectNoMessage(100 millis)
+    transport.send(peerConnection, TransportHandler.HandshakeCompleted(remoteNodeId))
+    authTracker.expectMsg(PeerConnection.Authenticated(peerConnection, remoteNodeId, outgoing = true))
+  }
+
   test("disconnect if init timeout") { f =>
     import f._
     val probe = TestProbe()

### eclair-core/src/test/scala/fr/acinq/eclair/io/ServerSpec.scala
@@ -0,0 +1,202 @@
+/*
+ * Copyright 2026 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.io
+
+import akka.Done
+import akka.actor.ActorRef
+import akka.testkit.TestProbe
+import fr.acinq.eclair.{TestConstants, TestKitBaseClass, TestUtils, randomKey}
+import org.scalatest.concurrent.Eventually
+import org.scalatest.funsuite.AnyFunSuiteLike
+
+import java.io.IOException
+import java.net.InetSocketAddress
+import java.nio.ByteBuffer
+import java.nio.channels.SocketChannel
+import scala.concurrent.duration._
+import scala.concurrent.{Await, Promise}
+
+/**
+ * We drive the [[Server]] through real TCP connections: this lets us verify what peers actually observe (in
+ * particular whether their connection was closed), which is the behavior we care about.
+ */
+class ServerSpec extends TestKitBaseClass with AnyFunSuiteLike with Eventually {
+
+  override implicit val patienceConfig: PatienceConfig = PatienceConfig(timeout = 30 seconds, interval = 100 millis)
+
+  private val nodeParams = TestConstants.Alice.nodeParams
+
+  private def defaultConf: PeerConnection.Conf = nodeParams.peerConnectionConf.copy(
+    maxPendingIncomingConnections = 2,
+    pendingConnectionMinAge = 1 minute,
+    pendingConnectionAcceptDelay = 10 millis,
+  )
+
+  private def withServer(conf: PeerConnection.Conf)(f: (ActorRef, Int) => Unit): Unit = {
+    val port = TestUtils.availablePort
+    val bound = Promise[Done]()
+    val server = system.actorOf(Server.props(nodeParams.keyPair, conf, TestProbe().ref, TestProbe().ref, new InetSocketAddress("127.0.0.1", port), Some(bound)))
+    Await.result(bound.future, 10 seconds)
+    try {
+      f(server, port)
+    } finally {
+      system.stop(server)
+    }
+  }
+
+  /** Open a real TCP connection to our server, without sending anything: we never complete the BOLT 8 handshake. */
+  private def connect(port: Int): SocketChannel = {
+    val channel = SocketChannel.open(new InetSocketAddress("127.0.0.1", port))
+    channel.configureBlocking(false)
+    channel
+  }
+
+  /** Whether our peer closed the connection: `read` returns -1 after a TCP FIN and throws after a TCP RST. */
+  private def isDisconnected(channel: SocketChannel): Boolean = {
+    try {
+      channel.read(ByteBuffer.allocate(16)) < 0
+    } catch {
+      case _: IOException => true
+    }
+  }
+
+  private def pendingConnections(server: ActorRef): Set[ActorRef] = {
+    val probe = TestProbe()
+    probe.send(server, Server.GetPendingConnections(probe.ref))
+    probe.expectMsgType[Server.PendingConnections].peerConnections
+  }
+
+  /** Connect and wait until the server has accepted that connection. */
+  private def connectAndWait(server: ActorRef, port: Int, expectedPending: Int): (SocketChannel, Set[ActorRef]) = {
+    val channel = connect(port)
+    eventually {
+      assert(pendingConnections(server).size == expectedPending)
+    }
+    (channel, pendingConnections(server))
+  }
+
+  test("track incoming connections that haven't authenticated") {
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 3)) { (server, port) =>
+      assert(pendingConnections(server).isEmpty)
+      val (_, pending1) = connectAndWait(server, port, 1)
+      val (_, pending2) = connectAndWait(server, port, 2)
+      assert(pending1.subsetOf(pending2))
+    }
+  }
+
+  test("drop the oldest pending connection when reaching the limit") {
+    // We disable the grace period: pending connections can be dropped immediately.
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 2, pendingConnectionMinAge = 0 millis)) { (server, port) =>
+      val (channel1, pending1) = connectAndWait(server, port, 1)
+      val oldest = pending1.head
+      val (channel2, _) = connectAndWait(server, port, 2)
+      val probe = TestProbe()
+      probe.watch(oldest)
+      // We're at capacity: this connection evicts the oldest pending one.
+      val channel3 = connect(port)
+      probe.expectTerminated(oldest, 10 seconds)
+      eventually {
+        val pending = pendingConnections(server)
+        assert(pending.size == 2)
+        assert(!pending.contains(oldest))
+      }
+      // The peer whose connection was dropped sees it closed, the two others are still connected.
+      eventually {
+        assert(isDisconnected(channel1))
+      }
+      assert(!isDisconnected(channel2))
+      assert(!isDisconnected(channel3))
+    }
+  }
+
+  test("reject incoming connections when every pending connection is too recent") {
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 2, pendingConnectionMinAge = 1 minute)) { (server, port) =>
+      val (channel1, _) = connectAndWait(server, port, 1)
+      val (channel2, pending) = connectAndWait(server, port, 2)
+      // We're at capacity and no pending connection is old enough to be dropped: we reject the incoming connection
+      // instead, which protects the peers that are already busy authenticating.
+      val channel3 = connect(port)
+      eventually {
+        assert(isDisconnected(channel3))
+      }
+      assert(pendingConnections(server) == pending)
+      assert(!isDisconnected(channel1))
+      assert(!isDisconnected(channel2))
+    }
+  }
+
+  test("free up a slot when a connection is authenticated") {
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 1, pendingConnectionMinAge = 1 minute)) { (server, port) =>
+      val (channel1, pending) = connectAndWait(server, port, 1)
+      val probe = TestProbe()
+      // Once authenticated, a connection doesn't count towards the limit anymore.
+      probe.send(server, PeerConnection.Authenticated(pending.head, randomKey().publicKey, outgoing = false))
+      eventually {
+        assert(pendingConnections(server).isEmpty)
+      }
+      // We can thus accept another incoming connection, even though the previous one is still alive.
+      val (channel2, _) = connectAndWait(server, port, 1)
+      assert(!isDisconnected(channel1))
+      assert(!isDisconnected(channel2))
+    }
+  }
+
+  test("free up a slot when a pending connection dies") {
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 1, pendingConnectionMinAge = 1 minute)) { (server, port) =>
+      val (channel1, _) = connectAndWait(server, port, 1)
+      channel1.close()
+      eventually {
+        assert(pendingConnections(server).isEmpty)
+      }
+      val (channel2, _) = connectAndWait(server, port, 1)
+      assert(!isDisconnected(channel2))
+    }
+  }
+
+  test("accept all incoming connections when the limit is disabled") {
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 0)) { (server, port) =>
+      val channels = (1 to 4).map(_ => connect(port))
+      eventually {
+        assert(pendingConnections(server).size == 4)
+      }
+      channels.foreach(channel => assert(!isDisconnected(channel)))
+    }
+  }
+
+  test("delay accepting incoming connections after dropping one") {
+    val acceptDelay = 3.seconds
+    withServer(defaultConf.copy(maxPendingIncomingConnections = 1, pendingConnectionMinAge = 0 millis, pendingConnectionAcceptDelay = acceptDelay)) { (server, port) =>
+      val (_, pending1) = connectAndWait(server, port, 1)
+      // This connection evicts the previous one, after which we stop accepting connections for a while.
+      connect(port)
+      eventually {
+        assert(pendingConnections(server) != pending1)
+      }
+      val pending2 = pendingConnections(server)
+      // The next connection attempt waits in the kernel backlog instead of consuming our resources.
+      connect(port)
+      // NB: expectNoMessage dilates durations by the test time factor, but the server's delay isn't dilated.
+      TestProbe().expectNoMessage((acceptDelay.toMillis * 2 / 3 / testKitSettings.TestTimeFactor).toLong.millis)
+      assert(pendingConnections(server) == pending2)
+      // But we do accept it once the delay has elapsed.
+      eventually {
+        assert(pendingConnections(server) != pending2)
+      }
+    }
+  }
+
+}

### eclair-core/src/test/scala/fr/acinq/eclair/remote/EclairInternalsSerializerSpec.scala
@@ -16,11 +16,34 @@
 
 package fr.acinq.eclair.remote
 
+import fr.acinq.eclair.TestConstants
+import fr.acinq.eclair.io.PeerConnection
 import fr.acinq.eclair.router.Router.GossipDecision
 import org.scalatest.funsuite.AnyFunSuite
 
 class EclairInternalsSerializerSpec extends AnyFunSuite {
 
+  test("codec peer connection conf") {
+    // The peer connection conf is sent to remote frontends: it must round-trip exactly.
+    val conf = TestConstants.Alice.nodeParams.peerConnectionConf
+    val encoded = EclairInternalsSerializer.peerConnectionConfCodec.encode(conf).require
+    assert(EclairInternalsSerializer.peerConnectionConfCodec.decode(encoded).require.value == conf)
+  }
+
+  test("codec peer connection kill reason") {
+    val reasons = Seq(
+      PeerConnection.KillReason.UserRequest,
+      PeerConnection.KillReason.NoRemainingChannel,
+      PeerConnection.KillReason.AllChannelsFail,
+      PeerConnection.KillReason.ConnectionReplaced,
+      PeerConnection.KillReason.TooManyPendingConnections,
+    )
+    reasons.foreach { reason =>
+      val encoded = EclairInternalsSerializer.peerConnectionKillCodec.encode(PeerConnection.Kill(reason)).require
+      assert(EclairInternalsSerializer.peerConnectionKillCodec.decode(encoded).require.value == PeerConnection.Kill(reason))
+    }
+  }
+
   test("canary test codec gossip decision") {
     
     def codec(d: GossipDecision) = d match {