bkpr: add in-mem & datastore storage for external blockheights.

We won't be able to "UPDATE chain_events", so keep a separate record
of these blockheights, and lookup that when the blockheight is 0.

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

plugins/bkpr/Makefile

@@ -3,6 +3,7 @@
 BOOKKEEPER_PLUGIN_SRC := 			\
 	plugins/bkpr/account.c			\
 	plugins/bkpr/account_entry.c		\
+	plugins/bkpr/blockheights.c		\
 	plugins/bkpr/bookkeeper.c		\
 	plugins/bkpr/chain_event.c		\
 	plugins/bkpr/channel_event.c		\
@@ -22,6 +23,7 @@ BOOKKEEPER_SRC := $(BOOKKEEPER_PLUGIN_SRC) $(BOOKKEEPER_DB_QUERIES)
 BOOKKEEPER_HEADER := 				\
 	plugins/bkpr/account.h			\
 	plugins/bkpr/account_entry.h		\
+	plugins/bkpr/blockheights.h		\
 	plugins/bkpr/bookkeeper.h		\
 	plugins/bkpr/chain_event.h		\
 	plugins/bkpr/channel_event.h		\

plugins/bkpr/blockheights.c

@@ -0,0 +1,162 @@
+#include "config.h"
+
+#include <ccan/htable/htable_type.h>
+#include <ccan/json_out/json_out.h>
+#include <ccan/str/hex/hex.h>
+#include <ccan/str/str.h>
+#include <ccan/tal/str/str.h>
+#include <common/memleak.h>
+#include <common/utils.h>
+#include <inttypes.h>
+#include <plugins/bkpr/blockheights.h>
+#include <plugins/bkpr/bookkeeper.h>
+#include <plugins/libplugin.h>
+
+struct blockheight_entry {
+	struct bitcoin_txid txid;
+	u32 height;
+};
+
+static size_t hash_txid(const struct bitcoin_txid *txid)
+{
+	return siphash24(siphash_seed(), txid->shad.sha.u.u8,
+			 sizeof(txid->shad.sha.u.u8));
+}
+
+static const struct bitcoin_txid *
+blockheight_key(const struct blockheight_entry *e)
+{
+	return &e->txid;
+}
+
+static bool blockheight_key_eq(const struct blockheight_entry *e,
+                               const struct bitcoin_txid *k)
+{
+	return bitcoin_txid_eq(&e->txid, k);
+}
+
+HTABLE_DEFINE_NODUPS_TYPE(struct blockheight_entry,
+			  blockheight_key,
+			  hash_txid,
+			  blockheight_key_eq,
+			  blockheight_htable);
+
+struct blockheights {
+	struct blockheight_htable *map;
+};
+
+static void memleak_scan_blockheight_htable(struct htable *memtable,
+					    struct blockheight_htable *ht)
+{
+	memleak_scan_htable(memtable, &ht->raw);
+}
+
+static const char *ds_blockheight_path(const tal_t *ctx,
+				       const struct bitcoin_txid *txid)
+{
+	/* Keys like: bookkeeper/blockheights/<txid> */
+	return tal_fmt(ctx, "bookkeeper/blockheights/%s",
+		       fmt_bitcoin_txid(tmpctx, txid));
+}
+
+void add_blockheight(struct command *cmd,
+		     struct bkpr *bkpr,
+		     const struct bitcoin_txid *txid,
+		     u32 blockheight)
+{
+	struct blockheights *bh = bkpr->blockheights;
+	struct blockheight_entry *e;
+	be32 be_blockheight;
+	const char *path = ds_blockheight_path(tmpctx, txid);
+
+	/* Update in-memory map (replace or insert) */
+	e = blockheight_htable_get(bh->map, txid);
+	if (e) {
+		e->height = blockheight;
+	} else {
+		e = tal(bh->map, struct blockheight_entry);
+		e->txid = *txid;
+		e->height = blockheight;
+		blockheight_htable_add(bh->map, e);
+	}
+
+	be_blockheight = cpu_to_be32(blockheight);
+	jsonrpc_set_datastore_binary(cmd, path,
+				     &be_blockheight, sizeof(be_blockheight),
+				     "create-or-replace",
+				     ignore_datastore_reply, NULL, NULL);
+}
+
+u32 find_blockheight(const struct bkpr *bkpr,
+		     const struct bitcoin_txid *txid)
+{
+	const struct blockheight_entry *e;
+
+	e = blockheight_htable_get(bkpr->blockheights->map, txid);
+	return e ? e->height : 0;
+}
+
+static bool json_hex_to_be32(const char *buffer, const jsmntok_t *tok,
+			     be32 *val)
+{
+	return hex_decode(buffer + tok->start, tok->end - tok->start,
+			  val, sizeof(*val));
+}
+
+struct blockheights *init_blockheights(const tal_t *ctx,
+				       struct command *init_cmd)
+{
+	struct json_out *params = json_out_new(tmpctx);
+	const jsmntok_t *result;
+	const char *buf;
+	const jsmntok_t *datastore, *t;
+	size_t i;
+
+	struct blockheights *bh = tal(ctx, struct blockheights);
+	bh->map = tal(bh, struct blockheight_htable);
+	blockheight_htable_init(bh->map);
+	memleak_add_helper(bh->map, memleak_scan_blockheight_htable);
+
+	/* Query all keys under bookkeeper/blockheights */
+	json_out_start(params, NULL, '{');
+	json_out_start(params, "key", '[');
+	json_out_addstr(params, NULL, "bookkeeper");
+	json_out_addstr(params, NULL, "blockheights");
+	json_out_end(params, ']');
+	json_out_end(params, '}');
+
+	result = jsonrpc_request_sync(tmpctx, init_cmd,
+	                              "listdatastore", params, &buf);
+
+	datastore = json_get_member(buf, result, "datastore");
+	json_for_each_arr(i, t, datastore) {
+		const jsmntok_t *keytok = json_get_member(buf, t, "key");
+		const jsmntok_t *hextok = json_get_member(buf, t, "hex");
+		struct blockheight_entry *e;
+		struct bitcoin_txid txid;
+		be32 be_blockheight;
+
+		/* Expect: ["bookkeeper","blockheights","<txid>"] */
+		if (keytok->size != 3)
+			goto weird;
+
+		if (!json_to_txid(buf, keytok + 2, &txid))
+			goto weird;
+		if (!json_hex_to_be32(buf, hextok, &be_blockheight))
+			goto weird;
+
+		/* Insert into map */
+		e = tal(bh->map, struct blockheight_entry);
+		e->txid = txid;
+		e->height = be32_to_cpu(be_blockheight);
+		blockheight_htable_add(bh->map, e);
+		continue;
+
+weird:
+		plugin_log(init_cmd->plugin, LOG_BROKEN,
+		           "Unparsable blockheight datastore entry: %.*s",
+		           json_tok_full_len(t), json_tok_full(buf, t));
+	}
+
+	return bh;
+}

