Sangbida/bwatch wallet

lightningd/lightningd.c

@@ -46,6 +46,7 @@
 
 /*~ This is common code: routines shared by one or more executables
  *  (separate daemons, or the lightning-cli program). */
+#include <common/configvar.h>
 #include <common/daemon.h>
 #include <common/deprecation.h>
 #include <common/ecdh_hsmd.h>
@@ -78,6 +79,7 @@
 #include <lightningd/watchman.h>
 #include <sys/resource.h>
 #include <wallet/invoices.h>
+#include <wallet/wallet.h>
 #include <wally_bip32.h>
 
 static void destroy_alt_subdaemons(struct lightningd *ld);
@@ -288,6 +290,11 @@ static struct lightningd *new_lightningd(const tal_t *ctx)
 	 * so set it to NULL explicitly now. */
 	ld->wallet = NULL;
 
+	/*~ Only created if the user opts into --experimental-bwatch, but
+	 * plugin startup (watchman_notify_plugin_ready) examines it, so set
+	 * it to NULL explicitly now. */
+	ld->watchman = NULL;
+
 	/*~ Behavioral options */
 	ld->accept_extra_tlv_types = tal_arr(ld, u64, 0);
 
@@ -1164,6 +1171,18 @@ static void setup_fd_limit(struct lightningd *ld, size_t num_channels)
 	}
 }
 
