connectd: Implement sending of start_batch

Implement the sending of `start_batch` and `protocol_batch_element` from `channeld` to `connectd`.

Each real peer wire message is prefixed with `protocol_batch_element` so connectd can know the size of the message that were batched together.

`connectd` intercepts `protocol_batch_element` messages and eats them (doesn’t forward them to peer) to get individual messages out of the batch.

It needs this to be able to encrypt them individiaully. Afterwards it recombines the now encrypted messages into a single message to send over the wire to the peer.

`channeld` remains responsible for making `start_batch` the first message of the message bundle.

Author: ddustin

channeld/channeld.c

@@ -1232,6 +1232,78 @@ static s64 sats_diff(struct amount_sat a, struct amount_sat b)
         return (s64)a.satoshis - (s64)b.satoshis; /* Raw: splicing numbers can wrap! */
 }
 
+static void send_message_batch(struct peer *peer, u8 **msgs)
+{
+	size_t size;
+	size_t hdr_size = tal_bytelen(towire_protocol_batch_element(tmpctx,
+								    &peer->channel_id,
+								    0));
+	u8 *batch_msg, *final_msg, *final_msg_ptr;
+	struct tlv_start_batch_tlvs *tlvs;
+
+	assert(tal_count(msgs) > 0);
+
+	/* When sending one message, no batching is required */
+	if (tal_count(msgs) == 1) {
+		peer_write(peer->pps, msgs[0]);
+		return;
+	}
+
+	/* We prefix each message with an interal wire type,
+	 * protocol_batch_element. connectd will eat each message so they don't
+	 * actually go out to the peer. It's just so connectd can chop up the
+	 * message batch back out into individual messages. */
+
+	/* We start by calculating the total size */
+	size = 0;
+
+	/* Build the `start_batch` msg now so know it's size */
+	tlvs = tlv_start_batch_tlvs_new(tmpctx);
+	tlvs->batch_info = tal(tlvs, u16);
+	*tlvs->batch_info = WIRE_COMMITMENT_SIGNED;
+	batch_msg = towire_start_batch(tmpctx, &peer->channel_id,
+				       tal_count(msgs), tlvs);
+	size += tal_bytelen(batch_msg) + hdr_size;
+
+	/* Count the size of all the messages in the batch */
+	for(u32 i = 0; i < tal_count(msgs); i++)
+		size += tal_bytelen(msgs[i]) + hdr_size;
+
+	/* Now we know the size of our `final_msg` so we allocate */
+	final_msg = tal_arr(tmpctx, u8, size);
+	final_msg_ptr = final_msg;
+
+	status_debug("proto_batch Building batch with %zu bytes, msgs: %zu",
+		     size, tal_count(msgs));
+
+	/* Copy the bytes for `start_batch` prefix */
+	memcpy(final_msg_ptr,
+	       towire_protocol_batch_element(tmpctx,
+	       				     &peer->channel_id,
+	       				     tal_bytelen(batch_msg)),
+	       				     hdr_size);
+	final_msg_ptr += hdr_size;
+
+	memcpy(final_msg_ptr, batch_msg, tal_bytelen(batch_msg));
+	final_msg_ptr += tal_bytelen(batch_msg);
+
+	/* Now copy the bytes from all messages in `msgs` */
+	for(u32 i = 0; i < tal_count(msgs); i++) {
+		memcpy(final_msg_ptr,
+		       towire_protocol_batch_element(tmpctx,
+	       					     &peer->channel_id,
+		       				     tal_bytelen(msgs[i])),
+		       				     hdr_size);
+		final_msg_ptr += hdr_size;
+
+		memcpy(final_msg_ptr, msgs[i], tal_bytelen(msgs[i]));
+		final_msg_ptr += tal_bytelen(msgs[i]);
+	}
+
+	assert(final_msg + size == final_msg_ptr);
+	peer_write(peer->pps, take(final_msg));
+}
+
 static void send_commit(struct peer *peer)
 {
 	const struct htlc **changed_htlcs;
@@ -1398,8 +1470,7 @@ static void send_commit(struct peer *peer)
 
 	peer->next_index[REMOTE]++;
 
-	for(u32 i = 0; i < tal_count(msgs); i++)
-		peer_write(peer->pps, take(msgs[i]));
+	send_message_batch(peer, msgs);
 
 	maybe_send_shutdown(peer);
 
@@ -5195,8 +5266,7 @@ static void resend_commitment(struct peer *peer, struct changed_htlc *last)
 						peer->splice_state->inflights[i]->remote_funding));
 	}
 