plugins/bkpr/blockheights.h

@@ -0,0 +1,19 @@
+#ifndef LIGHTNING_PLUGINS_BKPR_BLOCKHEIGHTS_H
+#define LIGHTNING_PLUGINS_BKPR_BLOCKHEIGHTS_H
+#include "config.h"
+
+struct command;
+struct bkpr;
+struct bitcoin_txid;
+
+void add_blockheight(struct command *cmd,
+		     struct bkpr *bkpr,
+		     const struct bitcoin_txid *txid,
+		     u32 blockheight);
+
+/* Returns blockheight for this txid, or 0 if not found. */
+u32 find_blockheight(const struct bkpr *bkpr, const struct bitcoin_txid *txid);
+
+struct blockheights *init_blockheights(const tal_t *ctx,
+				       struct command *init_cmd);
+#endif /* LIGHTNING_PLUGINS_BKPR_BLOCKHEIGHTS_H */

plugins/bkpr/bookkeeper.c

@@ -16,6 +16,7 @@
 #include <errno.h>
 #include <plugins/bkpr/account.h>
 #include <plugins/bkpr/account_entry.h>
+#include <plugins/bkpr/blockheights.h>
 #include <plugins/bkpr/bookkeeper.h>
 #include <plugins/bkpr/chain_event.h>
 #include <plugins/bkpr/channel_event.h>