+/*~ Has the user opted into the experimental bwatch chain watcher?  The
+ * --experimental-bwatch flag is registered by the bwatch plugin, not by
+ * lightningd, so we look for it in the parsed configvars rather than
+ * keeping our own copy. */
+static bool bwatch_enabled(const struct lightningd *ld)
+{
+	const char **names = tal_arr(tmpctx, const char *, 1);
+
+	names[0] = "experimental-bwatch";
+	return configvar_first(ld->configvars, names) != NULL;
+}
+
 int main(int argc, char *argv[])
 {
 	struct lightningd *ld;
@@ -1349,8 +1368,28 @@ int main(int argc, char *argv[])
 	trace_span_end(ld->topology);
 
 	/*~ Stand up the watchman: it queues bwatch RPC requests until the
-	 * bwatch plugin reports ready, then replays them. */
-	ld->watchman = watchman_new(ld, ld);
+	 * bwatch plugin reports ready, then replays them.  Must come before
+	 * init_wallet_scriptpubkey_watches so the watches have somewhere to
+	 * enqueue, and after setup_topology so start_block reflects the
+	 * last-processed height.
+	 *
+	 * bwatch is opt-in for now: the --experimental-bwatch flag is
+	 * registered by the bwatch plugin, so peek at the configvar to gate
+	 * the lightningd side too.  Without it, ld->watchman stays NULL and
+	 * the watchman_* entry points are no-ops, leaving chain_topology as
+	 * the only chain watcher. */
+	if (bwatch_enabled(ld)) {
+		ld->watchman = watchman_new(ld, ld);
+
+		/*~ Reads intvars and the addresses table, so needs a
+		 * transaction (the datastore writes it triggers self-wrap
+		 * when necessary). */
+		db_begin_transaction(ld->wallet->db);
+		trace_span_start("init_wallet_scriptpubkey_watches", ld->wallet);
+		init_wallet_scriptpubkey_watches(ld->wallet);
+		trace_span_end(ld->wallet);
+		db_commit_transaction(ld->wallet->db);
+	}
 
 	db_begin_transaction(ld->wallet->db);
 	trace_span_start("delete_old_htlcs", ld->wallet);

lightningd/watchman.c

@@ -4,7 +4,6 @@
 #include <bitcoin/short_channel_id.h>
 #include <ccan/array_size/array_size.h>
 #include <ccan/str/str.h>
-#include <ccan/tal/str/str.h>
 #include <common/autodata.h>
 #include <common/json_command.h>
 #include <common/json_param.h>
@@ -38,7 +37,7 @@
 
 /* A pending operation - method and params to send to bwatch */
 struct pending_op {
-	/* "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/p2wpkh/42".
+	/* "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/spk/42/p2tr".
 	 * Method and owner are recoverable from this without a separate field. */
 	const char *op_id;
 	const char *json_params; /* JSON params to send to bwatch */
@@ -181,7 +180,7 @@ struct watchman *watchman_new(const tal_t *ctx, struct lightningd *ld)
  * callback never needs to parse the JSON-RPC response id. */
 struct bwatch_ack_arg {
 	struct watchman *wm;
-	const char *op_id; /* "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/p2wpkh/42" */
+	const char *op_id; /* "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/spk/42/p2tr" */
 };
 
 /* Response callback for bwatch RPC requests; handles both success and error. */
@@ -218,7 +217,7 @@ static const char *method_from_op_id(const tal_t *ctx, const char *op_id)
 }
 
 /* Send an RPC request to the bwatch plugin.
- * op_id must be "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/p2wpkh/42". */
+ * op_id must be "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/spk/42/p2tr". */
 static void send_to_bwatch(struct watchman *wm, const char *method,
 			   const char *op_id, const char *json_params)
 {
@@ -287,7 +286,13 @@ static void watchman_add(struct lightningd *ld, const char *method,
 			 const char *owner, const char *json_params)
 {
 	struct watchman *wm = ld->watchman;
-	char *op_id = tal_fmt(tmpctx, "%s:%s", method, owner);
+	char *op_id;
+
+	/* No-op unless --experimental-bwatch stood up the watchman. */
+	if (!wm)
+		return;
+
+	op_id = tal_fmt(tmpctx, "%s:%s", method, owner);
 
 	/* Remove any existing add for this owner */
 	watchman_ack(ld, op_id);
@@ -305,12 +310,18 @@ static void watchman_del(struct lightningd *ld, const char *method,
 			 const char *owner, const char *json_params)
 {
 	struct watchman *wm = ld->watchman;
-	char *op_id = tal_fmt(tmpctx, "%s:%s", method, owner);
+	char *op_id, *add_op_id;
+
+	/* No-op unless --experimental-bwatch stood up the watchman. */
+	if (!wm)
+		return;
+
+	op_id = tal_fmt(tmpctx, "%s:%s", method, owner);
 
 	/* Cancel any pending add for this owner.  All del-methods are named
 	 * "del<suffix>" and their paired add-method is "add<suffix>" */
 	assert(strstarts(method, "del"));
-	char *add_op_id = tal_fmt(tmpctx, "add%s:%s", method + strlen("del"), owner);
+	add_op_id = tal_fmt(tmpctx, "add%s:%s", method + strlen("del"), owner);
 	watchman_ack(ld, add_op_id);
 	enqueue_op(wm, method, op_id, json_params);
 }
@@ -320,13 +331,16 @@ static void watchman_del(struct lightningd *ld, const char *method,
  *
  * Called when bwatch confirms it has processed an add/del operation.
  * Removes the operation from the pending queue and datastore.
- * op_id must be the bare stored id (e.g. "add:wallet/p2wpkh/0"), not the
- * full JSON-RPC response id.
+ * op_id must be the bare stored id (e.g. "addscriptpubkeywatch:wallet/spk/0/p2wpkh"),
+ * not the full JSON-RPC response id.
  */
 void watchman_ack(struct lightningd *ld, const char *op_id)
 {
 	struct watchman *wm = ld->watchman;
 
+	if (!wm)
+		return;
+
 	for (size_t i = 0; i < tal_count(wm->pending_ops); i++) {
 		if (streq(wm->pending_ops[i]->op_id, op_id)) {
 			db_remove(wm, op_id);
@@ -347,6 +361,9 @@ void watchman_replay_pending(struct lightningd *ld)
 {
 	struct watchman *wm = ld->watchman;
 
+	if (!wm)
+		return;
+
 	for (size_t i = 0; i < tal_count(wm->pending_ops); i++) {
 		struct pending_op *op = wm->pending_ops[i];
 		send_to_bwatch(wm, method_from_op_id(tmpctx, op->op_id),
@@ -466,7 +483,13 @@ static const struct watch_dispatch {
 	watch_found_fn handler;
 	watch_revert_fn revert;
 } watch_handlers[] = {
-	/* Entries added in subsequent commits alongside their handler functions. */
+	/* wallet/utxo/<txid>:<outnum>: WATCH_OUTPOINT, fires when a wallet UTXO is spent */
+	{ "wallet/utxo/", wallet_utxo_spent_watch_found, wallet_utxo_spent_watch_revert },
+	/* wallet/spk/<keyidx>/<form>: WATCH_SCRIPTPUBKEY, fires when an
+	 * address form (p2wpkh/p2tr/p2sh_p2wpkh) of this HD key receives
+	 * funds.  One dispatch entry serves all forms: the handler parses the
+	 * keyindex and recovers the form from the matched output script. */
+	{ "wallet/spk/", wallet_watch_spk, wallet_scriptpubkey_watch_revert },
 	{ NULL, NULL, NULL },
 };
 
@@ -801,6 +824,9 @@ static struct command_result *json_chaininfo(struct command *cmd,
 		   NULL))
 		return command_param_failed();
 
+	if (!cmd->ld->watchman)
+		return command_fail(cmd, LIGHTNINGD, "Watchman not initialized");
+
 	if (!streq(chain, chainparams->bip70_name))
 		fatal("Wrong network! Our Bitcoin backend is running on '%s',"
 		      " but we expect '%s'.", chain, chainparams->bip70_name);

lightningd/watchman.h

@@ -3,7 +3,9 @@
 
 #include "config.h"
 #include <bitcoin/tx.h>
+#include <ccan/tal/str/str.h>
 #include <ccan/tal/tal.h>
+#include <inttypes.h>
 
 struct lightningd;
 struct pending_op;
@@ -25,9 +27,10 @@ struct watchman {
 /**
  * watch_found_fn - Handler for watch_found notifications (tx-based watches)
  * @ld: lightningd instance
- * @suffix: the owner string after the prefix (e.g. "42" for wallet/p2wpkh/42,
- *          or "100x1x0" for gossip/100x1x0); the handler is responsible for
- *          parsing whatever identifier it stored in that suffix
+ * @suffix: the owner string after the prefix (e.g. "42/p2tr" for
+ *          wallet/spk/42/p2tr, or "100x1x0" for gossip/100x1x0); the handler
+ *          is responsible for parsing whatever identifier it stored in that
+ *          suffix
  * @tx: the transaction that matched
  * @outnum: which output matched (for scriptpubkey watches) or input for outpoint watches
  * @blockheight: the block height where tx was found
@@ -144,4 +147,25 @@ void watchman_unwatch_blockdepth(struct lightningd *ld,
 				 const char *owner,
 				 u32 confirm_height);
 
+/*
+ * Owner string constructors.
+ *
+ * Always use these instead of raw tal_fmt() to build owner strings.  Sharing
+ * one constructor between watchman_watch_* and watchman_unwatch_* guarantees
+ * the strings are identical and the unwatch can never silently fail due to a
+ * format mismatch (e.g. %u vs PRIu64).
+ */
+
+/* wallet/ owners */
+static inline const char *owner_wallet_utxo(const tal_t *ctx,
+					    const struct bitcoin_outpoint *op)
+{ return tal_fmt(ctx, "wallet/utxo/%s", fmt_bitcoin_outpoint(ctx, op)); }
+
+/* One owner per (HD keyindex, address form) pair, e.g. "wallet/spk/42/p2tr".
+ * A single key can be watched in several forms at once (p2wpkh, p2tr and
+ * legacy p2sh-p2wpkh)*/
+static inline const char *owner_wallet_spk(const tal_t *ctx, u64 keyidx,
+					   const char *form)
+{ return tal_fmt(ctx, "wallet/spk/%"PRIu64"/%s", keyidx, form); }
+
 #endif /* LIGHTNING_LIGHTNINGD_WATCHMAN_H */

tests/test_connection.py

@@ -1038,18 +1038,18 @@ def test_funding_change(node_factory, bitcoind):
     bitcoind.generate_block(1)
     sync_blockheight(bitcoind, [l1])
 
-    outputs = l1.db_query('SELECT value FROM outputs WHERE status=0;')
+    outputs = l1.db_query('SELECT satoshis AS value FROM our_outputs WHERE spendheight IS NULL AND reserved_til = 0;')
     assert only_one(outputs)['value'] == 10000000
 
     l1.rpc.fundchannel(l2.info['id'], 1000000)
     bitcoind.generate_block(1, wait_for_mempool=1)
     sync_blockheight(bitcoind, [l1])
-    outputs = {r['status']: r['value'] for r in l1.db_query(
-        'SELECT status, SUM(value) AS value FROM outputs GROUP BY status;')}
 
     # The 10m out is spent and we have a change output of 9m-fee
-    assert outputs[0] > 8990000
-    assert outputs[2] == 10000000
+    spent = l1.db_query('SELECT SUM(satoshis) AS value FROM our_outputs WHERE spendheight IS NOT NULL;')
+    unspent = l1.db_query('SELECT SUM(satoshis) AS value FROM our_outputs WHERE spendheight IS NULL;')
+    assert only_one(unspent)['value'] > 8990000
+    assert only_one(spent)['value'] == 10000000
 
 
 @pytest.mark.openchannel('v1')
@@ -1063,13 +1063,13 @@ def test_funding_all(node_factory, bitcoind):
     bitcoind.generate_block(1)
     sync_blockheight(bitcoind, [l1])
 
-    outputs = l1.db_query('SELECT value FROM outputs WHERE status=0;')
+    outputs = l1.db_query('SELECT satoshis AS value FROM our_outputs WHERE spendheight IS NULL AND reserved_til = 0;')
     assert only_one(outputs)['value'] == 10000000
 
     l1.rpc.fundchannel(l2.info['id'], "all")
 
     # Keeps emergency reserve!
-    outputs = l1.db_query('SELECT value FROM outputs WHERE status=0;')
+    outputs = l1.db_query('SELECT satoshis AS value FROM our_outputs WHERE spendheight IS NULL AND reserved_til = 0;')
     if 'anchors/even' in only_one(l1.rpc.listpeerchannels()['channels'])['channel_type']['names']:
         assert outputs == [{'value': 25000}]
     else:

tests/test_db.py

@@ -233,6 +233,60 @@ def test_last_tx_psbt_upgrade(node_factory, bitcoind):
     bitcoind.rpc.decoderawtransaction(last_txs[1].hex())
 
 
+@unittest.skipIf(os.getenv('TEST_DB_PROVIDER', 'sqlite3') != 'sqlite3', "This test is based on a sqlite3 snapshot")
+@unittest.skipIf(TEST_NETWORK != 'regtest', "The network must match the DB snapshot")
+def test_our_tables_backfill_upgrade(node_factory, bitcoind):
+    """Upgrading a pre-bwatch db backfills our_outputs/our_txs from the
+    legacy outputs/transactions tables, and the wallet stays usable."""
+    bitcoind.generate_block(1)
+    l1 = node_factory.get_node(dbfile="l1-before-moves-in-db.sqlite3.xz",
+                               options={'database-upgrade': True},
+                               old_hsmsecret=True)
+
+    # The legacy tables were mirrored into the new bwatch tables.  The
+    # fixture's chain doesn't exist on this bitcoind, so heights past the
+    # common tip get reorged away at startup: that must hit both old and
+    # new tables in lockstep, so compare them (our_outputs uses a 0
+    # blockheight sentinel where outputs uses NULL).
+    legacy = l1.db_query('SELECT lower(hex(prev_out_tx)) AS txid,'
+                         ' prev_out_index AS outnum, value AS satoshis,'
+                         ' COALESCE(confirmation_height, 0) AS blockheight,'
+                         ' spend_height AS spendheight, keyindex'
+                         ' FROM outputs ORDER BY txid')
+    new = l1.db_query('SELECT lower(hex(txid)) AS txid, outnum, satoshis,'
+                      ' blockheight, spendheight, keyindex'
+                      ' FROM our_outputs ORDER BY txid')
+    assert len(new) == 2
+    assert new == legacy
+
+    assert (l1.db_query('SELECT lower(hex(txid)) AS txid FROM our_txs ORDER BY txid')
+            == l1.db_query('SELECT lower(hex(id)) AS txid FROM transactions ORDER BY id'))
+
+    # Both wallet UTXOs survived the upgrade (the reorg unconfirmed one and
+    # unspent the other; listfunds txids are in display byte order).
+    assert {(o['txid'], o['output']) for o in l1.rpc.listfunds()['outputs']} == {
+        ('63c59b312976320528552c258ae51563498dfd042b95bb0c842696614d59bb89', 1),
+        ('675ab2a8c43afcf98b82a1120d1a4d36768c898792fe1282c5be4ac055377fbe', 1)}
+
+    # The wallet still works after the upgrade: deposit fresh funds and
+    # spend them (the pre-upgrade UTXOs aren't on this chain, so pick the
+    # new one explicitly).
+    bitcoind.rpc.sendtoaddress(l1.rpc.newaddr()['p2tr'], 0.01)
+    bitcoind.generate_block(1, wait_for_mempool=1)
+    sync_blockheight(bitcoind, [l1])
+    wait_for(lambda: len(l1.rpc.listfunds()['outputs']) == 3)
+
+    new_utxo = only_one([o for o in l1.rpc.listfunds()['outputs']
+                         if o['amount_msat'] == 1000000000])
+    withdraw = l1.rpc.withdraw(l1.rpc.newaddr()['p2tr'], 500000,
+                               utxos=['{}:{}'.format(new_utxo['txid'],
+                                                     new_utxo['output'])])
+    bitcoind.generate_block(1, wait_for_mempool=1)
+    sync_blockheight(bitcoind, [l1])
+    wait_for(lambda: [o for o in l1.rpc.listfunds()['outputs']
+                      if o['txid'] == withdraw['txid'] and o['status'] == 'confirmed'] != [])
+
+
 @pytest.mark.slow_test
 @unittest.skipIf(os.getenv('TEST_DB_PROVIDER', 'sqlite3') != 'sqlite3', "This test is based on a sqlite3 snapshot")
 @unittest.skipIf(TEST_NETWORK != 'regtest', "The network must match the DB snapshot")