Splice: Complete tx_abort implementation

channeld/channeld_wire.csv

@@ -10,6 +10,7 @@
 #include <common/derive_basepoints.h>
 #include <common/features.h>
 #include <common/fee_states.h>
+#include <wire/peer_wire.h>
 
 # Begin!  (passes gossipd-client fd)
 msgtype,channeld_init,1000
@@ -280,6 +281,12 @@ msgdata,channeld_splice_funding_error,opener_error,bool,
 msgtype,channeld_splice_state_error,7221
 msgdata,channeld_splice_state_error,state_error,wirestring,
 
+# channeld->master: Peer rejected our splice
+msgtype,channeld_splice_abort,7223
+msgdata,channeld_splice_abort,did_i_initiate,bool,
+msgdata,channeld_splice_abort,inflight_outpoint,?bitcoin_outpoint,
+msgdata,channeld_splice_abort,reason,?wirestring,
+
 # Tell peer to shut down channel.
 msgtype,channeld_send_shutdown,1023
 msgdata,channeld_send_shutdown,final_index,?u32,

channeld/splice.c

@@ -30,6 +30,7 @@ struct splicing *splicing_new(const tal_t *ctx)
 	splicing->current_psbt = NULL;
 	splicing->received_tx_complete = false;
 	splicing->sent_tx_complete = false;
+	splicing->tx_sig_msg = NULL;
 
 	return splicing;
 }

channeld/splice.h

@@ -51,6 +51,8 @@ struct splicing {
 	bool received_tx_complete;
 	/* If, in the last splice_update, we sent tx_complete */
 	bool sent_tx_complete;
+	/* If our peer signs early, we allow that and cache it here */
+	const u8 *tx_sig_msg;
 };
 
 /* Sets `splice` items to default values */

common/interactivetx.c

