Fix latency spike when old high-traffic channels close.

lightningd/lightningd.c

@@ -1371,6 +1371,9 @@ int main(int argc, char *argv[])
 	trace_span_end(ld->topology);
 
 	db_begin_transaction(ld->wallet->db);
+	trace_span_start("delete_old_htlcs", ld->wallet);
+	wallet_delete_old_htlcs(ld->wallet);
+	trace_span_end(ld->wallet);
 
 	/*~ Pull peers, channels and HTLCs from db. Needs to happen after the
 	 *  topology is initialized since some decisions rely on being able to

lightningd/test/run-find_my_abspath.c

@@ -286,6 +286,9 @@ void waitblockheight_notify_new_block(struct lightningd *ld UNNEEDED)
 /* Generated stub for wallet_begin_old_close_rescan */
 void wallet_begin_old_close_rescan(struct lightningd *ld UNNEEDED)
 { fprintf(stderr, "wallet_begin_old_close_rescan called!\n"); abort(); }
+/* Generated stub for wallet_delete_old_htlcs */
+void wallet_delete_old_htlcs(struct wallet *w UNNEEDED)
+{ fprintf(stderr, "wallet_delete_old_htlcs called!\n"); abort(); }
 /* Generated stub for wallet_new */
 struct wallet *wallet_new(struct lightningd *ld UNNEEDED, struct timers *timers UNNEEDED)
 { fprintf(stderr, "wallet_new called!\n"); abort(); }

tests/test_misc.py

@@ -3461,6 +3461,9 @@ def test_listforwards_and_listhtlcs(node_factory, bitcoind):
     l2.rpc.delforward(in_channel=c12, in_htlc_id=2, status='local_failed')
     assert l2.rpc.listforwards() == {'forwards': []}
 
+    l2.restart()
+    assert l2.rpc.wait('htlcs', 'deleted', 0)['deleted'] == 5
+
 
 def test_listforwards_wait(node_factory, executor):
     l1, l2, l3 = node_factory.line_graph(3, wait_for_announce=True)

tests/test_wallet.py

@@ -1885,3 +1885,26 @@ def test_onchain_missing_no_p2tr_migrate(node_factory, bitcoind):
 
     # This can actually take a while for 100 blocks!
     l2.daemon.wait_for_log('Rescan finished! 1 outputs recovered')
+
+
+def test_old_htlcs_cleanup(node_factory, bitcoind):
+    """We lazily delete htlcs from channel_htlcs table"""
+    l1, l2 = node_factory.line_graph(2)
+
+    for _ in range(10):
+        l1.pay(l2, 1000)
+
+    l1.rpc.close(l2.info['id'])
+    bitcoind.generate_block(100, wait_for_mempool=1)
+    wait_for(lambda: l1.rpc.listpeerchannels() == {'channels': []})
+    # We don't see them!
+    assert l1.rpc.listhtlcs() == {'htlcs': []}
+
+    l1.stop()
+    # They're still there.
+    assert l1.db_query('SELECT COUNT(*) as c FROM channel_htlcs')[0]['c'] == 10
+
+    l1.start()
+    # Now they're not
+    assert l1.db_query('SELECT COUNT(*) as c FROM channel_htlcs')[0]['c'] == 0
+    assert l1.rpc.listhtlcs() == {'htlcs': []}

wallet/wallet.c

@@ -2957,6 +2957,21 @@ void wallet_channel_insert(struct wallet *w, struct channel *chan)
 	wallet_channel_save(w, chan);
 }
 
+void wallet_delete_old_htlcs(struct wallet *w)
+{
+	struct db_stmt *stmt;
+
+	/* Delete htlcs for closed channels */
+	stmt = db_prepare_v2(w->db, SQL("DELETE FROM channel_htlcs"
+					" WHERE id IN ("
+					"  SELECT ch.id"
+					"  FROM channel_htlcs AS ch"
+					"  JOIN channels AS c ON c.id = ch.channel_id"
+					"  WHERE c.state = ?);"));
+	db_bind_int(stmt, channel_state_in_db(CLOSED));
+	db_exec_prepared_v2(take(stmt));
+}
+
 void wallet_channel_close(struct wallet *w,
 			  const struct channel *chan)
 {
@@ -2967,15 +2982,21 @@ void wallet_channel_close(struct wallet *w,
 	 * dbs to recover. */
 	struct db_stmt *stmt;
 	u64 new_move_id;
+	u64 htlcs;
 
-	/* Delete entries from `channel_htlcs` */
-	stmt = db_prepare_v2(w->db, SQL("DELETE FROM channel_htlcs "
+	/* The channel_htlcs table is quite large, and deleting it can take a
+	 * while.  So we do that on next restart by calling
+	 * wallet_delete_old_htlcs.  But update delete count in case anyone
+	 * is watching. */
+	stmt = db_prepare_v2(w->db, SQL("SELECT COUNT(*) FROM channel_htlcs "
 					"WHERE channel_id=?"));
 	db_bind_u64(stmt, chan->dbid);
-	db_exec_prepared_v2(stmt);
-	/* FIXME: We don't actually tell them what was deleted! */
-	if (db_count_changes(stmt) != 0)
-		htlcs_index_deleted(w->ld, chan, db_count_changes(stmt));
+	db_query_prepared(stmt);
+	db_step(stmt);
+
+	htlcs = db_col_u64(stmt, "COUNT(*)");
+	if (htlcs != 0)
+		htlcs_index_deleted(w->ld, chan, htlcs);
 	tal_free(stmt);
 
 	/* Delete entries from `htlc_sigs` */
@@ -6558,7 +6579,8 @@ struct wallet_htlc_iter *wallet_htlcs_first(const tal_t *ctx,
 					    ", h.updated_index"
 					    " FROM channel_htlcs h"
 					    " JOIN channels ON channels.id = h.channel_id"
-					    " WHERE h.updated_index >= ?"
+					    " WHERE channels.state != ?"
+					    "   AND h.updated_index >= ?"
 					    " ORDER BY h.updated_index ASC"
 					    " LIMIT ?;"));
 		} else {
@@ -6575,10 +6597,12 @@ struct wallet_htlc_iter *wallet_htlcs_first(const tal_t *ctx,
 					    ", h.updated_index"
 					    " FROM channel_htlcs h"
 					    " JOIN channels ON channels.id = h.channel_id"
-					    " WHERE h.id >= ?"
+					    " WHERE channels.state != ?"
+					    "   AND h.id >= ?"
 					    " ORDER BY h.id ASC"
 					    " LIMIT ?;"));
 		}
+		db_bind_int(i->stmt, channel_state_in_db(CLOSED));
 	}
 	db_bind_u64(i->stmt, liststart);
 	if (listlimit)

wallet/wallet.h

@@ -725,6 +725,14 @@ void wallet_state_change_add(struct wallet *w,
  */
 void wallet_delete_peer_if_unused(struct wallet *w, u64 peer_dbid);
 
+/**
+ * wallet_delete_old_htlcs -- delete htlcs associated with CLOSED channels.
+ *
+ * We do this at startup, instead of when we finally CLOSED a channel, to
+ * avoid a significant pause.
+ */
+void wallet_delete_old_htlcs(struct wallet *w);
+
 /**
  * wallet_init_channels -- Loads active channels into peers
  *    and inits the dbid counter for next channel.