gossipd: use file offset within store as broadcast index.

Instead of an arbitrary counter, we can use the file offset for our
partial ordering, removing a field.  It takes some care when we compact
the store, however, as this field changes.

MCP results from 5 runs, min-max(mean +/- stddev):
	store_load_msec:34271-35283(34789.6+/-3.3e+02)
	vsz_kb:2288784
	store_rewrite_sec:38.060000-39.130000(38.426+/-0.39)
	listnodes_sec:0.750000-0.850000(0.794+/-0.042)
	listchannels_sec:30.740000-31.760000(31.096+/-0.35)
	routing_sec:29.600000-33.560000(30.472+/-1.5)
	peer_write_all_sec:49.220000-52.690000(50.892+/-1.3)

MCP notable changes from previous patch (>1 stddev):
	-store_load_msec:35685-38538(37090.4+/-9.1e+02)
	+store_load_msec:34271-35283(34789.6+/-3.3e+02)
	-vsz_kb:2288768
	+vsz_kb:2288784
	-peer_write_all_sec:51.140000-58.350000(55.69+/-2.4)
	+peer_write_all_sec:49.220000-52.690000(50.892+/-1.3)

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

gossipd/broadcast.c

@@ -8,6 +8,7 @@
 #include <common/pseudorand.h>
 #include <common/status.h>
 #include <common/type_to_string.h>
+#include <errno.h>
 #include <gossipd/broadcast.h>
 #include <gossipd/gossip_store.h>
 #include <wire/gen_peer_wire.h>
@@ -24,9 +25,7 @@ struct broadcast_state *new_broadcast_state(struct routing_state *rstate)
 	struct broadcast_state *bstate = tal(rstate, struct broadcast_state);
 	uintmap_init(&bstate->broadcasts);
 	bstate->count = 0;
-	/* Skip 0 because we initialize peers with 0 */
-	bstate->next_index = 1;
-	bstate->gs = gossip_store_new(rstate, bstate);
+	bstate->gs = gossip_store_new(rstate);
 	return bstate;
 }
 
@@ -60,14 +59,51 @@ static struct queued_message *new_queued_message(struct broadcast_state *bstate,
 	return msg;
 }
 
