gossipd: embed broadcast information into each structure.

This is more compact, but also required once we replace the arbitrary
"index" with an actual offset into the gossip store.  That will let us
remove the in-memory variants entirely.

MCP results from 5 runs, min-max(mean +/- stddev):
	store_load_msec:35685-38538(37090.4+/-9.1e+02)
	vsz_kb:2288768
	store_rewrite_sec:35.530000-41.230000(37.904+/-2.3)
	listnodes_sec:0.720000-0.810000(0.762+/-0.041)
	listchannels_sec:30.750000-35.990000(32.704+/-2)
	routing_sec:29.570000-34.010000(31.374+/-1.8)
	peer_write_all_sec:51.140000-58.350000(55.69+/-2.4)

MCP notable changes from previous patch (>1 stddev):
	-vsz_kb:2621808
	+vsz_kb:2288768

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

gossipd/broadcast.c

@@ -9,83 +9,80 @@
 #include <common/status.h>
 #include <common/type_to_string.h>
 #include <gossipd/broadcast.h>
+#include <gossipd/gossip_store.h>
 #include <wire/gen_peer_wire.h>
 
 struct queued_message {
-	/* Broadcast index. */
-	u64 index;
-
-	/* Timestamp, for filtering. */
-	u32 timestamp;
+	struct broadcastable *bcast;
 
 	/* Serialized payload */
 	const u8 *payload;
 };
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx)
+struct broadcast_state *new_broadcast_state(struct routing_state *rstate)
 {
-	struct broadcast_state *bstate = tal(ctx, struct broadcast_state);
+	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->count = 0;
+	bstate->gs = gossip_store_new(rstate, bstate);
 	return bstate;
 }
 
-void broadcast_del(struct broadcast_state *bstate, u64 index, const u8 *payload)
+void broadcast_del(struct broadcast_state *bstate,
+		   struct broadcastable *bcast)
 {
-	const struct queued_message *q = uintmap_del(&bstate->broadcasts, index);
+	const struct queued_message *q
+		= uintmap_del(&bstate->broadcasts, bcast->index);
 	if (q != NULL) {
-		assert(q->payload == payload);
+		assert(q->bcast == bcast);
 		tal_free(q);
+		bstate->count--;
 		broadcast_state_check(bstate, "broadcast_del");
+		bcast->index = 0;
 	}
 }
 
-static void destroy_queued_message(struct queued_message *msg,
-				   struct broadcast_state *bstate)
-{
-	broadcast_del(bstate, msg->index, msg->payload);
-	bstate->count--;
-}
-
-static struct queued_message *new_queued_message(const tal_t *ctx,
-						 struct broadcast_state *bstate,
+static struct queued_message *new_queued_message(struct broadcast_state *bstate,
 						 const u8 *payload,
-						 u32 timestamp,
-						 u64 index)
+						 struct broadcastable *bcast)
 {
-	struct queued_message *msg = tal(ctx, struct queued_message);
+	struct queued_message *msg = tal(bstate, struct queued_message);
 	assert(payload);
+	assert(bcast);
+	assert(bcast->index);
 	msg->payload = payload;
-	msg->index = index;
-	msg->timestamp = timestamp;
-	uintmap_add(&bstate->broadcasts, index, msg);
-	tal_add_destructor2(msg, destroy_queued_message, bstate);
+	msg->bcast = bcast;
+	if (!uintmap_add(&bstate->broadcasts, bcast->index, msg))
+		abort();
 	bstate->count++;
 	return msg;
 }
 