@@ -1578,7 +1579,7 @@ parse_and_log_chain_move(struct command *cmd,
 		/* Go see if there's any deposits to an external
 		 * that are now confirmed */
 		/* FIXME: might need updating when we can splice? */
-		maybe_closeout_external_deposits(bkpr, e->spending_txid,
+		maybe_closeout_external_deposits(cmd, bkpr, e->spending_txid,
 						 e->blockheight);
 		db_commit_transaction(bkpr->db);
 	}
@@ -1879,7 +1880,7 @@ static struct command_result *json_utxo_spend(struct command *cmd, const char *b
 	/* Go see if there's any deposits to an external
 	 * that are now confirmed */
 	/* FIXME: might need updating when we can splice? */
-	maybe_closeout_external_deposits(bkpr, ev->spending_txid,
+	maybe_closeout_external_deposits(cmd, bkpr, ev->spending_txid,
 					 ev->blockheight);
 	db_commit_transaction(bkpr->db);
 
@@ -2037,6 +2038,7 @@ static const char *init(struct command *init_cmd, const char *b, const jsmntok_t
 	bkpr->onchain_fees = init_onchain_fees(bkpr, init_cmd);
 	bkpr->descriptions = init_descriptions(bkpr, init_cmd);
 	bkpr->rebalances = init_rebalances(bkpr, init_cmd);
+	bkpr->blockheights = init_blockheights(bkpr, init_cmd);
 
 	return NULL;
 }

plugins/bkpr/bookkeeper.h

@@ -13,6 +13,7 @@ struct bkpr {
 	struct onchain_fees *onchain_fees;
 	struct descriptions *descriptions;
 	struct rebalances *rebalances;
+	struct blockheights *blockheights;
 
 	char *db_dsn;
 	char *datadir;

plugins/bkpr/recorder.c

@@ -11,6 +11,7 @@
 #include <inttypes.h>
 #include <plugins/bkpr/account.h>
 #include <plugins/bkpr/account_entry.h>
+#include <plugins/bkpr/blockheights.h>
 #include <plugins/bkpr/bookkeeper.h>
 #include <plugins/bkpr/chain_event.h>
 #include <plugins/bkpr/channel_event.h>
@@ -45,6 +46,9 @@ static struct chain_event *stmt2chain_event(const tal_t *ctx,
 	db_col_txid(stmt, "e.utxo_txid", &e->outpoint.txid);
 	e->outpoint.n = db_col_int(stmt, "e.outnum");
 
+	if (e->blockheight == 0)
+		e->blockheight = find_blockheight(bkpr, &e->outpoint.txid);
+
 	if (!db_col_is_null(stmt, "e.payment_id")) {
 		e->payment_id = tal(e, struct sha256);
 		db_col_sha256(stmt, "e.payment_id", e->payment_id);
@@ -936,7 +940,8 @@ void maybe_record_rebalance(struct command *cmd,
 	tal_free(stmt);
 }
 
-void maybe_closeout_external_deposits(struct bkpr *bkpr,
+void maybe_closeout_external_deposits(struct command *cmd,
+				      struct bkpr *bkpr,
 			              const struct bitcoin_txid *txid,
 				      u32 blockheight)
 {
@@ -944,7 +949,7 @@ void maybe_closeout_external_deposits(struct bkpr *bkpr,
 
 	assert(txid);
 	stmt = db_prepare_v2(bkpr->db, SQL("SELECT "
-				     "  e.id"
+				     "  1"
 				     " FROM chain_events e"
 				     " WHERE e.blockheight = ?"
 				     " AND e.utxo_txid = ?"
@@ -956,18 +961,9 @@ void maybe_closeout_external_deposits(struct bkpr *bkpr,
 	db_bind_text(stmt, ACCOUNT_NAME_EXTERNAL);
 	db_query_prepared(stmt);
 
-	while (db_step(stmt)) {
-		struct db_stmt *update_stmt;
-		u64 id;
-
-		id = db_col_u64(stmt, "e.id");
-		update_stmt = db_prepare_v2(bkpr->db, SQL("UPDATE chain_events SET"
-						    " blockheight = ?"
-						    " WHERE id = ?"));
-
-		db_bind_int(update_stmt, blockheight);
-		db_bind_u64(update_stmt, id);
-		db_exec_prepared_v2(take(update_stmt));
+	if (db_step(stmt)) {
+		db_col_ignore(stmt, "1");
+		add_blockheight(cmd, bkpr, txid, blockheight);
 	}
 
 	tal_free(stmt);

plugins/bkpr/recorder.h

@@ -133,9 +133,10 @@ u64 account_onchain_closeheight(const struct bkpr *bkpr, const struct account *a
  * count them until any output that was spent *into* them is
  * confirmed onchain.
  *
- * This method updates the blockheight on these events to the
+ * This method updates bkpr->blockheights to show the
  * height an input was spent into */
-void maybe_closeout_external_deposits(struct bkpr *bkpr,
+void maybe_closeout_external_deposits(struct command *cmd,
+				      struct bkpr *bkpr,
 				      const struct bitcoin_txid *txid,
 				      u32 blockheight);
 

plugins/bkpr/test/run-recorder.c

@@ -31,6 +31,7 @@
 
 #include "plugins/bkpr/db.c"
 #include "plugins/bkpr/account.c"
+#include "plugins/bkpr/blockheights.c"
 #include "plugins/bkpr/recorder.c"
 #include "plugins/bkpr/onchain_fee.c"
 #include "plugins/bkpr/rebalances.c"
@@ -68,6 +69,10 @@ struct command_result *ignore_datastore_reply(struct command *cmd UNNEEDED,
 					      const jsmntok_t *result UNNEEDED,
 					      void *arg UNNEEDED)
 { fprintf(stderr, "ignore_datastore_reply called!\n"); abort(); }
+/* Generated stub for json_to_txid */
+bool json_to_txid(const char *buffer UNNEEDED, const jsmntok_t *tok UNNEEDED,
+		  struct bitcoin_txid *txid UNNEEDED)
+{ fprintf(stderr, "json_to_txid called!\n"); abort(); }
 /* Generated stub for json_tok_bin_from_hex */
 u8 *json_tok_bin_from_hex(const tal_t *ctx UNNEEDED, const char *buffer UNNEEDED, const jsmntok_t *tok UNNEEDED)
 { fprintf(stderr, "json_tok_bin_from_hex called!\n"); abort(); }