-void insert_broadcast(struct broadcast_state *bstate, const u8 *msg,
+void insert_broadcast_nostore(struct broadcast_state *bstate,
+			      const u8 *msg,
 			      struct broadcastable *bcast)
 {
-	assert(!bcast->index);
-	bcast->index = bstate->next_index++;
 	new_queued_message(bstate, msg, bcast);
 	broadcast_state_check(bstate, "insert_broadcast");
-	gossip_store_add(bstate->gs, msg);
+}
+
+void insert_broadcast(struct broadcast_state **bstate,
+		      const u8 *msg,
+		      struct broadcastable *bcast)
+{
+	/* If we're loading from the store, we already have index */
+	if (!bcast->index) {
+		u64 idx;
+
+		bcast->index = idx = gossip_store_add((*bstate)->gs, msg);
+		if (!idx)
+			status_failed(STATUS_FAIL_INTERNAL_ERROR,
+				      "Could not add to gossip store: %s",
+				      strerror(errno));
+		/* We assume we can fit in 32 bits for now! */
+		assert(idx == bcast->index);
+	}
+
+	insert_broadcast_nostore(*bstate, msg, bcast);
+
+	/* If it compacts, it replaces *bstate */
+	gossip_store_maybe_compact((*bstate)->gs, bstate);
+}
+
+const u8 *pop_first_broadcast(struct broadcast_state *bstate,
+			      struct broadcastable **bcast)
+{
+	u64 idx;
+	const u8 *msg;
+	struct queued_message *q = uintmap_first(&bstate->broadcasts, &idx);
+	if (!q)
+		return NULL;
+
+	*bcast = q->bcast;
+	msg = q->payload;
+
+	broadcast_del(bstate, *bcast);
+	return msg;
 }
 
 const u8 *next_broadcast(struct broadcast_state *bstate,
@@ -87,6 +123,15 @@ const u8 *next_broadcast(struct broadcast_state *bstate,
 	return NULL;
 }
 
+u64 broadcast_final_index(const struct broadcast_state *bstate)
+{
+	u64 idx;
+
+	if (!uintmap_last(&bstate->broadcasts, &idx))
+		return 0;
+	return idx;
+}
+
 #ifdef PEDANTIC
 static const struct pubkey *
 pubkey_keyof(const struct pubkey *pk)

gossipd/broadcast.h

@@ -12,7 +12,6 @@
 struct routing_state;
 
 struct broadcast_state {
-	u64 next_index;
 	UINTMAP(struct queued_message *) broadcasts;
 	size_t count;
 	struct gossip_store *gs;
@@ -37,7 +36,13 @@ struct broadcast_state *new_broadcast_state(struct routing_state *rstate);
 
 /* Append a queued message for broadcast.  Must be explicitly deleted.
  * Also adds it to the gossip store. */
-void insert_broadcast(struct broadcast_state *bstate, const u8 *msg,
+void insert_broadcast(struct broadcast_state **bstate,
+		      const u8 *msg,
+		      struct broadcastable *bcast);
+
+/* Add to broadcast, but not store: for gossip store compaction. */
+void insert_broadcast_nostore(struct broadcast_state *bstate,
+			      const u8 *msg,
 			      struct broadcastable *bcast);
 
 /* Delete a broadcast: not usually needed, since destructor does it */
@@ -51,6 +56,13 @@ const u8 *next_broadcast(struct broadcast_state *bstate,
 			 u32 timestamp_min, u32 timestamp_max,
 			 u32 *last_index);
 
+/* index of last entry. */
+u64 broadcast_final_index(const struct broadcast_state *bstate);
+
+/* Return and remove first element: used by gossip_store_compact */
+const u8 *pop_first_broadcast(struct broadcast_state *bstate,
+			      struct broadcastable **bcast);
+
 /* Returns b if all OK, otherwise aborts if abortstr non-NULL, otherwise returns
  * NULL. */
 struct broadcast_state *broadcast_state_check(struct broadcast_state *b,

gossipd/gossip_store.c

@@ -19,17 +19,17 @@
 #define GOSSIP_STORE_TEMP_FILENAME "gossip_store.tmp"
 
 struct gossip_store {
+	/* This is -1 when we're loading */
 	int fd;
 	u8 version;
 
+	/* Offset of current EOF */
+	u64 len;
+
 	/* Counters for entries in the gossip_store entries. This is used to
 	 * decide whether we should rewrite the on-disk store or not */
 	size_t count;
 
-	/* The broadcast struct we source messages from when rewriting the
-	 * gossip_store */
-	struct broadcast_state *broadcast;
-
 	/* Handle to the routing_state to retrieve additional information,
 	 * should it be needed */
 	struct routing_state *rstate;
@@ -44,15 +44,14 @@ static void gossip_store_destroy(struct gossip_store *gs)
 	close(gs->fd);
 }
 
-struct gossip_store *gossip_store_new(struct routing_state *rstate,
-				      struct broadcast_state *bstate)
+struct gossip_store *gossip_store_new(struct routing_state *rstate)
 {
 	struct gossip_store *gs = tal(rstate, struct gossip_store);
 	gs->count = 0;
 	gs->fd = open(GOSSIP_STORE_FILENAME, O_RDWR|O_APPEND|O_CREAT, 0600);
-	gs->broadcast = bstate;
 	gs->rstate = rstate;
 	gs->disable_compaction = false;
+	gs->len = sizeof(gs->version);
 
 	tal_add_destructor(gs, gossip_store_destroy);
 
@@ -114,10 +113,15 @@ static u8 *gossip_store_wrap_channel_announcement(const tal_t *ctx,
  * Wrap the raw gossip message and write it to fd
  *
  * @param fd File descriptor to write the wrapped message into
+ * @param rstate Routing state if we need to look up channel capacity
  * @param gossip_msg The message to write
+ * @param len The length to increase by amount written.
  * @return true if the message was wrapped and written
  */
-static bool gossip_store_append(int fd, struct routing_state *rstate, const u8 *gossip_msg)
+static bool gossip_store_append(int fd,
+				struct routing_state *rstate,
+				const u8 *gossip_msg,
+				u64 *len)
 {
 	int t =  fromwire_peektype(gossip_msg);
 	u32 msglen;
@@ -144,6 +148,8 @@ static bool gossip_store_append(int fd, struct routing_state *rstate, const u8 *
 	belen = cpu_to_be32(msglen);
 	checksum = cpu_to_be32(crc32c(0, msg, msglen));
 
+	*len += sizeof(belen) + sizeof(checksum) + msglen;
+
 	return (write(fd, &belen, sizeof(belen)) == sizeof(belen) &&
 		write(fd, &checksum, sizeof(checksum)) == sizeof(checksum) &&
 		write(fd, msg, msglen) == msglen);
@@ -157,7 +163,8 @@ static bool gossip_store_append(int fd, struct routing_state *rstate, const u8 *
 */
 static bool add_local_unnannounced(int fd,
 				   struct routing_state *rstate,
-				   struct node *self)
+				   struct node *self,
+				   u64 *len)
 {
 	struct chan_map_iter i;
 	struct chan *c;
@@ -172,15 +179,18 @@ static bool add_local_unnannounced(int fd,
 
 		msg = towire_gossipd_local_add_channel(tmpctx, &c->scid,
 						       &peer->id, c->sat);
-		if (!gossip_store_append(fd, rstate, msg))
+		if (!gossip_store_append(fd, rstate, msg, len))
 			return false;
 
 		for (size_t i = 0; i < 2; i++) {
+			u32 idx;
 			msg = c->half[i].channel_update;
 			if (!msg)
 				continue;
-			if (!gossip_store_append(fd, rstate, msg))
+			idx = *len;
+			if (!gossip_store_append(fd, rstate, msg, len))
 				return false;
+			c->half[i].bcast.index = idx;
 		}
 	}
 
@@ -193,19 +203,27 @@ static bool add_local_unnannounced(int fd,
  * Creates a new file, writes all the updates from the `broadcast_state`, and
  * then atomically swaps the files.
  */
-bool gossip_store_compact(struct gossip_store *gs)
+bool gossip_store_compact(struct gossip_store *gs,
+			  struct broadcast_state **bs)
 {
 	size_t count = 0;
-	u32 index = 0;
 	int fd;
 	const u8 *msg;
 	struct node *self;
+	u64 len = sizeof(gs->version);
+	struct broadcastable *bcast;
+	struct broadcast_state *oldb = *bs;
+	struct broadcast_state *newb;
+
+	if (gs->disable_compaction)
+		return false;
 
-	assert(gs->broadcast);
+	assert(oldb);
 	status_trace(
 	    "Compacting gossip_store with %zu entries, %zu of which are stale",
-	    gs->count, gs->count - gs->broadcast->count);
+	    gs->count, gs->count - oldb->count);
 
+	newb = new_broadcast_state(gs->rstate);
 	fd = open(GOSSIP_STORE_TEMP_FILENAME, O_RDWR|O_APPEND|O_CREAT, 0600);
 
 	if (fd < 0) {
@@ -220,8 +238,10 @@ bool gossip_store_compact(struct gossip_store *gs)
 		goto unlink_disable;
 	}
 
-	while ((msg = next_broadcast(gs->broadcast, 0, UINT32_MAX, &index)) != NULL) {
-		if (!gossip_store_append(fd, gs->rstate, msg)) {
+	while ((msg = pop_first_broadcast(oldb, &bcast)) != NULL) {
+		bcast->index = len;
+		insert_broadcast_nostore(newb, msg, bcast);
+		if (!gossip_store_append(fd, gs->rstate, msg, &len)) {
 			status_broken("Failed writing to gossip store: %s",
 				      strerror(errno));
 			goto unlink_disable;
@@ -231,7 +251,8 @@ bool gossip_store_compact(struct gossip_store *gs)
 
 	/* Local unannounced channels are not in the store! */
 	self = get_node(gs->rstate, &gs->rstate->local_id);
-	if (self && !add_local_unnannounced(fd, gs->rstate, self)) {
+	if (self && !add_local_unnannounced(fd, gs->rstate, self,
+					    &len)) {
 		status_broken("Failed writing unannounced to gossip store: %s",
 			      strerror(errno));
 		goto unlink_disable;
@@ -248,8 +269,12 @@ bool gossip_store_compact(struct gossip_store *gs)
 	    "Compaction completed: dropped %zu messages, new count %zu",
 	    gs->count - count, count);
 	gs->count = count;
+	gs->len = len;
 	close(gs->fd);
 	gs->fd = fd;
+
+	tal_free(oldb);
+	*bs = newb;
 	return true;
 
 unlink_disable:
@@ -258,32 +283,49 @@ disable:
 	status_trace("Encountered an error while compacting, disabling "
 		     "future compactions.");
 	gs->disable_compaction = true;
+	tal_free(newb);
 	return false;
 }
 
-void gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg)
+void gossip_store_maybe_compact(struct gossip_store *gs,
+				struct broadcast_state **bs)
 {
-	/* Only give error message once. */
+	/* Don't compact while loading! */
 	if (gs->fd == -1)
 		return;
+	if (gs->count < 1000)
+		return;
+	if (gs->count < (*bs)->count * 1.25)
+		return;
 
-	if (!gossip_store_append(gs->fd, gs->rstate, gossip_msg)) {
+	gossip_store_compact(gs, bs);
+}
+
+u64 gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg)
+{
+	u64 off = gs->len;
+
+	/* Should never get here during loading! */
+	assert(gs->fd != -1);
+
+	if (!gossip_store_append(gs->fd, gs->rstate, gossip_msg, &gs->len)) {
 		status_broken("Failed writing to gossip store: %s",
 			      strerror(errno));
-		gs->fd = -1;
+		return 0;
 	}
 
 	gs->count++;
-	if (gs->count >= 1000 && gs->count > gs->broadcast->count * 1.25 &&
-	    !gs->disable_compaction)
-		gossip_store_compact(gs);
+	return off;
 }
 
 void gossip_store_add_channel_delete(struct gossip_store *gs,
 				     const struct short_channel_id *scid)
 {
 	u8 *msg = towire_gossip_store_channel_delete(NULL, scid);
-	gossip_store_append(gs->fd, gs->rstate, msg);
+	if (!gossip_store_append(gs->fd, gs->rstate, msg, &gs->len))
+		status_failed(STATUS_FAIL_INTERNAL_ERROR,
+			      "Failed writing channel_delete to gossip store: %s",
+			      strerror(errno));
 	tal_free(msg);
 }
 
@@ -294,15 +336,13 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs)
 	u8 *msg, *gossip_msg;
 	struct amount_sat satoshis;
 	struct short_channel_id scid;
-	/* We set/check version byte on creation */
-	off_t known_good = 1;
 	const char *bad;
 	size_t stats[] = {0, 0, 0, 0};
 	int fd = gs->fd;
 	gs->fd = -1;
 	struct timeabs start = time_now();
 
-	if (lseek(fd, known_good, SEEK_SET) < 0) {
+	if (lseek(fd, gs->len, SEEK_SET) < 0) {
 		status_unusual("gossip_store: lseek failure");
 		goto truncate_nomsg;
 	}
@@ -327,7 +367,8 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs)
 							       &satoshis)) {
 			if (!routing_add_channel_announcement(rstate,
 							      take(gossip_msg),
-							      satoshis)) {
+							      satoshis,
+							      gs->len)) {
 				bad = "Bad channel_announcement";
 				goto truncate;
 			}
@@ -335,7 +376,8 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs)
 		} else if (fromwire_gossip_store_channel_update(msg, msg,
 								&gossip_msg)) {
 			if (!routing_add_channel_update(rstate,
-							take(gossip_msg))) {
+							take(gossip_msg),
+							gs->len)) {
 				bad = "Bad channel_update";
 				goto truncate;
 			}
@@ -343,7 +385,8 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs)
 		} else if (fromwire_gossip_store_node_announcement(msg, msg,
 								   &gossip_msg)) {
 			if (!routing_add_node_announcement(rstate,
-							   take(gossip_msg))) {
+							   take(gossip_msg),
+							   gs->len)) {
 				bad = "Bad node_announcement";
 				goto truncate;
 			}
@@ -363,7 +406,7 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs)
 			bad = "Unknown message";
 			goto truncate;
 		}
-		known_good += sizeof(belen) + sizeof(becsum) + msglen;
+		gs->len += sizeof(belen) + sizeof(becsum) + msglen;
 		gs->count++;
 		tal_free(msg);
 	}
@@ -385,13 +428,13 @@ out:
 	status_info("total store load time: %"PRIu64" msec (%zu entries, %zu bytes)",
 		    time_to_msec(time_between(time_now(), start)),
 		    stats[0] + stats[1] + stats[2] + stats[3],
-		    (size_t)known_good);
+		    (size_t)gs->len);
 #else
 	status_trace("total store load time: %"PRIu64" msec",
 		     time_to_msec(time_between(time_now(), start)));
 #endif
 	status_trace("gossip_store: Read %zu/%zu/%zu/%zu cannounce/cupdate/nannounce/cdelete from store in %"PRIu64" bytes",
 		     stats[0], stats[1], stats[2], stats[3],
-		     (u64)known_good);
+		     gs->len);
 	gs->fd = fd;
 }

gossipd/gossip_store.h

@@ -17,8 +17,7 @@ struct broadcast_state;
 struct gossip_store;
 struct routing_state;
 
-struct gossip_store *gossip_store_new(struct routing_state *rstate,
-				      struct broadcast_state *bstate);
+struct gossip_store *gossip_store_new(struct routing_state *rstate);
 
 /**
  * Load the initial gossip store, if any.
@@ -31,7 +30,7 @@ void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs);
 /**
  * Add a gossip message to the gossip_store
  */
-void gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg);
+u64 gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg);
 
 /**
  * Remember that we deleted a channel as a result of its outpoint being spent
@@ -39,6 +38,16 @@ void gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg);
 void gossip_store_add_channel_delete(struct gossip_store *gs,
 				     const struct short_channel_id *scid);
 
-/* Expose for dev-compact-gossip-store */
-bool gossip_store_compact(struct gossip_store *gs);
+/**
+ * If we need to compact the gossip store, do so.
+ * @gs: the gossip store.
+ * @bs: a pointer to the broadcast state: replaced if we compact it.
+ */
+void gossip_store_maybe_compact(struct gossip_store *gs,
+				struct broadcast_state **bs);
+
+
+/* Expose for dev-compact-gossip-store to force compaction. */
+bool gossip_store_compact(struct gossip_store *gs,
+			  struct broadcast_state **bs);
 #endif /* LIGHTNING_GOSSIPD_GOSSIP_STORE_H */