-	for(i = 0; i < tal_count(msgs); i++)
-		peer_write(peer->pps, take(msgs[i]));
+	send_message_batch(peer, msgs);
 
 	/* If we have already received the revocation for the previous, the
 	 * other side shouldn't be asking for a retransmit! */

connectd/multiplex.c

@@ -399,6 +399,69 @@ static struct io_plan *io_sock_shutdown_cb(struct io_conn *conn, struct peer *un
 	return io_sock_shutdown(conn);
 }
 
+/* Process and eat protocol_batch_element messages, encrypt each element message
+ * and return the encrypted messages as one long byte array. */
+static u8 *process_batch_elements(struct peer *peer, const u8 *msg TAKES)
+{
+	u8 *ret = tal_arr(peer, u8, 0);
+	size_t ret_size = 0;
+	const u8 *cursor = msg;
+	size_t plen = tal_count(msg);
+
+	status_debug("Processing batch elements of %zu bytes. %s", plen,
+		     tal_hex(tmpctx, msg));
+
+	do {
+		u8 *element_bytes;
+		u16 element_size;
+		struct channel_id channel_id;
+		u8 *enc_msg;
+
+		if (fromwire_u16(&cursor, &plen) != WIRE_PROTOCOL_BATCH_ELEMENT) {
+			status_broken("process_batch_elements on msg that is"
+				      " not WIRE_PROTOCOL_BATCH_ELEMENT. %s",
+				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+		}
+
+		fromwire_channel_id(&cursor, &plen, &channel_id);
+
+	 	element_size = fromwire_u16(&cursor, &plen);
+	 	if (!element_size) {
+			status_broken("process_batch_elements cannot have zero"
+				      " length elements. %s",
+				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+	 	}
+
+		element_bytes = fromwire_tal_arrn(NULL, &cursor, &plen,
+						  element_size);
+		if (!element_bytes) {
+			status_broken("process_batch_elements fromwire_tal_arrn"
+				      " %s",
+				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+		}
+
+		status_debug("Processing batch extracted item %s. %s",
+			     peer_wire_name(fromwire_peektype(element_bytes)),
+			     tal_hex(tmpctx, element_bytes));
+
+		enc_msg = cryptomsg_encrypt_msg(tmpctx, &peer->cs,
+						take(element_bytes));
+
+		tal_resize(&ret, ret_size + tal_bytelen(enc_msg));
+		memcpy(&ret[ret_size], enc_msg, tal_bytelen(enc_msg));
+		ret_size += tal_bytelen(enc_msg);
+
+	} while(plen);
+
+	if (taken(msg))
+		tal_free(msg);
+
+	return ret;
+}
+
 static struct io_plan *encrypt_and_send(struct peer *peer,
 					const u8 *msg TAKES,
 					struct io_plan *(*next)
@@ -442,8 +505,17 @@ static struct io_plan *encrypt_and_send(struct peer *peer,
 
 	set_urgent_flag(peer, is_urgent(type));
 
+	/* Special message type directing us to process batch items. */
+	if (type == WIRE_PROTOCOL_BATCH_ELEMENT) {
+		peer->sent_to_peer = process_batch_elements(peer, msg);
+		if (!peer->sent_to_peer)
+			return io_close(peer->to_peer);
+	}
+	else {
+		peer->sent_to_peer = cryptomsg_encrypt_msg(peer, &peer->cs, msg);
+	}
 	/* We free this and the encrypted version in next write_to_peer */
-	peer->sent_to_peer = cryptomsg_encrypt_msg(peer, &peer->cs, msg);
+
 	return io_write(peer->to_peer,
 			peer->sent_to_peer,
 			tal_bytelen(peer->sent_to_peer),