@@ -128,8 +128,7 @@ static u8 *read_next_msg(const tal_t *ctx,
 		desc = is_peer_warning(msg, msg);
 		if (desc) {
 			status_info("They sent %s", desc);
-			tal_free(msg);
-			continue;
+			return tal_free(msg);
 		}
 
 		/* In theory, we're in the middle of an open/RBF/splice, but
@@ -142,9 +141,8 @@ static u8 *read_next_msg(const tal_t *ctx,
 		case WIRE_TX_ADD_OUTPUT:
 		case WIRE_TX_REMOVE_OUTPUT:
 		case WIRE_TX_COMPLETE:
-			return msg;
 		case WIRE_TX_ABORT:
-			/* TODO */
+			return msg;
 		case WIRE_TX_SIGNATURES:
 		case WIRE_CHANNEL_READY:
 		case WIRE_TX_INIT_RBF:
@@ -345,13 +343,16 @@ bool interactivetx_has_changes(struct interactivetx_context *ictx,
 
 char *process_interactivetx_updates(const tal_t *ctx,
 			 	    struct interactivetx_context *ictx,
-				    bool *received_tx_complete)
+				    bool *received_tx_complete,
+				    u8 **abort_msg)
 {
 	bool we_complete = false, they_complete = false;
 	u8 *msg;
 	char *error = NULL;
 	struct wally_psbt *next_psbt;
 
+	*abort_msg = NULL;
+
 	if (received_tx_complete)
 		they_complete = *received_tx_complete;
 
@@ -696,7 +697,8 @@ char *process_interactivetx_updates(const tal_t *ctx,
 				*received_tx_complete = true;
 			break;
 		case WIRE_TX_ABORT:
-			/* Todo */
+			*abort_msg = msg;
+			return NULL;
 		case WIRE_INIT:
 		case WIRE_ERROR:
 		case WIRE_WARNING:

common/interactivetx.h

@@ -77,10 +77,13 @@ struct interactivetx_context *new_interactivetx_context(const tal_t *ctx,
  * out -> true means the last message from the peer was 'tx_complete'.
  *
  * Returns NULL on success or a description of the error on failure.
+ *
+ * If `tx_abort` is received, NULL is returned and `abort_msg` will be set to
  */
 char *process_interactivetx_updates(const tal_t *ctx,
 				    struct interactivetx_context *ictx,
-				    bool *received_tx_complete);
+				    bool *received_tx_complete,
+				    u8 **abort_msg);
 
 /* If the given ictx would cause `process_interactivetx_updates to send tx
  * changes when called. Returns true if an error occurs

common/jsonrpc_errors.h

@@ -77,6 +77,7 @@ enum jsonrpc_errcode {
 	SPLICE_STATE_ERROR = 358,
 	SPLICE_LOW_FEE = 359,
 	SPLICE_HIGH_FEE = 360,
+	SPLICE_ABORT = 362,
 
 	/* `connect` errors */
 	CONNECT_NO_KNOWN_ADDRESS = 400,

common/wire_error.c

@@ -84,17 +84,17 @@ char *sanitize_error(const tal_t *ctx, const u8 *errmsg,
 	struct channel_id dummy;
 	u8 *data;
 	size_t i;
-	bool warning;
+	const char *tag;
 
 	if (!channel_id)
 		channel_id = &dummy;
 
 	if (fromwire_error(ctx, errmsg, channel_id, &data))
-		warning = false;
+		tag = "ERROR";
 	else if (fromwire_warning(ctx, errmsg, channel_id, &data))
-		warning = true;
+		tag = "WARNING";
 	else if (fromwire_tx_abort(ctx, errmsg, channel_id, &data))
-		warning = true;
+		tag = "ABORT";
 	else
 		return tal_fmt(ctx, "Invalid ERROR message '%s'",
 			       tal_hex(ctx, errmsg));
@@ -118,7 +118,7 @@ char *sanitize_error(const tal_t *ctx, const u8 *errmsg,
 	}
 
 	return tal_fmt(ctx, "%s%s%s: %.*s",
-		       warning ? "WARNING" : "ERROR",
+		       tag,
 		       channel_id_is_all(channel_id) ? "": " channel ",
 		       channel_id_is_all(channel_id) ? ""
 		       : type_to_string(tmpctx, struct channel_id, channel_id),

lightningd/channel_control.c

@@ -284,6 +284,99 @@ static void handle_splice_feerate_error(struct lightningd *ld,
 	}
 }
 
+static void handle_splice_abort(struct lightningd *ld,
+				struct channel *channel,
+				const u8 *msg)
+{
+	struct splice_command *cc;
+	struct peer *peer = channel->peer;
+	bool did_i_abort;
+	struct bitcoin_outpoint *outpoint;
+	struct channel_inflight *inflight;
+	char *reason;
+	u8 *error;
+	int fds[2];
+
+	if (!fromwire_channeld_splice_abort(tmpctx, msg, &did_i_abort,
+					    &outpoint, &reason)) {
+		channel_internal_error(channel,
+				       "bad fromwire_channeld_splice_abort %s",
+				       tal_hex(channel, msg));
+		return;
+	}
+
+	if (outpoint) {
+		inflight = list_tail(&channel->inflights,
+				     struct channel_inflight,
+				     list);
+
+		if (!bitcoin_outpoint_eq(outpoint,
+					 &inflight->funding->outpoint))
+			channel_internal_error(channel,
+					       "abort outpoint %s does not"
+					       " match ours %s",
+					       type_to_string(tmpctx,
+					       		      struct bitcoin_outpoint,
+					       		      outpoint),
+					       type_to_string(tmpctx,
+					       		      struct bitcoin_outpoint,
+					       		      &inflight->funding->outpoint));
+
+		wallet_inflight_del(ld->wallet, channel, inflight);
+		tal_free(inflight);
+	}
+
+	cc = splice_command_for_chan(ld, channel);
+	if (cc)
+		was_pending(command_fail(cc->cmd, SPLICE_ABORT, "%s", reason));
+	else
+		log_peer_unusual(ld->log, &peer->id, "Splice aborted"
+				 " %s", reason);
+
+	log_debug(channel->log,
+		  "Restarting channeld after tx_abort on %s channel",
+		  channel_state_name(channel));
+
+	if (socketpair(AF_LOCAL, SOCK_STREAM, 0, fds) != 0) {
+		log_broken(channel->log,
+			   "Failed to create socketpair: %s",
+			   strerror(errno));
+
+		error = towire_warningfmt(tmpctx, &channel->cid,
+					  "Trouble in paradise?");
+		log_peer_debug(ld->log, &channel->peer->id,
+			       "Telling connectd to send error %s",
+			       tal_hex(tmpctx, error));
+		/* Get connectd to send error and close. */
+		subd_send_msg(ld->connectd,
+			      take(towire_connectd_peer_send_msg(NULL,
+								 &peer->id,
+								 peer->connectd_counter,
+								 error)));
+		subd_send_msg(ld->connectd,
+			      take(towire_connectd_discard_peer(NULL,
+								&peer->id,
+								peer->connectd_counter)));
+		return;
+	}
+	log_debug(channel->log, "made the socket pair");
+
+	if (peer_start_channeld(channel, new_peer_fd(tmpctx, fds[0]), NULL,
+						     true, false)) {
+		log_info(channel->log, "Sending the peer fd to connectd");
+		subd_send_msg(ld->connectd,
+			      take(towire_connectd_peer_connect_subd(NULL,
+			      					     &peer->id,
+								     peer->connectd_counter,
+								     &channel->cid)));
+		subd_send_fd(ld->connectd, fds[1]);
+		log_info(channel->log, "Sent the peer fd to channeld");
+	} else {
+		log_info(channel->log, "peer_start_channeld failed");
+		close(fds[1]);
+	}
+}
+
 /* When channeld finishes processing the `splice_init` command, this is called */
 static void handle_splice_confirmed_init(struct lightningd *ld,
 					 struct channel *channel,
@@ -1366,6 +1459,9 @@ static unsigned channel_msg(struct subd *sd, const u8 *msg, const int *fds)
 	case WIRE_CHANNELD_SPLICE_FUNDING_ERROR:
 		handle_splice_funding_error(sd->ld, sd->channel, msg);
 		break;
+	case WIRE_CHANNELD_SPLICE_ABORT:
+		handle_splice_abort(sd->ld, sd->channel, msg);
+		break;
 	case WIRE_CHANNELD_SPLICE_STATE_ERROR:
 		handle_splice_state_error(sd->ld, sd->channel, msg);
 		break;

tests/test_connection.py

@@ -396,18 +396,18 @@ def test_opening_tiny_channel(node_factory, anchors):
     l1.rpc.connect(l3.info['id'], 'localhost', l3.port)
     l1.rpc.connect(l4.info['id'], 'localhost', l4.port)
 
-    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING).*channel capacity is .*, which is below .*sat'):
+    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING|ABORT).*channel capacity is .*, which is below .*sat'):
         l1.fundchannel(l2, l2_min_capacity + overhead - 1)
     assert only_one(l1.rpc.listpeers(l2.info['id'])['peers'])['connected']
 
     l1.fundchannel(l2, l2_min_capacity + overhead)
 
-    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING).*channel capacity is .*, which is below .*sat'):
+    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING|ABORT).*channel capacity is .*, which is below .*sat'):
         l1.fundchannel(l3, l3_min_capacity + overhead - 1)
     assert only_one(l1.rpc.listpeers(l3.info['id'])['peers'])['connected']
     l1.fundchannel(l3, l3_min_capacity + overhead)
 
-    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING).*channel capacity is .*, which is below .*sat'):
+    with pytest.raises(RpcError, match=r'They sent (ERROR|WARNING|ABORT).*channel capacity is .*, which is below .*sat'):
         l1.fundchannel(l4, l4_min_capacity + overhead - 1)
     assert only_one(l1.rpc.listpeers(l4.info['id'])['peers'])['connected']
     l1.fundchannel(l4, l4_min_capacity + overhead)

