connectd: flush queue to subd once peer hangs up.

This is critical in the common case where peer sends an error and
hangs up: we almost never get to relay the error to the subd in time.

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

connectd/connectd.c

@@ -364,12 +364,15 @@ static struct peer *new_peer(struct daemon *daemon,
 	peer->dev_read_enabled = true;
 #endif
 
+	peer->to_peer = conn;
+
 	/* Aim for connection to shuffle data back and forth: sets up
 	 * peer->to_subd */
 	if (!multiplex_subd_setup(peer, fd_for_subd))
 		return tal_free(peer);
 
-	peer->to_peer = tal_steal(peer, conn);
+	/* Now we own it */
+	tal_steal(peer, peer->to_peer);
 	peer_htable_add(&daemon->peers, peer);
 	tal_add_destructor2(peer, destroy_peer, daemon);
 

connectd/multiplex.c

@@ -51,6 +51,7 @@ static void send_warning(struct peer *peer, const char *fmt, ...)
 
 	/* Close locally, send msg as final warning */
 	io_close(peer->to_subd);
+	peer->to_subd = NULL;
 
 	va_start(ap, fmt);
 	peer->final_msg = towire_warningfmtv(peer, NULL, fmt, ap);
@@ -480,6 +481,10 @@ static struct io_plan *write_to_subd(struct io_conn *subd_conn,
 
 	/* Nothing to send? */
 	if (!msg) {
+		/* If peer is closed, close this. */
+		if (!peer->to_peer)
+			return io_close(subd_conn);
+
 		/* Tell them to read again. */
 		io_wake(&peer->peer_in);
 
@@ -579,6 +584,10 @@ static void destroy_subd_conn(struct io_conn *subd_conn, struct peer *peer)
 	/* In case they were waiting for this to send final_msg */
 	if (peer->final_msg)
 		msg_wake(peer->peer_outq);
+
+	/* If no peer, finally time to close */
+	if (!peer->to_peer && peer->closing)
+		peer_conn_closed(peer);
 }
 
 void close_peer_conn(struct peer *peer)
@@ -610,9 +619,11 @@ static void destroy_peer_conn(struct io_conn *peer_conn, struct peer *peer)
 	assert(peer->to_peer == peer_conn);
 	peer->to_peer = NULL;
 
-	/* Close internal connections if not already. */
-	if (peer->to_subd)
-		io_close(peer->to_subd);
+	/* Flush internal connections if not already. */
+	if (peer->to_subd) {
+		msg_wake(peer->subd_outq);
+		return;
+	}
 
 	if (peer->closing)
 		peer_conn_closed(peer);