Make swap-in wait for peer ready (#533)

There is a race at startup between the swap-in and the peer connection. If the former wins the race, we drop the message, causing the swap-in to be stuck.

This PR introduces a `waitForPeerReady` helper method which blocks until the peer is fully connected (meaning that existing channels are synced).

---------

Co-authored-by: Bastien Teinturier <31281497+t-bast@users.noreply.github.com>

src/commonMain/kotlin/fr/acinq/lightning/io/Peer.kt

@@ -420,7 +420,10 @@ class Peer(
         logger.info { "starting swap-in watch job" }
         if (swapInJob != null) return
         // wait to have a swap-in feerate available
+        logger.info { "waiting for feerates" }
         swapInFeeratesFlow.filterNotNull().first()
+        logger.info { "waiting for peer to be ready" }
+        waitForPeerReady()
         swapInJob = launch {
             swapInWallet.walletStateFlow.combine(currentTipFlow.filterNotNull()) { walletState, currentTip -> currentTip.first to walletState }
                 .filter { (_, walletState) -> walletState.consistent }
@@ -452,6 +455,23 @@ class Peer(
         input.send(cmd)
     }
 
+    /**
+     * This function blocks until the peer is connected and existing channels have been fully reestablished.
+     */
+    private suspend fun waitForPeerReady() {
+        // In theory we would only need to verify that no channel is in state Offline/Syncing, but there is a corner
+        // case where a channel permanently stays in Syncing, because it is only present locally, and the peer will
+        // never send a channel_reestablish (this happens e.g. due to an error at funding). That is why we consider
+        // the peer ready if "all channels are synced" OR "peer has been connected for 10s".
+        connectionState.first { it is Connection.ESTABLISHED }
+        val result = withTimeoutOrNull(10.seconds) {
+            channelsFlow.first { it.values.all { channel -> channel !is Offline && channel !is Syncing } }
+        }
+        if (result == null) {
+            logger.info { "peer ready timeout elapsed, not all channels are synced but proceeding anyway" }
+        }
+    }
+
     /**
      * Estimate the actual feerate to use (and corresponding fee to pay) in order to reach the target feerate
      * for a splice out, taking into account potential unconfirmed parent splices.
@@ -983,7 +1003,8 @@ class Peer(
             }
             is RequestChannelOpen -> {
                 when (val channel = channels.values.firstOrNull { it is Normal }) {
-                    is ChannelStateWithCommitments -> {
+                    is Normal -> {
+                        // we have a channel and we are connected (otherwise state would be Offline/Syncing)
                         val targetFeerate = swapInFeeratesFlow.filterNotNull().first()
                         val weight = FundingContributions.computeWeightPaid(isInitiator = true, commitment = channel.commitments.active.first(), walletInputs = cmd.walletInputs, localOutputs = emptyList())
                         val (feerate, fee) = watcher.client.computeSpliceCpfpFeerate(channel.commitments, targetFeerate, spliceWeight = weight, logger)