gossipd/gossipd.c

@@ -1669,7 +1669,7 @@ static struct io_plan *connectd_new_peer(struct io_conn *conn,
 			peer->broadcast_index = 0;
 		else
 			peer->broadcast_index
-				= peer->daemon->rstate->broadcasts->next_index;
+				= broadcast_final_index(peer->daemon->rstate->broadcasts) + 1;
 	}
 
 	/* This is the new connection: calls dump_gossip when nothing else to
@@ -2480,7 +2480,8 @@ static struct io_plan *dev_compact_store(struct io_conn *conn,
 					 struct daemon *daemon,
 					 const u8 *msg)
 {
-	bool done = gossip_store_compact(daemon->rstate->broadcasts->gs);
+	bool done = gossip_store_compact(daemon->rstate->broadcasts->gs,
+					 &daemon->rstate->broadcasts);
 	daemon_conn_send(daemon->master,
 			 take(towire_gossip_dev_compact_store_reply(NULL,
 								    done)));

gossipd/routing.c

@@ -323,7 +323,7 @@ static void remove_chan_from_node(struct routing_state *rstate,
 		 * channel_announce, but we don't care that much about spurious
 		 * retransmissions in this corner case */
 		broadcast_del(rstate->broadcasts, &node->bcast);
-		insert_broadcast(rstate->broadcasts,
+		insert_broadcast(&rstate->broadcasts,
 				 node->node_announcement,
 				 &node->bcast);
 	}
