lightningd: wait for onchaind to ack new spends before continuing replay.

Christian noted that if we don't do this we could flood onchaind with messages:
particularly in Greenlight where the HSM (remote) may delay indefinitely, so
onchaind doesn't process messages.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

lightningd/channel.c

@@ -609,6 +609,7 @@ struct channel *new_channel(struct peer *peer, u64 dbid,
 	channel->last_stable_connection = last_stable_connection;
 	channel->stable_conn_timer = NULL;
 	channel->onchaind_replay_watches = NULL;
+	channel->num_onchain_spent_calls = 0;
 	channel->stats = *stats;
 	channel->state_changes = tal_steal(channel, state_changes);
 

lightningd/channel.h

@@ -196,6 +196,10 @@ struct channel {
 
 	/* If we're doing a replay for onchaind, here are the txids it's watching */
 	struct replay_tx_hash *onchaind_replay_watches;
+	/* Number of outstanding onchaind_spent calls */
+	size_t num_onchain_spent_calls;
+	/* Height we're replaying at (if onchaind_replay_watches set) */
+	u32 onchaind_replay_height;
 
 	/* Our original funds, in funding amount */
 	struct amount_sat our_funds;

lightningd/onchain_control.c

@@ -217,6 +217,7 @@ static enum watch_result onchain_tx_watched(struct lightningd *ld,
 
 static void watch_tx_and_outputs(struct channel *channel,
 				 const struct bitcoin_tx *tx);
+static void onchaind_replay(struct channel *channel);
 
 static void replay_unwatch_txid(struct channel *channel,
 				const struct bitcoin_txid *txid)
@@ -236,14 +237,16 @@ static void onchaind_spent_reply(struct subd *onchaind, const u8 *msg,
 		channel_internal_error(channel, "Invalid onchaind_spent_reply %s",
 				       tal_hex(tmpctx, msg));
 
+	channel->num_onchain_spent_calls--;
+
 	/* Only delete watch if it says it doesn't care */
 	if (interested)
-		return;
+		goto out;
 
 	/* If we're doing replay: */
 	if (channel->onchaind_replay_watches) {
 		replay_unwatch_txid(channel, txid);
-		return;
+		goto out;
 	}
 
 	/* Frees the txo watches, too: see watch_tx_and_outputs() */
@@ -253,6 +256,13 @@ static void onchaind_spent_reply(struct subd *onchaind, const u8 *msg,
 		log_unusual(channel->log, "Can't unwatch txid %s",
 			    fmt_bitcoin_txid(tmpctx, txid));
 	tal_free(txw);
+
+out:
+	/* If that's the last request, continue asking for blocks */
+	if (channel->onchaind_replay_watches
+	    && channel->num_onchain_spent_calls == 0) {
+		onchaind_replay(channel);
+	}
 }
 
 /**
@@ -276,7 +286,7 @@ static void onchain_txo_spent(struct channel *channel, const struct bitcoin_tx *
 	msg = towire_onchaind_spent(channel, parts, input_num, blockheight);
 	subd_req(channel->owner, channel->owner, take(msg), -1, 0,
 		 onchaind_spent_reply, take(txid));
-
+	channel->num_onchain_spent_calls++;
 }
 
 /**
@@ -405,8 +415,24 @@ static void replay_block(struct bitcoind *bitcoind,
 		return;
 	}
 
-	/* Otherwise, loop on next block. */
+	/* Ready for next block */
-	bitcoind_getrawblockbyheight(channel, bitcoind, height + 1, replay_block, channel);
+	channel->onchaind_replay_height = height + 1;
+
+	/* Otherwise, wait for those to be resolved (in case onchaind is slow,
+	 * e.g. waiting for HSM). */
+	if (channel->num_onchain_spent_calls == 0)
+		onchaind_replay(channel);
+}
+
+static void onchaind_replay(struct channel *channel)
+{
+	assert(channel->onchaind_replay_watches);
+	assert(channel->num_onchain_spent_calls == 0);
+
+	bitcoind_getrawblockbyheight(channel,
+				     channel->peer->ld->topology->bitcoind,
+				     channel->onchaind_replay_height,
+				     replay_block, channel);
 }
 
 static void handle_extracted_preimage(struct channel *channel, const u8 *msg)
@@ -1824,12 +1850,11 @@ void onchaind_replay_channels(struct lightningd *ld)
 
 			/* We're in replay mode */
 			channel->onchaind_replay_watches = tal(channel, struct replay_tx_hash);
+			channel->onchaind_replay_height = blockheight;
 			replay_tx_hash_init(channel->onchaind_replay_watches);
 
 			onchaind_funding_spent(channel, tx, blockheight);
-			/* Ask bitcoind to start grabbing those blocks for replay */
+			onchaind_replay(channel);
-			bitcoind_getrawblockbyheight(channel, ld->topology->bitcoind, blockheight,
-						     replay_block, channel);
 		}
 	}
 	db_commit_transaction(ld->wallet->db);

onchaind/onchaind.c

@@ -1195,7 +1195,7 @@ static bool output_spent(struct tracked_output ***outs,
 			 u32 input_num,
 			 u32 tx_blockheight)
 {
-	bool interesting;
+	bool interesting = false;
 
 	for (size_t i = 0; i < tal_count(*outs); i++) {
 		struct tracked_output *out = (*outs)[i];
@@ -1220,7 +1220,7 @@ static bool output_spent(struct tracked_output ***outs,
 
 			record_coin_movements(out, tx_blockheight,
 					      &tx_parts->txid);
-			return interesting;
+			break;
 		}
 
 		htlc_outpoint.txid = tx_parts->txid;