-u64 insert_broadcast(struct broadcast_state *bstate,
-		      const u8 *payload, u32 timestamp)
+void insert_broadcast(struct broadcast_state *bstate, const u8 *msg,
+		      struct broadcastable *bcast)
 {
-	/* Free payload, free index. */
-	new_queued_message(payload, bstate, payload, timestamp,
-			   bstate->next_index);
+	assert(!bcast->index);
+	bcast->index = bstate->next_index++;
+	new_queued_message(bstate, msg, bcast);
 	broadcast_state_check(bstate, "insert_broadcast");
-	return bstate->next_index++;
+	gossip_store_add(bstate->gs, msg);
 }
 
 const u8 *next_broadcast(struct broadcast_state *bstate,
 			 u32 timestamp_min, u32 timestamp_max,
-			 u64 *last_index)
+			 u32 *last_index)
 {
 	struct queued_message *m;
+	u64 idx = *last_index;
 
-	while ((m = uintmap_after(&bstate->broadcasts, last_index)) != NULL) {
-		if (m->timestamp >= timestamp_min
-		    && m->timestamp <= timestamp_max)
+	while ((m = uintmap_after(&bstate->broadcasts, &idx)) != NULL) {
+		if (m->bcast->timestamp >= timestamp_min
+		    && m->bcast->timestamp <= timestamp_max) {
+			*last_index = idx;
 			return m->payload;
+		}
 	}
 	return NULL;
 }
@@ -134,7 +131,8 @@ struct broadcast_state *broadcast_state_check(struct broadcast_state *b,
 	struct pubkey node_id_1,  node_id_2, bitcoin_key;
 	u32 timestamp, fees;
 	u16 flags, expiry;
-	u64 index = 0, htlc_minimum_msat;
+	u32 index = 0;
+	u64 htlc_minimum_msat;
 	struct pubkey_set pubkeys;
 	/* We actually only need a set, not a map. */
 	UINTMAP(u64 *) channels;

gossipd/broadcast.h

@@ -9,27 +9,47 @@
 
 /* Common functionality to implement staggered broadcasts with replacement. */
 
+struct routing_state;
+
 struct broadcast_state {
 	u64 next_index;
 	UINTMAP(struct queued_message *) broadcasts;
 	size_t count;
+	struct gossip_store *gs;
+};
+
+/* This is nested inside a node, chan or half_chan; rewriting the store can
+ * cause it to change! */
+struct broadcastable {
+	/* This is also the offset within the gossip_store; even with 1M
+	 * channels we still have a factor of 8 before this wraps. */
+	u32 index;
+	u32 timestamp;
 };
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx);
+static inline void broadcastable_init(struct broadcastable *bcast)
+{
+	bcast->index = 0;
+}
+
+struct broadcast_state *new_broadcast_state(struct routing_state *rstate);
+
 
-/* Append a queued message for broadcast.  Freeing the msg will remove it. */
-u64 insert_broadcast(struct broadcast_state *bstate, const u8 *msg,
-		     u32 timestamp);
+/* 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,
+		      struct broadcastable *bcast);
 
-/* Manually delete a broadcast: not usually needed, since destructor does it */
-void broadcast_del(struct broadcast_state *bstate, u64 index, const u8 *payload);
+/* Delete a broadcast: not usually needed, since destructor does it */
+void broadcast_del(struct broadcast_state *bstate,
+		   struct broadcastable *bcast);
 
 /* Return the broadcast with index >= *last_index, timestamp >= min and <= max
  * and update *last_index.
  * There's no broadcast with index 0. */
 const u8 *next_broadcast(struct broadcast_state *bstate,
 			 u32 timestamp_min, u32 timestamp_max,
-			 u64 *last_index);
+			 u32 *last_index);
 
 /* Returns b if all OK, otherwise aborts if abortstr non-NULL, otherwise returns
  * NULL. */

gossipd/gossip_store.c

@@ -44,14 +44,13 @@ static void gossip_store_destroy(struct gossip_store *gs)
 	close(gs->fd);
 }
 