@@ -856,7 +856,7 @@ static void add_channel_announce_to_broadcast(struct routing_state *rstate,
 					      u32 timestamp)
 {
 	chan->bcast.timestamp = timestamp;
-	insert_broadcast(rstate->broadcasts, chan->channel_announce,
+	insert_broadcast(&rstate->broadcasts, chan->channel_announce,
 			 &chan->bcast);
 	rstate->local_channel_announced |= is_local_channel(rstate, chan);
 
@@ -866,7 +866,7 @@ static void add_channel_announce_to_broadcast(struct routing_state *rstate,
 		if (!node->node_announcement)
 			continue;
 		if (!node->bcast.index) {
-			insert_broadcast(rstate->broadcasts,
+			insert_broadcast(&rstate->broadcasts,
 					 node->node_announcement,
 					 &node->bcast);
 		}
@@ -875,7 +875,8 @@ static void add_channel_announce_to_broadcast(struct routing_state *rstate,
 
 bool routing_add_channel_announcement(struct routing_state *rstate,
 				      const u8 *msg TAKES,
-				      struct amount_sat sat)
+				      struct amount_sat sat,
+				      u32 index)
 {
 	struct chan *chan;
 	secp256k1_ecdsa_signature node_signature_1, node_signature_2;
@@ -908,6 +909,11 @@ bool routing_add_channel_announcement(struct routing_state *rstate,
 	/* Channel is now public. */
 	chan->channel_announce = tal_dup_arr(chan, u8, msg, tal_count(msg), 0);
 
+	/* If we're loading from the store, save index now */
+	chan->bcast.index = index;
+	/* This is filled in when we get a channel_update */
+	chan->bcast.timestamp = 0;
+
 	/* Apply any private updates. */
 	for (size_t i = 0; i < ARRAY_SIZE(chan->half); i++) {
 		const u8 *update = chan->half[i].channel_update;
@@ -916,7 +922,9 @@ bool routing_add_channel_announcement(struct routing_state *rstate,
 
 		/* Remove from channel, otherwise it will be freed! */
 		chan->half[i].channel_update = NULL;
-		routing_add_channel_update(rstate, take(update));
+		/* If we loaded from store, index will be non-zero */
+		routing_add_channel_update(rstate, take(update),
+					   chan->half[i].bcast.index);
 	}
 
 	return true;
@@ -1156,7 +1164,7 @@ void handle_pending_cannouncement(struct routing_state *rstate,
 		return;
 	}
 
-	if (!routing_add_channel_announcement(rstate, pending->announce, sat))
+	if (!routing_add_channel_announcement(rstate, pending->announce, sat, 0))
 		status_failed(STATUS_FAIL_INTERNAL_ERROR,
 			      "Could not add channel_announcement");
 
@@ -1217,7 +1225,9 @@ static void set_connection_values(struct chan *chan,
 }
 
 bool routing_add_channel_update(struct routing_state *rstate,
-				const u8 *update TAKES)
+				const u8 *update TAKES,
+				u32 index)
+
 {
 	secp256k1_ecdsa_signature signature;
 	struct short_channel_id short_channel_id;
@@ -1291,12 +1301,18 @@ bool routing_add_channel_update(struct routing_state *rstate,
 	chan->half[direction].channel_update
 		= tal_dup_arr(chan, u8, update, tal_count(update), 0);
 
+	/* If we're loading from store, this means we don't re-add to store */
+	chan->half[direction].bcast.index = index;
+
 	/* For private channels, we get updates without an announce: don't
 	 * broadcast them!  But save local ones to store anyway. */
 	if (!chan->channel_announce) {
-		if (is_local_channel(rstate, chan))
-			gossip_store_add(rstate->broadcasts->gs,
-					 chan->half[direction].channel_update);
+		struct half_chan *hc = &chan->half[direction];
+		/* Don't save if we're loading from store */
+		if (is_local_channel(rstate, chan) && !hc->bcast.index) {
+			hc->bcast.index = gossip_store_add(rstate->broadcasts->gs,
+							   hc->channel_update);
+		}
 		return true;
 	}
 
@@ -1306,10 +1322,10 @@ bool routing_add_channel_update(struct routing_state *rstate,
 	 *   - MUST consider whether to send the `channel_announcement` after
 	 *     receiving the first corresponding `channel_update`.
 	 */
-	if (chan->bcast.index == 0)
+	if (chan->bcast.timestamp == 0)
 		add_channel_announce_to_broadcast(rstate, chan, timestamp);
 
-	insert_broadcast(rstate->broadcasts,
+	insert_broadcast(&rstate->broadcasts,
 			 chan->half[direction].channel_update,
 			 &chan->half[direction].bcast);
 	return true;
@@ -1461,7 +1477,7 @@ u8 *handle_channel_update(struct routing_state *rstate, const u8 *update TAKES,
 		     : "UNDEFINED",
 		     source);
 
-	if (!routing_add_channel_update(rstate, serialized))
+	if (!routing_add_channel_update(rstate, serialized, 0))
 		status_failed(STATUS_FAIL_INTERNAL_ERROR,
 			      "Failed adding channel_update");
 
@@ -1499,7 +1515,9 @@ static struct wireaddr *read_addresses(const tal_t *ctx, const u8 *ser)
 	return wireaddrs;
 }
 
-bool routing_add_node_announcement(struct routing_state *rstate, const u8 *msg TAKES)
+bool routing_add_node_announcement(struct routing_state *rstate,
+				   const u8 *msg TAKES,
+				   u32 index)
 {
 	struct node *node;
 	secp256k1_ecdsa_signature signature;
@@ -1537,6 +1555,7 @@ bool routing_add_node_announcement(struct routing_state *rstate, const u8 *msg T
 	node->addresses = tal_steal(node, wireaddrs);
 
 	node->bcast.timestamp = timestamp;
+	node->bcast.index = index;
 	memcpy(node->rgb_color, rgb_color, ARRAY_SIZE(node->rgb_color));
 	memcpy(node->alias, alias, ARRAY_SIZE(node->alias));
 	tal_free(node->globalfeatures);
@@ -1546,7 +1565,7 @@ bool routing_add_node_announcement(struct routing_state *rstate, const u8 *msg T
 
 	/* We might be waiting for channel_announce to be released. */
 	if (node_has_broadcastable_channels(node)) {
-		insert_broadcast(rstate->broadcasts,
+		insert_broadcast(&rstate->broadcasts,
 				 node->node_announcement,
 				 &node->bcast);
 	}
@@ -1684,7 +1703,7 @@ u8 *handle_node_announcement(struct routing_state *rstate, const u8 *node_ann)
 	status_trace("Received node_announcement for node %s",
 		     type_to_string(tmpctx, struct node_id, &node_id));
 
-	applied = routing_add_node_announcement(rstate, serialized);
+	applied = routing_add_node_announcement(rstate, serialized, 0);
 	assert(applied);
 	return NULL;
 }

gossipd/routing.h

@@ -326,10 +326,14 @@ void route_prune(struct routing_state *rstate);
  * Directly add the channel to the local network, without checking it first. Use
  * this only for messages from trusted sources. Untrusted sources should use the
  * @see{handle_channel_announcement} entrypoint to check before adding.
+ *
+ * index is usually 0, in which case it's set by insert_broadcast adding it
+ * to the store.
  */
 bool routing_add_channel_announcement(struct routing_state *rstate,
 				      const u8 *msg TAKES,
-				      struct amount_sat sat);
+				      struct amount_sat sat,
+				      u32 index);
 
 /**
  * Add a channel_update without checking for errors
@@ -340,7 +344,8 @@ bool routing_add_channel_announcement(struct routing_state *rstate,
  * @see{handle_channel_update}
  */
 bool routing_add_channel_update(struct routing_state *rstate,
-				const u8 *update TAKES);
+				const u8 *update TAKES,
+				u32 index);
 
 /**
  * Add a node_announcement to the network view without checking it
@@ -350,7 +355,8 @@ bool routing_add_channel_update(struct routing_state *rstate,
  * sources (peers) please use @see{handle_node_announcement}.
  */
 bool routing_add_node_announcement(struct routing_state *rstate,
-                                  const u8 *msg TAKES);
+				   const u8 *msg TAKES,
+				   u32 index);
 
 
 /**