tests/test_splicing.py

@@ -196,7 +196,7 @@ def test_invalid_splice(node_factory, bitcoind):
     # The splicing inflight should not have been left pending in the DB
     assert l1.db_query("SELECT count(*) as c FROM channel_funding_inflights;")[0]['c'] == 0
 
-    l1.daemon.wait_for_log(r'Peer has reconnected, state CHANNELD_NORMAL')
+    l1.daemon.wait_for_log(r'Restarting channeld after tx_abort on CHANNELD_NORMAL channel')
 
     assert l1.db_query("SELECT count(*) as c FROM channel_funding_inflights;")[0]['c'] == 0
 
@@ -250,7 +250,7 @@ def test_commit_crash_splice(node_factory, bitcoind):
     # The splicing inflight should have been left pending in the DB
     assert l1.db_query("SELECT count(*) as c FROM channel_funding_inflights;")[0]['c'] == 1
 
-    l1.daemon.wait_for_log(r'Peer has reconnected, state CHANNELD_NORMAL')
+    l1.daemon.wait_for_log(r'Restarting channeld after tx_abort on CHANNELD_NORMAL channel')
 
     assert l1.db_query("SELECT count(*) as c FROM channel_funding_inflights;")[0]['c'] == 1
 

tests/test_splicing_disconnect.py

