Store onchaind state in the database and reduce rescan on startup

lightningd/chaintopology.c

@@ -59,8 +59,10 @@ static void filter_block_txs(struct chain_topology *topo, struct block *b)
 			out.index = tx->input[j].index;
 
 			txo = txowatch_hash_get(&topo->txowatches, &out);
-			if (txo)
+			if (txo) {
+				wallet_transaction_add(topo->wallet, tx, b->height, i);
 				txowatch_fire(txo, tx, j, b);
+			}
 		}
 
 		satoshi_owned = 0;
@@ -517,17 +519,18 @@ static void get_init_block(struct bitcoind *bitcoind,
 static void get_init_blockhash(struct bitcoind *bitcoind, u32 blockcount,
 			       struct chain_topology *topo)
 {
-	/* This can happen if bitcoind still syncing, or first_blocknum is MAX */
-	if (blockcount < topo->first_blocknum)
-		topo->first_blocknum = blockcount;
-
-	/* For fork protection (esp. because we don't handle our own first
-	 * block getting reorged out), we always go 100 blocks further back
-	 * than we need. */
-	if (topo->first_blocknum < 100)
-		topo->first_blocknum = 0;
-	else
-		topo->first_blocknum -= 100;
+	/* If bitcoind's current blockheight is below the requested height, just
+	 * go back to that height. This might be a new node catching up, or
+	 * bitcoind is processing a reorg. */
+	if (blockcount < topo->first_blocknum) {
+		if (bitcoind->ld->config.rescan < 0) {
+			/* Absolute blockheight, but bitcoind's blockheight isn't there yet */
+			topo->first_blocknum = blockcount - 1;
+		} else if (topo->first_blocknum == UINT32_MAX) {
+			/* Relative rescan, but we didn't know the blockheight */
+			topo->first_blocknum = blockcount - bitcoind->ld->config.rescan;
+		}
+	}
 
 	/* Rollback to the given blockheight, so we start track
 	 * correctly again */

lightningd/channel.c

@@ -265,6 +265,19 @@ struct channel *active_channel_by_id(struct lightningd *ld,
 	return peer_active_channel(peer);
 }
 
+struct channel *channel_by_dbid(struct lightningd *ld, const u64 dbid)
+{
+	struct peer *p;
+	struct channel *chan;
+	list_for_each(&ld->peers, p, list) {
+		list_for_each(&p->channels, chan, list) {
+			if (chan->dbid == dbid)
+				return chan;
+		}
+	}
+	return NULL;
+}
+
 void channel_set_last_tx(struct channel *channel,
 			 struct bitcoin_tx *tx,
 			 const secp256k1_ecdsa_signature *sig)

lightningd/channel.h

@@ -165,6 +165,8 @@ struct channel *active_channel_by_id(struct lightningd *ld,
 				     const struct pubkey *id,
 				     struct uncommitted_channel **uc);
 
+struct channel *channel_by_dbid(struct lightningd *ld, const u64 dbid);
+
 void channel_set_last_tx(struct channel *channel,
 			 struct bitcoin_tx *tx,
 			 const secp256k1_ecdsa_signature *sig);

lightningd/lightningd.c

@@ -30,6 +30,7 @@
 #include <lightningd/invoice.h>
 #include <lightningd/jsonrpc.h>
 #include <lightningd/log.h>
+#include <lightningd/onchain_control.h>
 #include <lightningd/options.h>
 #include <onchaind/onchain_wire.h>
 #include <signal.h>
@@ -275,7 +276,7 @@ int main(int argc, char *argv[])
 {
 	struct lightningd *ld;
 	bool newdir;
-	u32 first_blocknum;
+	u32 blockheight;
 
 	daemon_setup(argv[0], log_backtrace_print, log_backtrace_exit);
 	ld = new_lightningd(NULL);
@@ -350,18 +351,31 @@ int main(int argc, char *argv[])
 	if (!wallet_htlcs_reconnect(ld->wallet, &ld->htlcs_in, &ld->htlcs_out))
 		fatal("could not reconnect htlcs loaded from wallet, wallet may be inconsistent.");
 
-	/* Worst case, scan back to the first lightning deployment */
-	first_blocknum = wallet_first_blocknum(ld->wallet,
-					       get_chainparams(ld)
-					       ->when_lightning_became_cool);
+	/* Get the blockheight we are currently at, UINT32_MAX is used to signal
+	 * an unitialized wallet and that we should start off of bitcoind's
+	 * current height */
+	blockheight = wallet_blocks_height(ld->wallet, UINT32_MAX);
+
+	/* If we were asked to rescan from an absolute height (--rescan < 0)
+	 * then just go there. Otherwise take compute the diff to our current
+	 * height, lowerbounded by 0. */
+	if (ld->config.rescan < 0)
+		blockheight = -ld->config.rescan;
+	else if (blockheight < (u32)ld->config.rescan)
+		blockheight = 0;
+	else if (blockheight != UINT32_MAX)
+		blockheight -= ld->config.rescan;
 
 	db_commit_transaction(ld->wallet->db);
 
 	/* Initialize block topology (does its own transaction) */
 	setup_topology(ld->topology,
 		       &ld->timers,
 		       ld->config.poll_time,
-		       first_blocknum);
+		       blockheight);
+
+	/* Replay transactions for all running onchainds */
+	onchaind_replay_channels(ld);
 
 	/* Create RPC socket (if any) */
 	setup_jsonrpc(ld, ld->rpc_filename);

lightningd/lightningd.h

@@ -72,6 +72,10 @@ struct config {
 
 	/* Do we let the funder set any fee rate they want */
 	bool ignore_fee_limits;
+
+	/* Number of blocks to rescan from the current head, or absolute
+	 * blockheight if rescan >= 500'000 */
+	s32 rescan;
 };
 
 struct lightningd {

lightningd/onchain_control.c

@@ -3,13 +3,11 @@
 #include <errno.h>
 #include <inttypes.h>
 #include <lightningd/chaintopology.h>
-#include <lightningd/lightningd.h>
 #include <lightningd/log.h>
 #include <lightningd/onchain_control.h>
 #include <lightningd/peer_control.h>
 #include <lightningd/subd.h>
 #include <lightningd/watch.h>
-#include <onchaind/gen_onchain_wire.h>
 #include <onchaind/onchain_wire.h>
 
 /* We dump all the known preimages when onchaind starts up. */
@@ -47,21 +45,32 @@ static void onchaind_tell_fulfill(struct channel *channel)
 	}
 }
 
-static void handle_onchain_init_reply(struct channel *channel, const u8 *msg)
+static void handle_onchain_init_reply(struct channel *channel, const u8 *msg UNUSED)
 {
 	/* FIXME: We may already be ONCHAIN state when we implement restart! */
 	channel_set_state(channel, FUNDING_SPEND_SEEN, ONCHAIN);
+}
 
-	/* Tell it about any preimages we know. */
-	onchaind_tell_fulfill(channel);
+/**
+ * Notify onchaind about the depth change of the watched tx.
+ */
+static void onchain_tx_depth(struct channel *channel,
+			     const struct bitcoin_txid *txid,
+			     unsigned int depth)
+{
+	u8 *msg;
+	msg = towire_onchain_depth(channel, txid, depth);
+	subd_send_msg(channel->owner, take(msg));
 }
 
+/**
+ * Entrypoint for the txwatch callback, calls onchain_tx_depth.
+ */
 static enum watch_result onchain_tx_watched(struct channel *channel,
 					    const struct bitcoin_txid *txid,
 					    unsigned int depth)
 {
-	u8 *msg;
-
+	u32 blockheight = channel->peer->ld->topology->tip->height;
 	if (depth == 0) {
 		log_unusual(channel->log, "Chain reorganization!");
 		channel_set_owner(channel, NULL);
@@ -74,25 +83,48 @@ static enum watch_result onchain_tx_watched(struct channel *channel,
 		return KEEP_WATCHING;
 	}
 
-	msg = towire_onchain_depth(channel, txid, depth);
-	subd_send_msg(channel->owner, take(msg));
+	/* Store the channeltx so we can replay later */
+	wallet_channeltxs_add(channel->peer->ld->wallet, channel,
+			      WIRE_ONCHAIN_DEPTH, txid, 0, blockheight);
+
+	onchain_tx_depth(channel, txid, depth);
 	return KEEP_WATCHING;
 }
 
 static void watch_tx_and_outputs(struct channel *channel,
 				 const struct bitcoin_tx *tx);
 
+/**
+ * Notify onchaind that an output was spent and register new watches.
+ */
+static void onchain_txo_spent(struct channel *channel, const struct bitcoin_tx *tx, size_t input_num, u32 blockheight)
+{
+	u8 *msg;
+
+	watch_tx_and_outputs(channel, tx);
+
+	msg = towire_onchain_spent(channel, tx, input_num, blockheight);
+	subd_send_msg(channel->owner, take(msg));
+
+}
+
+/**
+ * Entrypoint for the txowatch callback, stores tx and calls onchain_txo_spent.
+ */
 static enum watch_result onchain_txo_watched(struct channel *channel,
 					     const struct bitcoin_tx *tx,
 					     size_t input_num,
 					     const struct block *block)
 {
-	u8 *msg;
+	struct bitcoin_txid txid;
+	bitcoin_txid(tx, &txid);
 
-	watch_tx_and_outputs(channel, tx);
+	/* Store the channeltx so we can replay later */
+	wallet_channeltxs_add(channel->peer->ld->wallet, channel,
+			      WIRE_ONCHAIN_SPENT, &txid, input_num,
+			      block->height);
 
-	msg = towire_onchain_spent(channel, tx, input_num, block->height);
-	subd_send_msg(channel->owner, take(msg));
+	onchain_txo_spent(channel, tx, input_num, block->height);
 
 	/* We don't need to keep watching: If this output is double-spent
 	 * (reorg), we'll get a zero depth cb to onchain_tx_watched, and
@@ -201,7 +233,7 @@ static void handle_onchain_htlc_timeout(struct channel *channel, const u8 *msg)
 	onchain_failed_our_htlc(channel, &htlc, "timed out");
 }
 
-static void handle_irrevocably_resolved(struct channel *channel, const u8 *msg)
+static void handle_irrevocably_resolved(struct channel *channel, const u8 *msg UNUSED)
 {
 	/* FIXME: Implement check_htlcs to ensure no dangling hout->in ptrs! */
 	free_htlcs(channel->peer->ld, channel);
@@ -338,10 +370,9 @@ static void onchain_error(struct channel *channel,
 
 /* With a reorg, this can get called multiple times; each time we'll kill
  * onchaind (like any other owner), and restart */
-enum watch_result funding_spent(struct channel *channel,
-				const struct bitcoin_tx *tx,
-				size_t input_num UNUSED,
-				const struct block *block)
+enum watch_result onchaind_funding_spent(struct channel *channel,
+					 const struct bitcoin_tx *tx,
+					 u32 blockheight)
 {
 	u8 *msg;
 	struct bitcoin_txid our_last_txid;
@@ -411,7 +442,7 @@ enum watch_result funding_spent(struct channel *channel,
 				  &channel->channel_info.theirbase.htlc,
 				  &channel->channel_info.theirbase.delayed_payment,
 				  tx,
-				  block->height,
+				  blockheight,
 				  /* FIXME: config for 'reasonable depth' */
 				  3,
 				  channel->last_htlc_sigs,
@@ -429,8 +460,55 @@ enum watch_result funding_spent(struct channel *channel,
 		subd_send_msg(channel->owner, take(msg));
 	}
 
+	/* Tell it about any preimages we know. */
+	onchaind_tell_fulfill(channel);
+
 	watch_tx_and_outputs(channel, tx);
 
 	/* We keep watching until peer finally deleted, for reorgs. */
 	return KEEP_WATCHING;
 }
+
+void onchaind_replay_channels(struct lightningd *ld)
+{
+	u32 *onchaind_ids;
+	struct channeltx *txs;
+	struct channel *chan;
+
+	db_begin_transaction(ld->wallet->db);
+	onchaind_ids = wallet_onchaind_channels(ld->wallet, ld);
+
+	for (size_t i = 0; i < tal_count(onchaind_ids); i++) {
+		log_info(ld->log, "Restarting onchaind for channel %d",
+			 onchaind_ids[i]);
+
+		txs = wallet_channeltxs_get(ld->wallet, onchaind_ids,
+					    onchaind_ids[i]);
+		chan = channel_by_dbid(ld, onchaind_ids[i]);
+
+		for (size_t j = 0; j < tal_count(txs); j++) {
+			if (txs[j].type == WIRE_ONCHAIN_INIT) {
+				onchaind_funding_spent(chan, txs[j].tx,
+						       txs[j].blockheight);
+
+			} else if (txs[j].type == WIRE_ONCHAIN_SPENT) {
+				onchain_txo_spent(chan, txs[j].tx,
+						  txs[j].input_num,
+						  txs[j].blockheight);
+
+			} else if (txs[j].type == WIRE_ONCHAIN_DEPTH) {
+				onchain_tx_depth(chan, &txs[j].txid,
+						 txs[j].depth);
+
+			} else {
+				fatal("unknown message of type %d during "
+				      "onchaind replay",
+				      txs[j].type);
+			}
+		}
+		tal_free(txs);
+	}
+	tal_free(onchaind_ids);
+
+	db_commit_transaction(ld->wallet->db);
+}

lightningd/onchain_control.h

@@ -2,14 +2,17 @@
 #define LIGHTNING_LIGHTNINGD_ONCHAIN_CONTROL_H
 #include "config.h"
 #include <ccan/short_types/short_types.h>
+#include <lightningd/lightningd.h>
+#include <onchaind/gen_onchain_wire.h>
 
 struct channel;
 struct bitcoin_tx;
 struct block;
 
-enum watch_result funding_spent(struct channel *channel,
-				const struct bitcoin_tx *tx,
-				size_t input_num,
-				const struct block *block);
+enum watch_result onchaind_funding_spent(struct channel *channel,
+					 const struct bitcoin_tx *tx,
+					 u32 blockheight);
+
+void onchaind_replay_channels(struct lightningd *ld);
 
 #endif /* LIGHTNING_LIGHTNINGD_ONCHAIN_CONTROL_H */

lightningd/options.c

@@ -309,6 +309,10 @@ static void config_register_opts(struct lightningd *ld)
 	opt_register_arg("--fee-base", opt_set_u32, opt_show_u32,
 			 &ld->config.fee_base,
 			 "Millisatoshi minimum to charge for HTLC");
+	opt_register_arg("--rescan", opt_set_s32, opt_show_s32,
+			 &ld->config.rescan,
+			 "Number of blocks to rescan from the current head, or "
+			 "absolute blockheight if negative");
 	opt_register_arg("--fee-per-satoshi", opt_set_s32, opt_show_s32,
 			 &ld->config.fee_per_satoshi,
 			 "Microsatoshi fee for every satoshi in HTLC");
@@ -420,6 +424,9 @@ static const struct config testnet_config = {
 
 	/* Testnet sucks */
 	.ignore_fee_limits = true,
+
+	/* Rescan 5 hours of blocks on testnet, it's reorg happy */
+	.rescan = 30,
 };
 
 /* aka. "Dude, where's my coins?" */
@@ -481,6 +488,9 @@ static const struct config mainnet_config = {
 
 	/* Mainnet should have more stable fees */
 	.ignore_fee_limits = false,
+
+	/* Rescan 2.5 hours of blocks on startup, it's not so reorg happy */
+	.rescan = 15,
 };
 
 static void check_config(struct lightningd *ld)