-struct gossip_store *gossip_store_new(const tal_t *ctx,
-				      struct routing_state *rstate,
-				      struct broadcast_state *broadcast)
+struct gossip_store *gossip_store_new(struct routing_state *rstate,
+				      struct broadcast_state *bstate)
 {
-	struct gossip_store *gs = tal(ctx, struct gossip_store);
+	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 = broadcast;
+	gs->broadcast = bstate;
 	gs->rstate = rstate;
 	gs->disable_compaction = false;
 
@@ -197,7 +196,7 @@ static bool add_local_unnannounced(int fd,
 bool gossip_store_compact(struct gossip_store *gs)
 {
 	size_t count = 0;
-	u64 index = 0;
+	u32 index = 0;
 	int fd;
 	const u8 *msg;
 	struct node *self;

gossipd/gossip_store.h

@@ -13,12 +13,12 @@
  */
 #define GOSSIP_STORE_VERSION 3
 
+struct broadcast_state;
 struct gossip_store;
 struct routing_state;
 
-struct gossip_store *gossip_store_new(const tal_t *ctx,
-				      struct routing_state *rstate,
-				      struct broadcast_state *broadcast);
+struct gossip_store *gossip_store_new(struct routing_state *rstate,
+				      struct broadcast_state *bstate);
 
 /**
  * Load the initial gossip store, if any.

gossipd/gossipd.c

@@ -133,7 +133,7 @@ struct peer {
 	bool gossip_queries_feature, initial_routing_sync_feature;
 
 	/* High water mark for the staggered broadcast */
-	u64 broadcast_index;
+	u32 broadcast_index;
 
 	/* Timestamp range the peer asked us to filter gossip by */
 	u32 gossip_timestamp_min, gossip_timestamp_max;
@@ -396,8 +396,8 @@ static void send_node_announcement(struct daemon *daemon)
 	 *   - MUST set `timestamp` to be greater than that of any previous
 	 *   `node_announcement` it has previously created.
 	 */
-	if (self && self->node_announcement && timestamp <= self->last_timestamp)
-		timestamp = self->last_timestamp + 1;
+	if (self && self->bcast.index && timestamp <= self->bcast.timestamp)
+		timestamp = self->bcast.timestamp + 1;
 
 	/* Get an unsigned one. */
 	nannounce = create_node_announcement(tmpctx, daemon, NULL, timestamp);
@@ -1074,7 +1074,7 @@ static void maybe_create_next_scid_reply(struct peer *peer)
 		/* Not every node announces itself (we know it exists because
 		 * of a channel_announcement, however) */
 		n = get_node(rstate, &peer->scid_query_nodes[i]);
-		if (!n || !n->node_announcement_index)
+		if (!n || !n->bcast.index)
 			continue;
 
 		queue_peer_msg(peer, n->node_announcement);
@@ -1199,7 +1199,7 @@ static void update_local_channel(struct daemon *daemon,
 	 *     - SHOULD base `timestamp` on a UNIX timestamp.
 	 */
 	if (is_halfchan_defined(&chan->half[direction])
-	    && timestamp == chan->half[direction].last_timestamp)
+	    && timestamp == chan->half[direction].bcast.timestamp)
 		timestamp++;
 
 	/* BOLT #7:
@@ -1562,7 +1562,8 @@ static struct io_plan *peer_msg_in(struct io_conn *conn,
 	case WIRE_GOSSIPD_LOCAL_ADD_CHANNEL:
 		ok = handle_local_add_channel(peer->daemon->rstate, msg);
 		if (ok)
-			gossip_store_add(peer->daemon->rstate->store, msg);
+			gossip_store_add(peer->daemon->rstate->broadcasts->gs,
+					 msg);
 		goto handled_cmd;
 	case WIRE_GOSSIPD_LOCAL_CHANNEL_UPDATE:
 		ok = handle_local_channel_update(peer, msg);
@@ -1646,7 +1647,7 @@ static struct io_plan *connectd_new_peer(struct io_conn *conn,
 	 *	- MUST NOT relay any gossip messages unless explicitly requested.
 	 */
 	if (peer->gossip_queries_feature) {
-		peer->broadcast_index = UINT64_MAX;
+		peer->broadcast_index = UINT32_MAX;
 		/* Nothing in this "impossible" range */
 		peer->gossip_timestamp_min = UINT32_MAX;
 		peer->gossip_timestamp_max = 0;
@@ -1806,7 +1807,7 @@ static void gossip_refresh_network(struct daemon *daemon)
 				continue;
 			}
 
-			if (hc->last_timestamp > highwater) {
+			if (hc->bcast.timestamp > highwater) {
 				/* No need to send a keepalive update message */
 				continue;
 			}
@@ -1876,7 +1877,7 @@ static struct io_plan *gossip_init(struct io_conn *conn,
 					   dev_unknown_channel_satoshis);
 
 	/* Load stored gossip messages */
-	gossip_store_load(daemon->rstate, daemon->rstate->store);
+	gossip_store_load(daemon->rstate, daemon->rstate->broadcasts->gs);
 
 	/* Now disable all local channels, they can't be connected yet. */
 	gossip_disable_local_channels(daemon);
@@ -1961,7 +1962,7 @@ static struct gossip_halfchannel_entry *hc_entry(const tal_t *ctx,
 	e = tal(ctx, struct gossip_halfchannel_entry);
 	e->channel_flags = c->channel_flags;
 	e->message_flags = c->message_flags;
-	e->last_update_timestamp = c->last_timestamp;
+	e->last_update_timestamp = c->bcast.timestamp;
 	e->base_fee_msat = c->base_fee;
 	e->fee_per_millionth = c->proportional_fee;
 	e->delay = c->delay;
@@ -2060,7 +2061,7 @@ static void append_node(const struct gossip_getnodes_entry ***entries,
 	if (!n->node_announcement)
 		e->last_timestamp = -1;
 	else {
-		e->last_timestamp = n->last_timestamp;
+		e->last_timestamp = n->bcast.timestamp;
 		e->globalfeatures = n->globalfeatures;
 		e->addresses = n->addresses;
 		BUILD_ASSERT(ARRAY_SIZE(e->alias) == ARRAY_SIZE(n->alias));
@@ -2479,7 +2480,7 @@ static struct io_plan *dev_compact_store(struct io_conn *conn,
 					 struct daemon *daemon,
 					 const u8 *msg)
 {
-	bool done = gossip_store_compact(daemon->rstate->store);
+	bool done = gossip_store_compact(daemon->rstate->broadcasts->gs);
 	daemon_conn_send(daemon->master,
 			 take(towire_gossip_dev_compact_store_reply(NULL,
 								    done)));
@@ -2656,7 +2657,7 @@ static struct io_plan *handle_outpoint_spent(struct io_conn *conn,
 		tal_free(chan);
 		/* We put a tombstone marker in the channel store, so we don't
 		 * have to replay blockchain spends on restart. */
-		gossip_store_add_channel_delete(rstate->store, &scid);
+		gossip_store_add_channel_delete(rstate->broadcasts->gs, &scid);
 	}
 
 	return daemon_conn_read_next(conn, daemon->master);

gossipd/routing.c

@@ -158,7 +158,6 @@ struct routing_state *new_routing_state(const tal_t *ctx,
 	rstate->chainparams = chainparams;
 	rstate->local_id = *local_id;
 	rstate->prune_timeout = prune_timeout;
-	rstate->store = gossip_store_new(rstate, rstate, rstate->broadcasts);
 	rstate->local_channel_announced = false;
 	list_head_init(&rstate->pending_cannouncement);
 	uintmap_init(&rstate->chanmap);
@@ -203,6 +202,9 @@ static void destroy_node(struct node *node, struct routing_state *rstate)
 	struct chan *c;
 	node_map_del(rstate->nodes, node);
 
+	/* Safe even if never placed in broadcast map */
+	broadcast_del(rstate->broadcasts, &node->bcast);
+
 	/* These remove themselves from chans[]. */
 	while ((c = first_chan(node, &i)) != NULL)
 		tal_free(c);
@@ -230,7 +232,7 @@ static struct node *new_node(struct routing_state *rstate,
 	memset(n->chans.arr, 0, sizeof(n->chans.arr));
 	n->globalfeatures = NULL;
 	n->node_announcement = NULL;
-	n->node_announcement_index = 0;
+	broadcastable_init(&n->bcast);
 	n->addresses = tal_arr(n, struct wireaddr, 0);
 	node_map_add(rstate->nodes, n);
 	tal_add_destructor2(n, destroy_node, rstate);
@@ -277,21 +279,12 @@ static bool node_announce_predates_channels(const struct node *node)
 		if (!is_chan_announced(c))
 			continue;
 
-		if (c->channel_announcement_index
-		    < node->node_announcement_index)
+		if (c->bcast.index < node->bcast.index)
 			return false;
 	}
 	return true;
 }
 
-static u64 persistent_broadcast(struct routing_state *rstate, const u8 *msg, u32 timestamp)
-{
-	u64 index = insert_broadcast(rstate->broadcasts, msg, timestamp);
-	if (index)
-		gossip_store_add(rstate->store, msg);
-	return index;
-}
-
 static void remove_chan_from_node(struct routing_state *rstate,
 				  struct node *node, const struct chan *chan)
 {
@@ -318,23 +311,21 @@ static void remove_chan_from_node(struct routing_state *rstate,
 		return;
 	}
 
-	if (!node->node_announcement_index)
+	if (!node->bcast.index)
 		return;
 
 	/* Removed only public channel?  Remove node announcement. */
 	if (!node_has_broadcastable_channels(node)) {
-		broadcast_del(rstate->broadcasts, node->node_announcement_index,
-			      node->node_announcement);
-		node->node_announcement_index = 0;
+		broadcast_del(rstate->broadcasts, &node->bcast);
 	} else if (node_announce_predates_channels(node)) {
 		/* node announcement predates all channel announcements?
 		 * Move to end (we could, in theory, move to just past next
 		 * channel_announce, but we don't care that much about spurious
 		 * retransmissions in this corner case */
-		broadcast_del(rstate->broadcasts, node->node_announcement_index,
-			      node->node_announcement);
-		node->node_announcement_index = persistent_broadcast(
-		    rstate, node->node_announcement, node->last_timestamp);
+		broadcast_del(rstate->broadcasts, &node->bcast);
+		insert_broadcast(rstate->broadcasts,
+				 node->node_announcement,
+				 &node->bcast);
 	}
 }
 
@@ -343,6 +334,11 @@ static void destroy_chan(struct chan *chan, struct routing_state *rstate)
 	remove_chan_from_node(rstate, chan->nodes[0], chan);
 	remove_chan_from_node(rstate, chan->nodes[1], chan);
 
+	/* Safe even if never placed in map */
+	broadcast_del(rstate->broadcasts, &chan->bcast);
+	broadcast_del(rstate->broadcasts, &chan->half[0].bcast);
+	broadcast_del(rstate->broadcasts, &chan->half[1].bcast);
+
 	uintmap_del(&rstate->chanmap, chan->scid.u64);
 }
 
@@ -358,9 +354,11 @@ static void init_half_chan(struct routing_state *rstate,
 	c->channel_flags = channel_idx;
 	// TODO: wireup message_flags
 	c->message_flags = 0;
+	broadcastable_init(&c->bcast);
+
 	/* We haven't seen channel_update: make it halfway to prune time,
 	 * which should be older than any update we'd see. */
-	c->last_timestamp = gossip_time_now(rstate).ts.tv_sec - rstate->prune_timeout/2;
+	c->bcast.timestamp = gossip_time_now(rstate).ts.tv_sec - rstate->prune_timeout/2;
 }
 
 static void bad_gossip_order(const u8 *msg, const char *source,
@@ -397,7 +395,7 @@ struct chan *new_chan(struct routing_state *rstate,
 	chan->nodes[!n1idx] = n2;
 	chan->txout_script = NULL;
 	chan->channel_announce = NULL;
-	chan->channel_announcement_index = 0;
+	broadcastable_init(&chan->bcast);
 	chan->sat = satoshis;
 	chan->local_disabled = false;
 
@@ -857,8 +855,9 @@ static void add_channel_announce_to_broadcast(struct routing_state *rstate,
 					      struct chan *chan,
 					      u32 timestamp)
 {
-	chan->channel_announcement_index =
-	    persistent_broadcast(rstate, chan->channel_announce, timestamp);
+	chan->bcast.timestamp = timestamp;
+	insert_broadcast(rstate->broadcasts, chan->channel_announce,
+			 &chan->bcast);
 	rstate->local_channel_announced |= is_local_channel(rstate, chan);
 
 	/* If we've been waiting for this, now we can announce node */
@@ -866,10 +865,10 @@ static void add_channel_announce_to_broadcast(struct routing_state *rstate,
 		struct node *node = chan->nodes[i];
 		if (!node->node_announcement)
 			continue;
-		if (!node->node_announcement_index) {
-			node->node_announcement_index = persistent_broadcast(
-			    rstate, node->node_announcement,
-			    node->last_timestamp);
+		if (!node->bcast.index) {
+			insert_broadcast(rstate->broadcasts,
+					 node->node_announcement,
+					 &node->bcast);
 		}
 	}
 }
@@ -1209,7 +1208,7 @@ static void set_connection_values(struct chan *chan,
 	c->proportional_fee = proportional_fee;
 	c->message_flags = message_flags;
 	c->channel_flags = channel_flags;
-	c->last_timestamp = timestamp;
+	c->bcast.timestamp = timestamp;
 	assert((c->channel_flags & ROUTING_FLAGS_DIRECTION) == idx);
 
 	SUPERVERBOSE("Channel %s/%d was updated.",
@@ -1286,6 +1285,9 @@ bool routing_add_channel_update(struct routing_state *rstate,
 
 	/* Replace any old one. */
 	tal_free(chan->half[direction].channel_update);
+	/* Safe even if was never added */
+	broadcast_del(rstate->broadcasts, &chan->half[direction].bcast);
+
 	chan->half[direction].channel_update
 		= tal_dup_arr(chan, u8, update, tal_count(update), 0);
 
@@ -1293,7 +1295,7 @@ bool routing_add_channel_update(struct routing_state *rstate,
 	 * broadcast them!  But save local ones to store anyway. */
 	if (!chan->channel_announce) {
 		if (is_local_channel(rstate, chan))
-			gossip_store_add(rstate->store,
+			gossip_store_add(rstate->broadcasts->gs,
 					 chan->half[direction].channel_update);
 		return true;
 	}
@@ -1304,11 +1306,12 @@ 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->channel_announcement_index == 0)
+	if (chan->bcast.index == 0)
 		add_channel_announce_to_broadcast(rstate, chan, timestamp);
 
-	persistent_broadcast(rstate, chan->half[direction].channel_update,
-			     timestamp);
+	insert_broadcast(rstate->broadcasts,
+			 chan->half[direction].channel_update,
+			 &chan->half[direction].bcast);
 	return true;
 }
 
@@ -1416,9 +1419,9 @@ u8 *handle_channel_update(struct routing_state *rstate, const u8 *update TAKES,
 
 	c = &chan->half[direction];
 
-	if (is_halfchan_defined(c) && timestamp <= c->last_timestamp) {
+	if (is_halfchan_defined(c) && timestamp <= c->bcast.timestamp) {
 		/* They're not supposed to do this! */
-		if (timestamp == c->last_timestamp
+		if (timestamp == c->bcast.timestamp
 		    && !memeq(c->channel_update, tal_count(c->channel_update),
 			      serialized, tal_count(serialized))) {
 			status_unusual("Bad gossip repeated timestamp for %s(%u): %s then %s",
@@ -1525,23 +1528,27 @@ bool routing_add_node_announcement(struct routing_state *rstate, const u8 *msg T
 	if (node == NULL)
 		return false;
 
+	tal_free(node->node_announcement);
+	/* Harmless if it was never added */
+	broadcast_del(rstate->broadcasts, &node->bcast);
+
 	wireaddrs = read_addresses(tmpctx, addresses);
 	tal_free(node->addresses);
 	node->addresses = tal_steal(node, wireaddrs);
 
-	node->last_timestamp = timestamp;
+	node->bcast.timestamp = timestamp;
 	memcpy(node->rgb_color, rgb_color, ARRAY_SIZE(node->rgb_color));
 	memcpy(node->alias, alias, ARRAY_SIZE(node->alias));
 	tal_free(node->globalfeatures);
 	node->globalfeatures = tal_steal(node, features);
 
-	tal_free(node->node_announcement);
 	node->node_announcement = tal_dup_arr(node, u8, msg, tal_count(msg), 0);
 
 	/* We might be waiting for channel_announce to be released. */
 	if (node_has_broadcastable_channels(node)) {
-		node->node_announcement_index = persistent_broadcast(
-		    rstate, node->node_announcement, timestamp);
+		insert_broadcast(rstate->broadcasts,
+				 node->node_announcement,
+				 &node->bcast);
 	}
 	return true;
 }
@@ -1669,7 +1676,7 @@ u8 *handle_node_announcement(struct routing_state *rstate, const u8 *node_ann)
 		return NULL;
 	}
 
-	if (node->node_announcement && node->last_timestamp >= timestamp) {
+	if (node->bcast.index && node->bcast.timestamp >= timestamp) {
 		SUPERVERBOSE("Ignoring node announcement, it's outdated.");
 		return NULL;
 	}
@@ -1866,14 +1873,14 @@ void route_prune(struct routing_state *rstate)
 
 		/* Rare case where we examine timestamp even without update;
 		 * it's used to prune channels where update never arrives */
-		if (chan->half[0].last_timestamp < highwater
-		    && chan->half[1].last_timestamp < highwater) {
+		if (chan->half[0].bcast.timestamp < highwater
+		    && chan->half[1].bcast.timestamp < highwater) {
 			status_trace(
 			    "Pruning channel %s from network view (ages %"PRIu64" and %"PRIu64"s)",
 			    type_to_string(tmpctx, struct short_channel_id,
 					   &chan->scid),
-			    now - chan->half[0].last_timestamp,
-			    now - chan->half[1].last_timestamp);
+			    now - chan->half[0].bcast.timestamp,
+			    now - chan->half[1].bcast.timestamp);
 
 			/* This may perturb iteration so do outside loop. */
 			tal_steal(pruned, chan);

gossipd/routing.h

@@ -25,10 +25,8 @@ struct half_chan {
 	/* Delay for HTLC in blocks.*/
 	u32 delay;
 
-	u32 last_timestamp;
-
-	/* Minimum and maximum number of msatoshi in an HTLC */
-	struct amount_msat htlc_minimum, htlc_maximum;
+	/* Timestamp and index into store file */
+	struct broadcastable bcast;
 
 	/* Flags as specified by the `channel_update`s, among other
 	 * things indicated direction wrt the `channel_id` */
@@ -37,6 +35,9 @@ struct half_chan {
 	/* Flags as specified by the `channel_update`s, indicates
 	 * optional fields.  */
 	u8 message_flags;
+
+	/* Minimum and maximum number of msatoshi in an HTLC */
+	struct amount_msat htlc_minimum, htlc_maximum;
 };
 
 struct chan {
@@ -53,8 +54,9 @@ struct chan {
 
 	/* NULL if not announced yet (ie. not public). */
 	const u8 *channel_announce;
-	/* Index in broadcast map, if public (otherwise 0) */
-	u64 channel_announcement_index;
+
+	/* Timestamp and index into store file */
+	struct broadcastable bcast;
 
 	/* Disabled locally (due to peer disconnect) */
 	bool local_disabled;
@@ -72,7 +74,7 @@ static inline bool is_chan_public(const struct chan *chan)
  * with it. */
 static inline bool is_chan_announced(const struct chan *chan)
 {
-	return chan->channel_announcement_index != 0;
+	return chan->bcast.index != 0;
 }
 
 static inline bool is_halfchan_defined(const struct half_chan *hc)
@@ -113,7 +115,8 @@ HTABLE_DEFINE_TYPE(struct chan, chan_map_scid, hash_scid, chan_eq_scid, chan_map
 struct node {
 	struct node_id id;
 
-	u32 last_timestamp;
+	/* Timestamp and index into store file */
+	struct broadcastable bcast;
 
 	/* IP/Hostname and port of this node (may be NULL) */
 	struct wireaddr *addresses;
@@ -145,8 +148,6 @@ struct node {
 
 	/* Cached `node_announcement` we might forward to new peers (or NULL). */
 	const u8 *node_announcement;
-	/* If public, this is non-zero. */
-	u64 node_announcement_index;
 };
 
 const struct node_id *node_map_keyof_node(const struct node *n);
@@ -200,6 +201,7 @@ struct routing_state {
 	/* channel_announcement which are pending short_channel_id lookup */
 	struct list_head pending_cannouncement;
 
+	/* Broadcast map, and access to gossip store */
 	struct broadcast_state *broadcasts;
 
 	/* Our own ID so we can identify local channels */
@@ -208,10 +210,6 @@ struct routing_state {
 	/* How old does a channel have to be before we prune it? */
 	u32 prune_timeout;
 
-	/* Store for processed messages that we might want to remember across
-	 * restarts */
-	struct gossip_store *store;
-
         /* A map of channels indexed by short_channel_ids */
 	UINTMAP(struct chan *) chanmap;
 

gossipd/test/run-bench-find_route.c

@@ -14,6 +14,7 @@
 
 #include "../routing.c"
 #include "../gossip_store.c"
+#include "../broadcast.c"
 
 void status_fmt(enum log_level level UNUSED, const char *fmt, ...)
 {
@@ -25,15 +26,7 @@ void status_fmt(enum log_level level UNUSED, const char *fmt, ...)
 	va_end(ap);
 }
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx UNNEEDED)
-{
-	return NULL;
-}
-
 /* AUTOGENERATED MOCKS START */
-/* Generated stub for broadcast_del */
-void broadcast_del(struct broadcast_state *bstate UNNEEDED, u64 index UNNEEDED, const u8 *payload UNNEEDED)
-{ fprintf(stderr, "broadcast_del called!\n"); abort(); }
 /* Generated stub for fromwire_channel_announcement */
 bool fromwire_channel_announcement(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, secp256k1_ecdsa_signature *node_signature_1 UNNEEDED, secp256k1_ecdsa_signature *node_signature_2 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_1 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_2 UNNEEDED, u8 **features UNNEEDED, struct bitcoin_blkid *chain_hash UNNEEDED, struct short_channel_id *short_channel_id UNNEEDED, struct node_id *node_id_1 UNNEEDED, struct node_id *node_id_2 UNNEEDED, struct pubkey *bitcoin_key_1 UNNEEDED, struct pubkey *bitcoin_key_2 UNNEEDED)
 { fprintf(stderr, "fromwire_channel_announcement called!\n"); abort(); }
@@ -70,15 +63,6 @@ int fromwire_peektype(const u8 *cursor UNNEEDED)
 /* Generated stub for fromwire_wireaddr */
 bool fromwire_wireaddr(const u8 **cursor UNNEEDED, size_t *max UNNEEDED, struct wireaddr *addr UNNEEDED)
 { fprintf(stderr, "fromwire_wireaddr called!\n"); abort(); }
-/* Generated stub for insert_broadcast */
-u64 insert_broadcast(struct broadcast_state *bstate UNNEEDED, const u8 *msg UNNEEDED,
-		     u32 timestamp UNNEEDED)
-{ fprintf(stderr, "insert_broadcast called!\n"); abort(); }
-/* Generated stub for next_broadcast */
-const u8 *next_broadcast(struct broadcast_state *bstate UNNEEDED,
-			 u32 timestamp_min UNNEEDED, u32 timestamp_max UNNEEDED,
-			 u64 *last_index UNNEEDED)
-{ fprintf(stderr, "next_broadcast called!\n"); abort(); }
 /* Generated stub for onion_type_name */
 const char *onion_type_name(int e UNNEEDED)
 { fprintf(stderr, "onion_type_name called!\n"); abort(); }

gossipd/test/run-find_route-specific.c

@@ -13,16 +13,9 @@
 
 #include "../routing.c"
 #include "../gossip_store.c"
-
-struct broadcast_state *new_broadcast_state(tal_t *ctx UNNEEDED)
-{
-	return NULL;
-}
+#include "../broadcast.c"
 
 /* AUTOGENERATED MOCKS START */
-/* Generated stub for broadcast_del */
-void broadcast_del(struct broadcast_state *bstate UNNEEDED, u64 index UNNEEDED, const u8 *payload UNNEEDED)
-{ fprintf(stderr, "broadcast_del called!\n"); abort(); }
 /* Generated stub for fromwire_channel_announcement */
 bool fromwire_channel_announcement(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, secp256k1_ecdsa_signature *node_signature_1 UNNEEDED, secp256k1_ecdsa_signature *node_signature_2 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_1 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_2 UNNEEDED, u8 **features UNNEEDED, struct bitcoin_blkid *chain_hash UNNEEDED, struct short_channel_id *short_channel_id UNNEEDED, struct node_id *node_id_1 UNNEEDED, struct node_id *node_id_2 UNNEEDED, struct pubkey *bitcoin_key_1 UNNEEDED, struct pubkey *bitcoin_key_2 UNNEEDED)
 { fprintf(stderr, "fromwire_channel_announcement called!\n"); abort(); }
@@ -59,15 +52,6 @@ int fromwire_peektype(const u8 *cursor UNNEEDED)
 /* Generated stub for fromwire_wireaddr */
 bool fromwire_wireaddr(const u8 **cursor UNNEEDED, size_t *max UNNEEDED, struct wireaddr *addr UNNEEDED)
 { fprintf(stderr, "fromwire_wireaddr called!\n"); abort(); }
-/* Generated stub for insert_broadcast */
-u64 insert_broadcast(struct broadcast_state *bstate UNNEEDED, const u8 *msg UNNEEDED,
-		     u32 timestamp UNNEEDED)
-{ fprintf(stderr, "insert_broadcast called!\n"); abort(); }
-/* Generated stub for next_broadcast */
-const u8 *next_broadcast(struct broadcast_state *bstate UNNEEDED,
-			 u32 timestamp_min UNNEEDED, u32 timestamp_max UNNEEDED,
-			 u64 *last_index UNNEEDED)
-{ fprintf(stderr, "next_broadcast called!\n"); abort(); }
 /* Generated stub for onion_type_name */
 const char *onion_type_name(int e UNNEEDED)
 { fprintf(stderr, "onion_type_name called!\n"); abort(); }
@@ -197,7 +181,7 @@ int main(void)
 	nc->delay = 5;
 	nc->channel_flags = 1;
 	nc->message_flags = 0;
-	nc->last_timestamp = 1504064344;
+	nc->bcast.timestamp = 1504064344;
 
 	/* {'active': True, 'short_id': '6989:2:1/0', 'fee_per_kw': 10, 'delay': 5, 'message_flags': 0, 'channel_flags': 0, 'destination': '03c173897878996287a8100469f954dd820fcd8941daed91c327f168f3329be0bf', 'source': '0230ad0e74ea03976b28fda587bb75bdd357a1938af4424156a18265167f5e40ae', 'last_update': 1504064344}, */
 	nc = get_or_make_connection(rstate, &b, &a, "6989x2x1", AMOUNT_SAT(1000));
@@ -206,7 +190,7 @@ int main(void)
 	nc->delay = 5;
 	nc->channel_flags = 0;
 	nc->message_flags = 0;
-	nc->last_timestamp = 1504064344;
+	nc->bcast.timestamp = 1504064344;
 
 	/* {'active': True, 'short_id': '6990:2:1/0', 'fee_per_kw': 10, 'delay': 5, 'message_flags': 0, 'channel_flags': 0, 'destination': '02ea622d5c8d6143f15ed3ce1d501dd0d3d09d3b1c83a44d0034949f8a9ab60f06', 'source': '0230ad0e74ea03976b28fda587bb75bdd357a1938af4424156a18265167f5e40ae', 'last_update': 1504064344}, */
 	nc = get_or_make_connection(rstate, &b, &c, "6990x2x1", AMOUNT_SAT(1000));
@@ -215,7 +199,7 @@ int main(void)
 	nc->delay = 5;
 	nc->channel_flags = 0;
 	nc->message_flags = 0;
-	nc->last_timestamp = 1504064344;
+	nc->bcast.timestamp = 1504064344;
 	nc->htlc_minimum = AMOUNT_MSAT(100);
 
 	/* {'active': True, 'short_id': '6989:2:1/1', 'fee_per_kw': 10, 'delay': 5, 'message_flags': 0, 'channel_flags': 1, 'destination': '0230ad0e74ea03976b28fda587bb75bdd357a1938af4424156a18265167f5e40ae', 'source': '03c173897878996287a8100469f954dd820fcd8941daed91c327f168f3329be0bf', 'last_update': 1504064344}]} */
@@ -225,7 +209,7 @@ int main(void)
 	nc->delay = 5;
 	nc->channel_flags = 1;
 	nc->message_flags = 0;
-	nc->last_timestamp = 1504064344;
+	nc->bcast.timestamp = 1504064344;
 
 	route = find_route(tmpctx, rstate, &a, &c, AMOUNT_MSAT(100000), riskfactor, 0.0, NULL,
 			   ROUTING_MAX_HOPS, &fee);
@@ -259,7 +243,7 @@ int main(void)
 	nc->delay = 5;
 	nc->channel_flags = 0;
 	nc->message_flags = 1;
-	nc->last_timestamp = 1504064344;
+	nc->bcast.timestamp = 1504064344;
 	nc->htlc_minimum = AMOUNT_MSAT(100);
 	nc->htlc_maximum = AMOUNT_MSAT(500000); /* half capacity */
 

gossipd/test/run-find_route.c

@@ -1,12 +1,8 @@
 #include "../routing.c"
 #include "../gossip_store.c"
+#include "../broadcast.c"
 #include <stdio.h>
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx UNNEEDED)
-{
-	return NULL;
-}
-
 void status_fmt(enum log_level level UNUSED, const char *fmt, ...)
 {
 	va_list ap;
@@ -18,9 +14,6 @@ void status_fmt(enum log_level level UNUSED, const char *fmt, ...)
 }
 
 /* AUTOGENERATED MOCKS START */
-/* Generated stub for broadcast_del */
-void broadcast_del(struct broadcast_state *bstate UNNEEDED, u64 index UNNEEDED, const u8 *payload UNNEEDED)
-{ fprintf(stderr, "broadcast_del called!\n"); abort(); }
 /* Generated stub for fromwire_channel_announcement */
 bool fromwire_channel_announcement(const tal_t *ctx UNNEEDED, const void *p UNNEEDED, secp256k1_ecdsa_signature *node_signature_1 UNNEEDED, secp256k1_ecdsa_signature *node_signature_2 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_1 UNNEEDED, secp256k1_ecdsa_signature *bitcoin_signature_2 UNNEEDED, u8 **features UNNEEDED, struct bitcoin_blkid *chain_hash UNNEEDED, struct short_channel_id *short_channel_id UNNEEDED, struct node_id *node_id_1 UNNEEDED, struct node_id *node_id_2 UNNEEDED, struct pubkey *bitcoin_key_1 UNNEEDED, struct pubkey *bitcoin_key_2 UNNEEDED)
 { fprintf(stderr, "fromwire_channel_announcement called!\n"); abort(); }
@@ -57,15 +50,6 @@ int fromwire_peektype(const u8 *cursor UNNEEDED)
 /* Generated stub for fromwire_wireaddr */
 bool fromwire_wireaddr(const u8 **cursor UNNEEDED, size_t *max UNNEEDED, struct wireaddr *addr UNNEEDED)
 { fprintf(stderr, "fromwire_wireaddr called!\n"); abort(); }
-/* Generated stub for insert_broadcast */
-u64 insert_broadcast(struct broadcast_state *bstate UNNEEDED, const u8 *msg UNNEEDED,
-		     u32 timestamp UNNEEDED)
-{ fprintf(stderr, "insert_broadcast called!\n"); abort(); }
-/* Generated stub for next_broadcast */
-const u8 *next_broadcast(struct broadcast_state *bstate UNNEEDED,
-			 u32 timestamp_min UNNEEDED, u32 timestamp_max UNNEEDED,
-			 u64 *last_index UNNEEDED)
-{ fprintf(stderr, "next_broadcast called!\n"); abort(); }
 /* Generated stub for onion_type_name */
 const char *onion_type_name(int e UNNEEDED)
 { fprintf(stderr, "onion_type_name called!\n"); abort(); }