diff --git a/db/exec.c b/db/exec.c index 382ad9f15e95..68c2ced55a3c 100644 --- a/db/exec.c +++ b/db/exec.c @@ -89,6 +89,39 @@ s64 db_get_intvar(struct db *db, const char *varname, s64 defval) return res; } +void db_set_blobvar(struct db *db, const char *varname, const u8 *val, size_t len) +{ + size_t changes; + struct db_stmt *stmt = db_prepare_v2(db, SQL("UPDATE vars SET blobval=? WHERE name=?;")); + db_bind_blob(stmt, val, len); + db_bind_text(stmt, varname); + db_exec_prepared_v2(stmt); + changes = db_count_changes(stmt); + tal_free(stmt); + + if (changes == 0) { + stmt = db_prepare_v2(db, SQL("INSERT INTO vars (name, blobval) VALUES (?, ?);")); + db_bind_text(stmt, varname); + db_bind_blob(stmt, val, len); + db_exec_prepared_v2(stmt); + tal_free(stmt); + } +} + +const u8 *db_get_blobvar(const tal_t *ctx, struct db *db, const char *varname) +{ + struct db_stmt *stmt = db_prepare_v2( + db, SQL("SELECT blobval FROM vars WHERE name=? LIMIT 1")); + db_bind_text(stmt, varname); + + const u8 *res = NULL; + if (db_query_prepared_canfail(stmt) && db_step(stmt)) + res = db_col_arr(ctx, stmt, "blobval", u8); + + tal_free(stmt); + return res; +} + /* Leak tracking. */ /* By making the update conditional on the current value we expect we diff --git a/db/exec.h b/db/exec.h index c852d9501e9a..242b2f52ff88 100644 --- a/db/exec.h +++ b/db/exec.h @@ -4,6 +4,7 @@ #include #include +#include struct db; @@ -23,6 +24,10 @@ void db_set_intvar(struct db *db, const char *varname, s64 val); */ s64 db_get_intvar(struct db *db, const char *varname, s64 defval); +void db_set_blobvar(struct db *db, const char *varname, const u8 *val, size_t len); +/* Returns a tal-allocated blob, or NULL if not found. */ +const u8 *db_get_blobvar(const tal_t *ctx, struct db *db, const char *varname); + /* Get the current data version (entries). */ u32 db_data_version_get(struct db *db); diff --git a/lightningd/Makefile b/lightningd/Makefile index 4fb5afbc269c..2d6950d39d7d 100644 --- a/lightningd/Makefile +++ b/lightningd/Makefile @@ -4,6 +4,7 @@ LIGHTNINGD_SRC := \ lightningd/anchorspend.c \ lightningd/bitcoind.c \ lightningd/chaintopology.c \ + lightningd/watchman.c \ lightningd/channel.c \ lightningd/channel_control.c \ lightningd/channel_gossip.c \ diff --git a/lightningd/lightningd.c b/lightningd/lightningd.c index 3e95e59c3f18..6810baee57d5 100644 --- a/lightningd/lightningd.c +++ b/lightningd/lightningd.c @@ -75,6 +75,7 @@ #include #include #include +#include #include #include #include @@ -1347,6 +1348,10 @@ int main(int argc, char *argv[]) setup_topology(ld->topology); 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); + db_begin_transaction(ld->wallet->db); trace_span_start("delete_old_htlcs", ld->wallet); wallet_delete_old_htlcs(ld->wallet); diff --git a/lightningd/lightningd.h b/lightningd/lightningd.h index 81a03d41261f..6d778c929e95 100644 --- a/lightningd/lightningd.h +++ b/lightningd/lightningd.h @@ -12,6 +12,7 @@ #include struct amount_msat; +struct watchman; /* Various adjustable things. */ struct config { @@ -244,6 +245,7 @@ struct lightningd { /* Derive all our BIP86 keys from here */ struct ext_key *bip86_base; struct wallet *wallet; + struct watchman *watchman; /* Outstanding waitsendpay commands. */ struct list_head waitsendpay_commands; diff --git a/lightningd/plugin.c b/lightningd/plugin.c index 444c626e136e..9edb27b0ea0c 100644 --- a/lightningd/plugin.c +++ b/lightningd/plugin.c @@ -29,6 +29,7 @@ #include #include #include +#include /* Only this file can include this generated header! */ # include @@ -2087,6 +2088,7 @@ static void plugin_config_cb(const char *buffer, } if (tal_count(plugin->custom_msgs)) tell_connectd_custommsgs(plugin->plugins); + watchman_notify_plugin_ready(plugin->plugins->ld, plugin); notify_plugin_started(plugin->plugins->ld, plugin); check_plugins_initted(plugin->plugins); } diff --git a/lightningd/test/run-find_my_abspath.c b/lightningd/test/run-find_my_abspath.c index 3d227bd9974d..fc3782295f15 100644 --- a/lightningd/test/run-find_my_abspath.c +++ b/lightningd/test/run-find_my_abspath.c @@ -222,6 +222,9 @@ struct wallet *wallet_new(struct lightningd *ld UNNEEDED, struct timers *timers /* Generated stub for wallet_sanity_check */ bool wallet_sanity_check(struct wallet *w UNNEEDED) { fprintf(stderr, "wallet_sanity_check called!\n"); abort(); } +/* Generated stub for watchman_new */ +struct watchman *watchman_new(const tal_t *ctx UNNEEDED, struct lightningd *ld UNNEEDED) +{ fprintf(stderr, "watchman_new called!\n"); abort(); } /* AUTOGENERATED MOCKS END */ struct logger *crashlog; diff --git a/lightningd/test/run-invoice-select-inchan.c b/lightningd/test/run-invoice-select-inchan.c index 0357cf94cd4c..d41bf5466e62 100644 --- a/lightningd/test/run-invoice-select-inchan.c +++ b/lightningd/test/run-invoice-select-inchan.c @@ -158,12 +158,6 @@ bool command_deprecated_in_ok(struct command *cmd UNNEEDED, const char *depr_start UNNEEDED, const char *depr_end UNNEEDED) { fprintf(stderr, "command_deprecated_in_ok called!\n"); abort(); } -/* Generated stub for command_deprecated_out_ok */ -bool command_deprecated_out_ok(struct command *cmd UNNEEDED, - const char *fieldname UNNEEDED, - const char *depr_start UNNEEDED, - const char *depr_end UNNEEDED) -{ fprintf(stderr, "command_deprecated_out_ok called!\n"); abort(); } /* Generated stub for command_dev_apis */ bool command_dev_apis(const struct command *cmd UNNEEDED) { fprintf(stderr, "command_dev_apis called!\n"); abort(); } diff --git a/lightningd/watchman.c b/lightningd/watchman.c new file mode 100644 index 000000000000..e5ebaf46a0ad --- /dev/null +++ b/lightningd/watchman.c @@ -0,0 +1,837 @@ +#include "config.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* + * Watchman is the interface between lightningd and the bwatch plugin. + * It manages a pending operation queue to ensure reliable delivery of + * watch add/delete requests to bwatch, even across crashes. + * + * Architecture: + * - Subsystems (channel, onchaind, wallet) call watchman_add/watchman_del + * - Watchman queues operations and sends them to bwatch via RPC + * - Operations stay in queue until bwatch acknowledges them + * - On crash/restart, pending ops are replayed from datastore + * - Bwatch handles duplicate operations idempotently + */ + +/* A pending operation - method and params to send to bwatch */ +struct pending_op { + /* "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/p2wpkh/42". + * 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 */ +}; + + +/* + * Datastore persistence helpers + * Pending operations are stored at ["watchman", "pending", op_id] + */ + +/* Generate datastore key for a pending operation */ +static const char **make_key(const tal_t *ctx, const char *op_id TAKES) +{ + return mkdatastorekey(ctx, "watchman", "pending", op_id); +} + + +/* Persist a pending operation to the datastore for crash recovery. + * The method is encoded in op_id (see struct pending_op), so we store + * only json_params as the value. */ +static void db_save(struct watchman *wm, const struct pending_op *op) +{ + const char **key = make_key(tmpctx, op->op_id); + const u8 *data = (const u8 *)op->json_params; + if (wallet_datastore_get(tmpctx, wm->ld->wallet, key, NULL)) + wallet_datastore_update(wm->ld->wallet, key, data); + else + wallet_datastore_create(wm->ld->wallet, key, data); +} + +/* Remove a pending operation from the datastore */ +static void db_remove(struct watchman *wm, const char *op_id) +{ + const char **key = make_key(tmpctx, op_id); + wallet_datastore_remove(wm->ld->wallet, key); +} + +static void save_tip(struct watchman *wm) +{ + struct db *db = wm->ld->wallet->db; + db_set_intvar(db, "last_watchman_block_height", wm->last_processed_height); + db_set_blobvar(db, "last_watchman_block_hash", + (const u8 *)&wm->last_processed_hash, + sizeof(wm->last_processed_hash)); +} + +static void load_tip(struct watchman *wm) +{ + struct db *db = wm->ld->wallet->db; + const u8 *blob; + + wm->last_processed_height = db_get_intvar(db, "last_watchman_block_height", 0); + + blob = db_get_blobvar(tmpctx, db, "last_watchman_block_hash"); + if (blob) { + assert(tal_bytelen(blob) == sizeof(struct bitcoin_blkid)); + memcpy(&wm->last_processed_hash, blob, sizeof(wm->last_processed_hash)); + } +} + +/* Load all pending operations from datastore on startup */ +static void load_pending_ops(struct watchman *wm) +{ + const char **startkey = mkdatastorekey(tmpctx, "watchman", "pending"); + const char **key; + const u8 *data; + u64 generation; + struct db_stmt *stmt; + + for (stmt = wallet_datastore_first(tmpctx, wm->ld->wallet, startkey, + &key, &data, &generation); + stmt; + stmt = wallet_datastore_next(tmpctx, startkey, stmt, + &key, &data, &generation)) { + if (tal_count(key) != 3) + continue; + + /* op_id is the datastore key; method is the prefix before ':'. + * Malformed keys (no ':') are skipped — they can't be replayed. */ + if (!strchr(key[2], ':')) { + log_broken(wm->ld->log, + "Skipping malformed pending op key '%s' (no ':' separator)", + key[2]); + continue; + } + + struct pending_op *op = tal(wm, struct pending_op); + op->op_id = tal_strdup(op, key[2]); + op->json_params = tal_strdup(op, (const char *)data); + tal_arr_expand(&wm->pending_ops, op); + + log_debug(wm->ld->log, "Loaded pending op: %s", op->op_id); + } +} + +/* Apply --rescan: negative means absolute height (only go back), + * positive means relative (go back N blocks from stored tip). */ +static void apply_rescan(struct watchman *wm, struct lightningd *ld) +{ + u32 stored = wm->last_processed_height; + u32 target; + + if (ld->config.rescan < 0) + target = (u32)(-ld->config.rescan); /* absolute height */ + else if (stored > (u32)ld->config.rescan) + target = stored - (u32)ld->config.rescan; /* go back N blocks */ + else + target = 0; /* rescan exceeds stored height, start from genesis */ + + /* Only adjust downward; upward targets are validated later in chaininfo */ + if (target < stored) { + log_debug(ld->log, + "Rescanning: adjusting watchman height from %u to %u", + stored, target); + wm->last_processed_height = target; + } +} + +struct watchman *watchman_new(const tal_t *ctx, struct lightningd *ld) +{ + struct watchman *wm = talz(ctx, struct watchman); + + wm->ld = ld; + wm->pending_ops = tal_arr(wm, struct pending_op *, 0); + + db_begin_transaction(ld->wallet->db); + load_pending_ops(wm); + load_tip(wm); + db_commit_transaction(ld->wallet->db); + apply_rescan(wm, ld); + + log_info(ld->log, "Watchman: height=%u, %zu pending ops", + wm->last_processed_height, tal_count(wm->pending_ops)); + + return wm; +} + +/* Per-request context for bwatch_ack_response. Carries the bare op_id so the + * 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" */ +}; + +/* Response callback for bwatch RPC requests; handles both success and error. */ +static void bwatch_ack_response(const char *buffer, + const jsmntok_t *toks, + const jsmntok_t *idtok UNUSED, + struct bwatch_ack_arg *arg) +{ + const jsmntok_t *err = json_get_member(buffer, toks, "error"); + + if (err) { + log_unusual(arg->wm->ld->log, "bwatch operation %s failed: %.*s", + arg->op_id, json_tok_full_len(err), json_tok_full(buffer, err)); + } else { + log_debug(arg->wm->ld->log, "Acknowledged pending op: %s", arg->op_id); + } + + watchman_ack(arg->wm->ld, arg->op_id); +} + +/* op_id is "{method}:{owner}"; return the owner suffix. */ +static const char *owner_from_op_id(const char *op_id) +{ + const char *colon = strchr(op_id, ':'); + return colon ? colon + 1 : ""; +} + +/* op_id is "{method}:{owner}"; return the method prefix. */ +static const char *method_from_op_id(const tal_t *ctx, const char *op_id) +{ + const char *colon = strchr(op_id, ':'); + assert(colon); /* op_id must always be "{method}:{owner}" */ + return tal_strndup(ctx, op_id, colon - op_id); +} + +/* Send an RPC request to the bwatch plugin. + * op_id must be "{method}:{owner}", e.g. "addscriptpubkeywatch:wallet/p2wpkh/42". */ +static void send_to_bwatch(struct watchman *wm, const char *method, + const char *op_id, const char *json_params) +{ + struct plugin *bwatch; + struct jsonrpc_request *req; + const char *owner; + size_t len; + + /* Find bwatch plugin by the command it registers */ + bwatch = find_plugin_for_command(wm->ld, method); + if (!bwatch) { + log_broken(wm->ld->log, "bwatch plugin not found, cannot send %s", method); + return; + } + + if (bwatch->plugin_state != INIT_COMPLETE) { + log_debug(wm->ld->log, "bwatch plugin not ready (state %d), queuing %s %s", + bwatch->plugin_state, method, op_id); + return; + } + + struct bwatch_ack_arg *arg = tal(tmpctx, struct bwatch_ack_arg); + arg->wm = wm; + arg->op_id = tal_strdup(arg, op_id); + + req = jsonrpc_request_start(wm, method, op_id, bwatch->log, + NULL, bwatch_ack_response, arg); + + /* Parent arg to req so it's freed when the request is freed, + * regardless of whether the callback fires. */ + tal_steal(req, arg); + + owner = owner_from_op_id(op_id); + if (!streq(owner, "")) + json_add_string(req->stream, "owner", owner); + + /* json_params is a JSON object string like {"type":"...","scriptpubkey":"...","start_block":N}. + * Append the rest (skip outer braces) so we get type, scriptpubkey, start_block, etc. */ + len = strlen(json_params); + if (len >= 2 && json_params[0] == '{' && json_params[len-1] == '}') { + json_stream_append(req->stream, ",", 1); + json_stream_append(req->stream, json_params + 1, len - 2); + } else { + json_stream_append(req->stream, ",", 1); + json_stream_append(req->stream, json_params, len); + } + + jsonrpc_request_end(req); + plugin_request_send(bwatch, req); +} + +/* Queue an operation, persist it for crash recovery, and send to bwatch. */ +static void enqueue_op(struct watchman *wm, const char *method, + const char *op_id, const char *json_params) +{ + struct pending_op *op = tal(wm, struct pending_op); + op->op_id = tal_strdup(op, op_id); + op->json_params = tal_strdup(op, json_params); + tal_arr_expand(&wm->pending_ops, op); + db_save(wm, op); + send_to_bwatch(wm, method, op_id, json_params); +} + +/* Internal: queue an add for a specific per-type bwatch command. */ +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); + + /* Remove any existing add for this owner */ + watchman_ack(ld, op_id); + enqueue_op(wm, method, op_id, json_params); +} + +/** + * watchman_del - Queue a delete watch operation + * + * Simply queues the operation and sends to bwatch. + * Bwatch handles duplicate deletes idempotently. + * Cancels any pending add for this owner. + */ +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); + + /* Cancel any pending add for this owner. All del-methods are named + * "del" and their paired add-method is "add" */ + assert(strstarts(method, "del")); + char *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); +} + +/** + * watchman_ack - Acknowledge a completed watch operation + * + * 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. + */ +void watchman_ack(struct lightningd *ld, const char *op_id) +{ + struct watchman *wm = ld->watchman; + + 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); + tal_free(wm->pending_ops[i]); + tal_arr_remove(&wm->pending_ops, i); + return; + } + } +} + +/** + * watchman_replay_pending - Resend all pending operations to bwatch + * + * Called on startup after bwatch is ready, to ensure any operations + * that were pending before a crash are sent to bwatch. + */ +void watchman_replay_pending(struct lightningd *ld) +{ + struct watchman *wm = ld->watchman; + + 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), + op->op_id, op->json_params); + } +} + +/* Replay pending ops when bwatch is ready. On a fresh node current_height + * is still 0, so we defer to json_block_processed where it's guaranteed > 0. */ +void watchman_notify_plugin_ready(struct lightningd *ld, struct plugin *plugin) +{ + struct watchman *wm = ld->watchman; + + if (!wm) + return; + /* Check if this is bwatch by seeing if it owns the "addscriptpubkeywatch" method. */ + if (find_plugin_for_command(ld, "addscriptpubkeywatch") != plugin) + return; + + if (wm->last_processed_height > 0) { + log_debug(ld->log, "bwatch reached INIT_COMPLETE, replaying pending ops (height=%u)", + wm->last_processed_height); + watchman_replay_pending(ld); + } +} + +void watchman_watch_scriptpubkey(struct lightningd *ld, + const char *owner, + const u8 *scriptpubkey, + size_t script_len, + u32 start_block) +{ + watchman_add(ld, "addscriptpubkeywatch", owner, + tal_fmt(tmpctx, "{\"scriptpubkey\":\"%s\",\"start_block\":%u}", + tal_hexstr(tmpctx, scriptpubkey, script_len), + start_block)); +} + +void watchman_unwatch_scriptpubkey(struct lightningd *ld, + const char *owner, + const u8 *scriptpubkey, + size_t script_len) +{ + watchman_del(ld, "delscriptpubkeywatch", owner, + tal_fmt(tmpctx, "{\"scriptpubkey\":\"%s\"}", + tal_hexstr(tmpctx, scriptpubkey, script_len))); +} + +void watchman_watch_outpoint(struct lightningd *ld, + const char *owner, + const struct bitcoin_outpoint *outpoint, + u32 start_block) +{ + watchman_add(ld, "addoutpointwatch", owner, + tal_fmt(tmpctx, "{\"outpoint\":\"%s:%u\",\"start_block\":%u}", + fmt_bitcoin_txid(tmpctx, &outpoint->txid), + outpoint->n, start_block)); +} + +void watchman_unwatch_outpoint(struct lightningd *ld, + const char *owner, + const struct bitcoin_outpoint *outpoint) +{ + watchman_del(ld, "deloutpointwatch", owner, + tal_fmt(tmpctx, "{\"outpoint\":\"%s:%u\"}", + fmt_bitcoin_txid(tmpctx, &outpoint->txid), + outpoint->n)); +} + +void watchman_watch_scid(struct lightningd *ld, + const char *owner, + const struct short_channel_id *scid, + u32 start_block) +{ + watchman_add(ld, "addscidwatch", owner, + tal_fmt(tmpctx, "{\"scid\":\"%s\",\"start_block\":%u}", + fmt_short_channel_id(tmpctx, *scid), start_block)); +} + +void watchman_unwatch_scid(struct lightningd *ld, + const char *owner, + const struct short_channel_id *scid) +{ + watchman_del(ld, "delscidwatch", owner, + tal_fmt(tmpctx, "{\"scid\":\"%s\"}", + fmt_short_channel_id(tmpctx, *scid))); +} + +void watchman_watch_blockdepth(struct lightningd *ld, + const char *owner, + u32 confirm_height) +{ + watchman_add(ld, "addblockdepthwatch", owner, + tal_fmt(tmpctx, "{\"start_block\":%u}", confirm_height)); +} + +void watchman_unwatch_blockdepth(struct lightningd *ld, + const char *owner, + u32 confirm_height) +{ + watchman_del(ld, "delblockdepthwatch", owner, + tal_fmt(tmpctx, "{\"start_block\":%u}", confirm_height)); +} + +/* Dispatch table - add new watch types here */ +static const struct depth_dispatch { + const char *prefix; + depth_found_fn handler; + watch_revert_fn revert; +} depth_handlers[] = { + /* Entries added in subsequent commits alongside their handler functions. */ + { NULL, NULL, NULL }, +}; + +static const struct watch_dispatch { + const char *prefix; + watch_found_fn handler; + watch_revert_fn revert; +} watch_handlers[] = { + /* Entries added in subsequent commits alongside their handler functions. */ + { NULL, NULL, NULL }, +}; + +/* dispatch_watch_found: search depth_handlers then watch_handlers for owner. + * depth is NULL for tx-based notifications, set for blockdepth notifications. */ +static void dispatch_watch_found(struct lightningd *ld, + const char *owner, + const struct bitcoin_tx *tx, + size_t outnum, + u32 blockheight, + u32 txindex, + const u32 *depth) +{ + for (size_t i = 0; i < ARRAY_SIZE(depth_handlers); i++) { + if (!depth_handlers[i].prefix) + continue; + if (strstarts(owner, depth_handlers[i].prefix)) { + const char *suffix = owner + strlen(depth_handlers[i].prefix); + depth_handlers[i].handler(ld, suffix, *depth, blockheight); + return; + } + } + for (size_t i = 0; i < ARRAY_SIZE(watch_handlers); i++) { + if (!watch_handlers[i].prefix) + continue; + if (strstarts(owner, watch_handlers[i].prefix)) { + const char *suffix = owner + strlen(watch_handlers[i].prefix); + watch_handlers[i].handler(ld, suffix, tx, outnum, blockheight, txindex); + return; + } + } + log_debug(ld->log, "No handler for watch owner: %s", owner); +} + +static void dispatch_watch_revert(struct lightningd *ld, + const char *owner, + u32 blockheight) +{ + for (size_t i = 0; i < ARRAY_SIZE(depth_handlers); i++) { + if (!depth_handlers[i].prefix) + continue; + if (strstarts(owner, depth_handlers[i].prefix)) { + const char *suffix = owner + strlen(depth_handlers[i].prefix); + depth_handlers[i].revert(ld, suffix, blockheight); + return; + } + } + for (size_t i = 0; i < ARRAY_SIZE(watch_handlers); i++) { + if (!watch_handlers[i].prefix) + continue; + if (strstarts(owner, watch_handlers[i].prefix)) { + const char *suffix = owner + strlen(watch_handlers[i].prefix); + watch_handlers[i].revert(ld, suffix, blockheight); + return; + } + } + log_debug(ld->log, "No revert handler for watch owner: %s", owner); +} + +static struct command_result *param_bitcoin_tx(struct command *cmd, + const char *name, + const char *buffer, + const jsmntok_t *tok, + struct bitcoin_tx **tx) +{ + *tx = bitcoin_tx_from_hex(cmd, buffer + tok->start, tok->end - tok->start); + if (!*tx) + return command_fail_badparam(cmd, name, buffer, tok, + "Expected a hex-encoded transaction"); + return NULL; +} + +static struct command_result *param_bitcoin_blkid_cmd(struct command *cmd, + const char *name, + const char *buffer, + const jsmntok_t *tok, + struct bitcoin_blkid **blkid) +{ + *blkid = tal(cmd, struct bitcoin_blkid); + if (!json_to_bitcoin_blkid(buffer, tok, *blkid)) + return command_fail_badparam(cmd, name, buffer, tok, + "Expected a blockhash"); + return NULL; +} + +/** + * json_watch_found - RPC handler for watch_found notifications from bwatch + * + * Handles both tx-based watches (scriptpubkey, outpoint, txid, scid) and + * blockdepth watches. Dispatches by owner prefix. + * + * For WATCH_SCID, bwatch may omit "tx" and "txindex" to signal that the + * SCID's expected tx/output was absent from the encoded block ("not found"). + * The handler (gossip_scid_watch_found) detects this via tx==NULL. + */ +static struct command_result *json_watch_found(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + const char **owners; + u32 *blockheight, *txindex, *index, *depth; + struct bitcoin_tx *tx; + + if (!param_check(cmd, buffer, params, + p_req("blockheight", param_number, &blockheight), + p_req("owners", param_string_array, &owners), + p_opt("tx", param_bitcoin_tx, &tx), + p_opt("txindex", param_number, &txindex), + p_opt("index", param_number, &index), + p_opt("depth", param_number, &depth), + NULL)) + return command_param_failed(); + + /* For normal tx-based watches tx+txindex are required. + * Exception: WATCH_SCID owners send watch_found with tx==NULL to + * signal "not found"; their handler checks for this explicitly. */ + if (!depth && !tx && txindex) + return command_fail(cmd, JSONRPC2_INVALID_PARAMS, + "txindex provided without tx in watch_found"); + if (!depth && tx && !txindex) + return command_fail(cmd, JSONRPC2_INVALID_PARAMS, + "tx provided without txindex in watch_found"); + + if (!cmd->ld->watchman) + return command_fail(cmd, LIGHTNINGD, "Watchman not initialized"); + + if (command_check_only(cmd)) + return command_check_done(cmd); + + log_debug(cmd->ld->log, "watch_found at block %u%s", *blockheight, + depth ? " (blockdepth)" : ""); + for (size_t i = 0; i < tal_count(owners); i++) + dispatch_watch_found(cmd->ld, owners[i], tx, + index ? *index : 0, + *blockheight, + txindex ? *txindex : 0, + depth); + + struct json_stream *response = json_stream_success(cmd); + json_add_u32(response, "blockheight", *blockheight); + return command_success(cmd, response); +} + +static const struct json_command watch_found_command = { + "watch_found", + json_watch_found, +}; +AUTODATA(json_command, &watch_found_command); + +/** + * json_watch_revert - RPC handler for watch_revert notifications from bwatch + * + * Called when a watched item's confirming block is reorged away. Dispatches + * to the appropriate revert handler (depth or tx) based on owner prefix. + */ +static struct command_result *json_watch_revert(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + const char *owner; + u32 *blockheight; + + if (!param(cmd, buffer, params, + p_req("owner", param_string, &owner), + p_req("blockheight", param_number, &blockheight), + NULL)) + return command_param_failed(); + + dispatch_watch_revert(cmd->ld, owner, *blockheight); + struct json_stream *response = json_stream_success(cmd); + json_add_u32(response, "blockheight", *blockheight); + return command_success(cmd, response); +} + +static const struct json_command watch_revert_command = { + "watch_revert", + json_watch_revert, +}; +AUTODATA(json_command, &watch_revert_command); + +static struct command_result *json_revert_block_processed(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + struct watchman *wm = cmd->ld->watchman; + u32 *blockheight; + struct bitcoin_blkid *blockhash; + + if (!param(cmd, buffer, params, + p_req("blockheight", param_number, &blockheight), + p_req("blockhash", param_bitcoin_blkid_cmd, &blockhash), + NULL)) + return command_param_failed(); + + if (!wm) + return command_fail(cmd, LIGHTNINGD, "Watchman not initialized"); + + log_debug(wm->ld->log, "block_reverted: %u -> %u", + wm->last_processed_height, *blockheight); + wm->last_processed_height = *blockheight; + wm->last_processed_hash = *blockhash; + save_tip(wm); + + struct json_stream *response = json_stream_success(cmd); + json_add_u32(response, "blockheight", *blockheight); + return command_success(cmd, response); +} + +static const struct json_command revert_block_processed_command = { + "revert_block_processed", + json_revert_block_processed, +}; +AUTODATA(json_command, &revert_block_processed_command); + +/** + * json_block_processed - RPC handler for block_processed notifications from bwatch + * + * Called by bwatch after it finishes processing all watches in a block. + * We track this height to know where bwatch is in the chain, which helps + * during startup/reorg scenarios. + */ +static struct command_result *json_block_processed(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + struct watchman *wm = cmd->ld->watchman; + u32 *blockheight; + struct bitcoin_blkid *blockhash; + + if (!param(cmd, buffer, params, + p_req("blockheight", param_number, &blockheight), + p_req("blockhash", param_bitcoin_blkid_cmd, &blockhash), + NULL)) + return command_param_failed(); + + if (!wm) + return command_fail(cmd, LIGHTNINGD, "Watchman not initialized"); + + if (*blockheight != wm->last_processed_height) { + log_info(wm->ld->log, "block_processed: %u -> %u", + wm->last_processed_height, *blockheight); + + /* Fresh node: replay wallet watches now that bwatch->current_height > 0, + * so add_watch_and_maybe_rescan will trigger historical rescans. */ + if (wm->last_processed_height == 0) { + log_debug(wm->ld->log, + "First block_processed on fresh node, replaying pending ops"); + watchman_replay_pending(wm->ld); + } + + wm->last_processed_height = *blockheight; + wm->last_processed_hash = *blockhash; + save_tip(wm); + send_account_balance_snapshot(wm->ld); + } + + notify_new_block(wm->ld); + + struct json_stream *response = json_stream_success(cmd); + json_add_u32(response, "blockheight", *blockheight); + if (wm->last_processed_height > 0) + json_add_string(response, "blockhash", + fmt_bitcoin_blkid(response, &wm->last_processed_hash)); + return command_success(cmd, response); +} + +static const struct json_command block_processed_command = { + "block_processed", + json_block_processed, +}; +AUTODATA(json_command, &block_processed_command); + +/** + * json_getwatchmanheight - RPC handler to return watchman's last processed height + * + * Called by bwatch on startup to determine what height to rescan from. + */ +static struct command_result *json_getwatchmanheight(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + struct watchman *wm = cmd->ld->watchman; + struct json_stream *response; + u32 height; + + if (!param(cmd, buffer, params, NULL)) + return command_param_failed(); + + height = wm ? wm->last_processed_height : 0; + log_debug(cmd->ld->log, "getwatchmanheight: returning height=%u (wm=%s)", + height, wm ? "ok" : "NULL"); + response = json_stream_success(cmd); + json_add_u32(response, "height", height); + if (wm && wm->last_processed_height > 0) + json_add_string(response, "blockhash", + fmt_bitcoin_blkid(response, &wm->last_processed_hash)); + return command_success(cmd, response); +} + +static const struct json_command getwatchmanheight_command = { + "getwatchmanheight", + json_getwatchmanheight, +}; +AUTODATA(json_command, &getwatchmanheight_command); + +/** + * json_chaininfo - RPC handler for chaininfo from bwatch + * + * Called by bwatch on startup to inform watchman about the chain name, + * IBD status, and sync state. Validates we're on the right network and + * sets bitcoind->synced accordingly. + */ +static struct command_result *json_chaininfo(struct command *cmd, + const char *buffer, + const jsmntok_t *obj UNUSED, + const jsmntok_t *params) +{ + const char *chain; + u32 *headercount, *blockcount; + bool *ibd; + + if (!param(cmd, buffer, params, + p_req("chain", param_string, &chain), + p_req("headercount", param_number, &headercount), + p_req("blockcount", param_number, &blockcount), + p_req("ibd", param_bool, &ibd), + NULL)) + return command_param_failed(); + + if (!streq(chain, chainparams->bip70_name)) + fatal("Wrong network! Our Bitcoin backend is running on '%s'," + " but we expect '%s'.", chain, chainparams->bip70_name); + if (*ibd) { + log_unusual(cmd->ld->log, + "Waiting for initial block download" + " (this can take a while!)"); + cmd->ld->topology->bitcoind->synced = false; + } else if (*headercount != *blockcount) { + log_unusual(cmd->ld->log, + "Waiting for bitcoind to catch up" + " (%u blocks of %u)", + *blockcount, *headercount); + cmd->ld->topology->bitcoind->synced = false; + } else { + if (!cmd->ld->topology->bitcoind->synced) + log_info(cmd->ld->log, "Bitcoin backend now synced"); + cmd->ld->topology->bitcoind->synced = true; + notify_new_block(cmd->ld); + } + + cmd->ld->watchman->bitcoind_blockcount = *blockcount; + + struct json_stream *response = json_stream_success(cmd); + json_add_string(response, "chain", chain); + json_add_bool(response, "synced", cmd->ld->topology->bitcoind->synced); + return command_success(cmd, response); +} + +static const struct json_command chaininfo_command = { + "chaininfo", + json_chaininfo, +}; +AUTODATA(json_command, &chaininfo_command); diff --git a/lightningd/watchman.h b/lightningd/watchman.h new file mode 100644 index 000000000000..8271c23a4a29 --- /dev/null +++ b/lightningd/watchman.h @@ -0,0 +1,147 @@ +#ifndef LIGHTNING_LIGHTNINGD_WATCHMAN_H +#define LIGHTNING_LIGHTNINGD_WATCHMAN_H + +#include "config.h" +#include +#include + +struct lightningd; +struct pending_op; +struct plugin; +struct short_channel_id; + +/* lightningd's view of bwatch. bwatch lives in a separate process and tells + * us about new/reverted blocks and watch hits via JSON-RPC; watchman tracks + * what we've already processed and queues outbound watch ops while bwatch is + * starting up. */ +struct watchman { + struct lightningd *ld; + u32 last_processed_height; + struct bitcoin_blkid last_processed_hash; + u32 bitcoind_blockcount; + struct pending_op **pending_ops; +}; + +/** + * 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 + * @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 + * @txindex: position of tx in block (0 = coinbase) + * + * Called when bwatch detects a watched item in a block. + */ +typedef void (*watch_found_fn)(struct lightningd *ld, + const char *suffix, + const struct bitcoin_tx *tx, + size_t outnum, + u32 blockheight, + u32 txindex); + +typedef void (*watch_revert_fn)(struct lightningd *ld, + const char *suffix, + u32 blockheight); + +/** + * depth_found_fn - Handler for blockdepth watch notifications. + * @depth: new_height - confirm_height + 1 (always >= 1) + * @blockheight: current chain tip height + * + * Called once per new block. When the confirming block is reorged away, + * watch_revert_fn is called instead. + */ +typedef void (*depth_found_fn)(struct lightningd *ld, + const char *suffix, + u32 depth, + u32 blockheight); + +/** + * watchman_notify_plugin_ready - Called by plugin.c when any plugin reaches INIT_COMPLETE + * + * Checks whether the newly-ready plugin is bwatch and, if so, replays any + * pending watch operations that were queued before bwatch was available. + */ +void watchman_notify_plugin_ready(struct lightningd *ld, struct plugin *plugin); + +/** + * watchman_new - Create and initialize a new watchman instance + * @ctx: tal context to allocate from + * @ld: lightningd instance + * + * Returns a new watchman instance, loading pending operations from datastore. + */ +struct watchman *watchman_new(const tal_t *ctx, struct lightningd *ld); + +/** + * watchman_ack - Acknowledge a completed watch operation + * @ld: lightningd instance + * @op_id: the operation ID that was acknowledged + * + * Called when bwatch acknowledges a watch operation. + */ +void watchman_ack(struct lightningd *ld, const char *op_id); + +/** + * watchman_replay_pending - Replay all pending operations + * @ld: lightningd instance + * + * Resends all pending watch operations to bwatch. + * Call this when bwatch is ready (e.g., on startup). + */ +void watchman_replay_pending(struct lightningd *ld); + +/** Register a WATCH_SCRIPTPUBKEY — fires when @scriptpubkey appears in a tx output. */ +void watchman_watch_scriptpubkey(struct lightningd *ld, + const char *owner, + const u8 *scriptpubkey, + size_t script_len, + u32 start_block); + +/** Remove a WATCH_SCRIPTPUBKEY. */ +void watchman_unwatch_scriptpubkey(struct lightningd *ld, + const char *owner, + const u8 *scriptpubkey, + size_t script_len); + +/** Register a WATCH_OUTPOINT — fires when @outpoint is spent. */ +void watchman_watch_outpoint(struct lightningd *ld, + const char *owner, + const struct bitcoin_outpoint *outpoint, + u32 start_block); + +/** Remove a WATCH_OUTPOINT (e.g. during splice before re-adding for new outpoint). */ +void watchman_unwatch_outpoint(struct lightningd *ld, + const char *owner, + const struct bitcoin_outpoint *outpoint); + +/** Register a WATCH_SCID — fires when bwatch finds the output (for gossip get_txout). */ +void watchman_watch_scid(struct lightningd *ld, + const char *owner, + const struct short_channel_id *scid, + u32 start_block); + +/** Remove a WATCH_SCID. */ +void watchman_unwatch_scid(struct lightningd *ld, + const char *owner, + const struct short_channel_id *scid); + +/** + * watchman_watch_blockdepth - Register a WATCH_BLOCKDEPTH + * @ld: lightningd instance + * @owner: the owner identifier (e.g. "channel/funding_depth/42") + * @confirm_height: the block height where the tx of interest was confirmed + */ +void watchman_watch_blockdepth(struct lightningd *ld, + const char *owner, + u32 confirm_height); + +/** Remove a WATCH_BLOCKDEPTH. */ +void watchman_unwatch_blockdepth(struct lightningd *ld, + const char *owner, + u32 confirm_height); + +#endif /* LIGHTNING_LIGHTNINGD_WATCHMAN_H */