More flake fixes!

common/dev_disconnect.c

@@ -50,20 +50,47 @@ void dev_disconnect_init(int fd)
 	dev_disconnect_fd = fd;
 }
 
-enum dev_disconnect dev_disconnect(const struct node_id *id, int pkt_type)
+enum dev_disconnect_out dev_disconnect_out(const struct node_id *id, int pkt_type)
 {
 	if (dev_disconnect_fd == -1)
-		return DEV_DISCONNECT_NORMAL;
+		return DEV_DISCONNECT_OUT_NORMAL;
 
 	if (!dev_disconnect_count)
 		next_dev_disconnect();
 
 	if (!dev_disconnect_line[0]
+	    || dev_disconnect_line[0] == DEV_DISCONNECT_IN_AFTER_RECV
 	    || !streq(peer_wire_name(pkt_type), dev_disconnect_line+1))
-		return DEV_DISCONNECT_NORMAL;
+		return DEV_DISCONNECT_OUT_NORMAL;
 
 	if (--dev_disconnect_count != 0) {
-		return DEV_DISCONNECT_NORMAL;
+		return DEV_DISCONNECT_OUT_NORMAL;
+	}
+
+	if (lseek(dev_disconnect_fd, dev_disconnect_len+1, SEEK_CUR) < 0) {
+		err(1, "lseek failure");
+	}
+
+	status_peer_debug(id, "dev_disconnect: %s (%s)",
+			  dev_disconnect_line,
+			  peer_wire_name(pkt_type));
+	return dev_disconnect_line[0];
+}
+
+enum dev_disconnect_in dev_disconnect_in(const struct node_id *id, int pkt_type)
+{
+	if (dev_disconnect_fd == -1)
+		return DEV_DISCONNECT_IN_NORMAL;
+
+	if (!dev_disconnect_count)
+		next_dev_disconnect();
+
+	if (dev_disconnect_line[0] != DEV_DISCONNECT_IN_AFTER_RECV
+	    || !streq(peer_wire_name(pkt_type), dev_disconnect_line+1))
+		return DEV_DISCONNECT_IN_NORMAL;
+
+	if (--dev_disconnect_count != 0) {
+		return DEV_DISCONNECT_IN_NORMAL;
 	}
 
 	if (lseek(dev_disconnect_fd, dev_disconnect_len+1, SEEK_CUR) < 0) {

common/dev_disconnect.h

@@ -5,23 +5,33 @@
 
 struct node_id;
 
-enum dev_disconnect {
+enum dev_disconnect_out {
 	/* Do nothing. */
-	DEV_DISCONNECT_NORMAL = '=',
+	DEV_DISCONNECT_OUT_NORMAL = '=',
 	/* Close connection before sending packet. */
-	DEV_DISCONNECT_BEFORE = '-',
+	DEV_DISCONNECT_OUT_BEFORE = '-',
 	/* Close connection after sending packet. */
-	DEV_DISCONNECT_AFTER = '+',
+	DEV_DISCONNECT_OUT_AFTER = '+',
 	/* Drop message (don't send to peer) */
-	DEV_DISCONNECT_DROP = '$',
+	DEV_DISCONNECT_OUT_DROP = '$',
 	/* Swallow all writes from now on, and do no more reads. */
-	DEV_DISCONNECT_BLACKHOLE = '0',
+	DEV_DISCONNECT_OUT_BLACKHOLE = '0',
 	/* Don't use connection after sending packet, but don't close. */
-	DEV_DISCONNECT_DISABLE_AFTER = 'x',
+	DEV_DISCONNECT_OUT_DISABLE_AFTER = 'x',
 };
 
 /* Force a close fd before or after a certain packet type */
-enum dev_disconnect dev_disconnect(const struct node_id *id, int pkt_type);
+enum dev_disconnect_out dev_disconnect_out(const struct node_id *id, int pkt_type);
+
+enum dev_disconnect_in {
+	/* Do nothing. */
+	DEV_DISCONNECT_IN_NORMAL = '=',
+	/* Close connection after receiving packet. */
+	DEV_DISCONNECT_IN_AFTER_RECV = '<',
+};
+
+/* Force a close fd after receiving a certain packet type */
+enum dev_disconnect_in dev_disconnect_in(const struct node_id *id, int pkt_type);
 
 /* Make next write on fd fail as if they'd disconnected. */
 void dev_sabotage_fd(int fd, bool close_fd);

connectd/connectd.c

@@ -2017,6 +2017,7 @@ static void peer_downgrade(struct daemon *daemon, const u8 *msg)
 	if (!fromwire_connectd_downgrade_peer(msg, &id))
 		master_badmsg(WIRE_CONNECTD_DOWNGRADE_PEER, msg);
 
+	status_peer_debug(&id, "peer_downgrade");
 	tal_free(important_id_htable_get(daemon->important_ids, &id));
 }
 

connectd/multiplex.c

@@ -415,33 +415,33 @@ static struct io_plan *encrypt_and_send(struct peer *peer,
 {
 	int type = fromwire_peektype(msg);
 
-	switch (dev_disconnect(&peer->id, type)) {
-	case DEV_DISCONNECT_BEFORE:
+	switch (dev_disconnect_out(&peer->id, type)) {
+	case DEV_DISCONNECT_OUT_BEFORE:
 		if (taken(msg))
 			tal_free(msg);
 		return io_close(peer->to_peer);
-	case DEV_DISCONNECT_AFTER:
+	case DEV_DISCONNECT_OUT_AFTER:
 		/* Disallow reads from now on */
 		peer->dev_read_enabled = false;
 		/* Using io_close here can lose the data we're about to send! */
 		next = io_sock_shutdown_cb;
 		break;
-	case DEV_DISCONNECT_BLACKHOLE:
+	case DEV_DISCONNECT_OUT_BLACKHOLE:
 		/* Disable both reads and writes from now on */
 		peer->dev_read_enabled = false;
 		peer->dev_writes_enabled = talz(peer, u32);
 		break;
-	case DEV_DISCONNECT_NORMAL:
+	case DEV_DISCONNECT_OUT_NORMAL:
 		break;
-	case DEV_DISCONNECT_DROP:
+	case DEV_DISCONNECT_OUT_DROP:
 		/* Drop this message and continue */
 		if (taken(msg))
 			tal_free(msg);
 		/* Tell them to read again, */
 		io_wake(&peer->subds);
 		return msg_queue_wait(peer->to_peer, peer->peer_outq,
 				      next, peer);
-	case DEV_DISCONNECT_DISABLE_AFTER:
+	case DEV_DISCONNECT_OUT_DISABLE_AFTER:
 		peer->dev_read_enabled = false;
 		peer->dev_writes_enabled = tal(peer, u32);
 		*peer->dev_writes_enabled = 1;
@@ -1136,6 +1136,13 @@ static struct subd *new_subd(struct peer *peer,
 	return subd;
 }
 
+static struct io_plan *close_peer_dev_disconnect(struct io_conn *peer_conn,
+						 struct peer *peer)
+{
+	assert(peer->to_peer == peer_conn);
+	return io_close_cb(peer_conn, peer);
+}
+
 static struct io_plan *read_hdr_from_peer(struct io_conn *peer_conn,
 					  struct peer *peer);
 static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
@@ -1145,7 +1152,8 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
        struct channel_id channel_id;
        struct subd *subd;
        enum peer_wire type;
-
+       struct io_plan *(*next_read)(struct io_conn *peer_conn,
+				    struct peer *peer) = read_hdr_from_peer;
 
        decrypted = cryptomsg_decrypt_body(tmpctx, &peer->cs,
 					  peer->peer_in);
@@ -1162,16 +1170,24 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
        if (!peer->dev_read_enabled)
 	       return read_hdr_from_peer(peer_conn, peer);
 
+       switch (dev_disconnect_in(&peer->id, type)) {
+       case DEV_DISCONNECT_IN_NORMAL:
+	       break;
+       case DEV_DISCONNECT_IN_AFTER_RECV:
+	       next_read = close_peer_dev_disconnect;
+	       break;
+       }
+
        /* We got something! */
        peer->last_recv_time = time_now();
 
        /* Don't process packets while we're closing */
        if (peer->draining)
-	       return read_hdr_from_peer(peer_conn, peer);
+	       return next_read(peer_conn, peer);
 
        /* If we swallow this, just try again. */
        if (handle_message_locally(peer, decrypted))
-	       return read_hdr_from_peer(peer_conn, peer);
+	       return next_read(peer_conn, peer);
 
        /* After this we should be able to match to subd by channel_id */
        if (!extract_channel_id(decrypted, &channel_id)) {
@@ -1186,15 +1202,15 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
 					"Received %s: %s",
 					peer_wire_name(type), desc);
 		       if (type == WIRE_WARNING)
-			       return read_hdr_from_peer(peer_conn, peer);
+			       return next_read(peer_conn, peer);
 		       return io_close(peer_conn);
 	       }
 
 	       /* This sets final_msg: will close after sending warning */
 	       send_warning(peer, "Unexpected message %s: %s",
 			    peer_wire_name(type),
 			    tal_hex(tmpctx, decrypted));
-	       return read_hdr_from_peer(peer_conn, peer);
+	       return next_read(peer_conn, peer);
        }
 
        /* If we don't find a subdaemon for this, create a new one. */
@@ -1237,7 +1253,7 @@ static struct io_plan *read_body_from_peer_done(struct io_conn *peer_conn,
        }
 
        /* Wait for them to wake us */
-       return io_wait(peer_conn, &peer->peer_in, read_hdr_from_peer, peer);
+       return io_wait(peer_conn, &peer->peer_in, next_read, peer);
 }
 
 static struct io_plan *read_body_from_peer(struct io_conn *peer_conn,

connectd/peer_exchange_initmsg.c

@@ -61,6 +61,13 @@ static struct io_plan *peer_init_received(struct io_conn *conn,
 
 	status_peer_io(LOG_IO_IN, &peer->id, msg);
 
+	switch (dev_disconnect_in(&peer->id, fromwire_peektype(msg))) {
+	case DEV_DISCONNECT_IN_NORMAL:
+		break;
+	case DEV_DISCONNECT_IN_AFTER_RECV:
+		return io_close(conn);
+	}
+
 	/* BOLT #1:
 	 *
 	 * A receiving node:
@@ -280,21 +287,21 @@ struct io_plan *peer_exchange_initmsg(struct io_conn *conn,
 	peer->msg = cryptomsg_encrypt_msg(peer, &peer->cs, take(peer->msg));
 
 	next = read_init;
-	switch (dev_disconnect(&peer->id, WIRE_INIT)) {
-	case DEV_DISCONNECT_BEFORE:
+	switch (dev_disconnect_out(&peer->id, WIRE_INIT)) {
+	case DEV_DISCONNECT_OUT_BEFORE:
 		dev_sabotage_fd(io_conn_fd(conn), true);
 		break;
-	case DEV_DISCONNECT_AFTER:
+	case DEV_DISCONNECT_OUT_AFTER:
 		next = dev_peer_write_postclose;
 		break;
-	case DEV_DISCONNECT_BLACKHOLE:
+	case DEV_DISCONNECT_OUT_BLACKHOLE:
 		status_failed(STATUS_FAIL_INTERNAL_ERROR,
 			      "Blackhole not supported during handshake");
 		break;
-	case DEV_DISCONNECT_NORMAL:
-	case DEV_DISCONNECT_DROP:
+	case DEV_DISCONNECT_OUT_NORMAL:
+	case DEV_DISCONNECT_OUT_DROP:
 		break;
-	case DEV_DISCONNECT_DISABLE_AFTER:
+	case DEV_DISCONNECT_OUT_DISABLE_AFTER:
 		next = dev_peer_write_post_sabotage;
 		break;
 	}

contrib/msggen/msggen/schema.json

@@ -20552,6 +20552,10 @@
                 "value_int": 3,
                 "source": "cmdline"
               },
+              "dev-crash-after": {
+                "value_str": "3600",
+                "source": "cmdline"
+              },
               "dev-fail-on-subdaemon-fail": {
                 "set": true,
                 "source": "cmdline"

devtools/gossipwith.c

@@ -56,6 +56,7 @@ struct io_conn {
 static struct secret notsosecret;
 static bool no_gossip = false, all_gossip = false;
 static unsigned long max_messages = -1UL;
+static u16 *accept_messages = NULL;
 
 /* Empty stubs to make us compile */
 void status_peer_io(enum log_level iodir,
@@ -101,6 +102,18 @@ bool is_unknown_msg_discardable(const u8 *cursor)
 	return false;
 }
 
+static bool accept_message(const u8 *msg)
+{
+	u16 type = fromwire_peektype(msg);
+	if (!accept_messages)
+		return true;
+	for (size_t i = 0; i < tal_count(accept_messages); i++) {
+		if (type == accept_messages[i])
+			return true;
+	}
+	return false;
+}
+
 static struct io_plan *simple_write(struct io_conn *conn,
 				    const void *data, size_t len,
 				    struct io_plan *(*next)(struct io_conn *, void *),
@@ -240,6 +253,10 @@ static struct io_plan *handshake_success(struct io_conn *conn,
 			msg = sync_crypto_read(NULL, peer_fd, cs);
 			if (!msg)
 				err(1, "Reading msg");
+			if (!accept_message(msg)) {
+				tal_free(msg);
+				continue;
+			}
 			if (hex) {
 				printf("%s\n", tal_hex(msg, msg));
 			} else {
@@ -248,8 +265,8 @@ static struct io_plan *handshake_success(struct io_conn *conn,
 				    || !write_all(STDOUT_FILENO, msg, tal_bytelen(msg)))
 					err(1, "Writing out msg");
 			}
-			tal_free(msg);
 			--max_messages;
+			tal_free(msg);
 		}
 	}
 
@@ -284,6 +301,15 @@ static char *opt_set_features(const char *arg, u8 **features)
 	return NULL;
 }
 
+static char *opt_set_filter(const char *arg, u16 **accept)
+{
+	char **elems = tal_strsplit(tmpctx, arg, ",", STR_EMPTY_OK);
+	*accept = tal_arr(NULL, u16, tal_count(elems)-1);
+	for (size_t i = 0; elems[i]; i++)
+		(*accept)[i] = atoi(elems[i]);
+	return NULL;
+}
+
 int main(int argc, char *argv[])
 {
 	struct io_conn *conn = tal(NULL, struct io_conn);
@@ -306,6 +332,8 @@ int main(int argc, char *argv[])
 			   "Stream complete gossip history at start");
 	opt_register_noarg("--no-gossip", opt_set_bool, &no_gossip,
 			   "Suppress all gossip at start");
+	opt_register_arg("--filter", opt_set_filter, NULL, &accept_messages,
+			 "Only process these message types");
 	opt_register_arg("--max-messages", opt_set_ulongval, opt_show_ulongval,
 			 &max_messages,
 			 "Terminate after reading this many messages");

doc/schemas/lightning-listconfigs.json

@@ -2678,6 +2678,10 @@
             "value_int": 3,
             "source": "cmdline"
           },
+          "dev-crash-after": {
+            "value_str": "3600",
+            "source": "cmdline"
+          },
           "dev-fail-on-subdaemon-fail": {
             "set": true,
             "source": "cmdline"

gossipd/gossmap_manage.c

@@ -320,15 +320,15 @@ static void remove_channel(struct gossmap_manage *gm,
 	}
 }
 
+/* If we don't know, we assume it's good */
 static u32 get_timestamp(struct gossmap *gossmap,
 			 const struct gossmap_chan *chan,
 			 int dir)
 {
 	u32 timestamp;
 
-	/* 0 is sufficient for our needs */
 	if (!gossmap_chan_set(chan, dir))
-		return 0;
+		return UINT32_MAX;
 
 	gossmap_chan_get_update_details(gossmap, chan, dir,
 					&timestamp,