@@ -100,17 +100,14 @@ def test_splice_disconnect_commit(node_factory, bitcoind, executor):
     # Should reconnect, and reestablish the splice.
     l2.start()
 
+    # Splice should be abandoned via tx_abort
+
     # Wait until nodes are reconnected
     l1.daemon.wait_for_log(r'peer_in WIRE_CHANNEL_REESTABLISH')
     l2.daemon.wait_for_log(r'peer_in WIRE_CHANNEL_REESTABLISH')
 
-    bitcoind.generate_block(6, wait_for_mempool=1)
-
-    l1.daemon.wait_for_log(r'CHANNELD_AWAITING_SPLICE to CHANNELD_NORMAL')
-    l2.daemon.wait_for_log(r'CHANNELD_AWAITING_SPLICE to CHANNELD_NORMAL')
-
-    inv = l2.rpc.invoice(10**2, '3', 'no_3')
-    l1.rpc.pay(inv['bolt11'])
+    l1.daemon.wait_for_log(r'peer_in WIRE_CHANNEL_READY')
+    l2.daemon.wait_for_log(r'peer_in WIRE_CHANNEL_READY')
 
     # Check that the splice doesn't generate a unilateral close transaction
     time.sleep(5)

wallet/wallet.c

@@ -1259,6 +1259,22 @@ void wallet_inflight_add(struct wallet *w, struct channel_inflight *inflight)
 	tal_free(stmt);
 }
 
+void wallet_inflight_del(struct wallet *w, const struct channel *chan,
+			 const struct channel_inflight *inflight)
+{
+	struct db_stmt *stmt;
+
+	/* Remove inflight from the channel */
+	stmt = db_prepare_v2(w->db, SQL("DELETE FROM channel_funding_inflights"
+					" WHERE channel_id = ?"
+					"   AND funding_tx_id = ?"
+					"   AND funding_tx_outnum = ?"));
+	db_bind_u64(stmt, chan->dbid);
+	db_bind_txid(stmt, &inflight->funding->outpoint.txid);
+	db_bind_int(stmt, inflight->funding->outpoint.n);
+	db_exec_prepared_v2(take(stmt));
+}
+
 void wallet_inflight_save(struct wallet *w,
 			  struct channel_inflight *inflight)
 {

wallet/wallet.h

@@ -620,6 +620,12 @@ void wallet_channel_insert(struct wallet *w, struct channel *chan);
  */
 void wallet_inflight_add(struct wallet *w, struct channel_inflight *inflight);
 
+/**
+ * Delete an inflight transaction for a channel
+ */
+void wallet_inflight_del(struct wallet *w, const struct channel *chan,
+			 const struct channel_inflight *inflight);
+
 /**
  * Update an existing inflight channel transaction
  */