#include "gossip_store.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #define GOSSIP_STORE_FILENAME "gossip_store" #define GOSSIP_STORE_TEMP_FILENAME "gossip_store.tmp" struct gossip_store { int fd; u8 version; /* 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; /* Disable compaction if we encounter an error during a prior * compaction */ bool disable_compaction; }; 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 *gs = tal(ctx, struct gossip_store); gs->count = 0; gs->fd = open(GOSSIP_STORE_FILENAME, O_RDWR|O_APPEND|O_CREAT, 0600); gs->broadcast = broadcast; gs->rstate = rstate; gs->disable_compaction = false; tal_add_destructor(gs, gossip_store_destroy); /* Try to read the version, write it if this is a new file, or truncate * if the version doesn't match */ if (read(gs->fd, &gs->version, sizeof(gs->version)) == sizeof(gs->version)) { /* Version match? All good */ if (gs->version == GOSSIP_STORE_VERSION) return gs; status_unusual("Gossip store version %u not %u: removing", gs->version, GOSSIP_STORE_VERSION); if (ftruncate(gs->fd, 0) != 0) status_failed(STATUS_FAIL_INTERNAL_ERROR, "Truncating store: %s", strerror(errno)); } /* Empty file, write version byte */ gs->version = GOSSIP_STORE_VERSION; if (write(gs->fd, &gs->version, sizeof(gs->version)) != sizeof(gs->version)) status_failed(STATUS_FAIL_INTERNAL_ERROR, "Writing version to store: %s", strerror(errno)); return gs; } static u8 *gossip_store_wrap_channel_announcement(const tal_t *ctx, struct routing_state *rstate, const u8 *gossip_msg) { secp256k1_ecdsa_signature node_signature_1, node_signature_2; secp256k1_ecdsa_signature bitcoin_signature_1, bitcoin_signature_2; u8 *features; struct bitcoin_blkid chain_hash; struct short_channel_id scid; struct node_id node_id_1; struct node_id node_id_2; struct pubkey bitcoin_key_1; struct pubkey bitcoin_key_2; /* Which channel are we talking about here? */ if (!fromwire_channel_announcement( tmpctx, gossip_msg, &node_signature_1, &node_signature_2, &bitcoin_signature_1, &bitcoin_signature_2, &features, &chain_hash, &scid, &node_id_1, &node_id_2, &bitcoin_key_1, &bitcoin_key_2)) status_failed(STATUS_FAIL_INTERNAL_ERROR, "Error parsing channel_announcement"); struct chan *chan = get_channel(rstate, &scid); assert(chan && amount_sat_greater(chan->sat, AMOUNT_SAT(0))); u8 *msg = towire_gossip_store_channel_announcement(ctx, gossip_msg, chan->sat); return msg; } /** * Wrap the raw gossip message and write it to fd * * @param fd File descriptor to write the wrapped message into * @param gossip_msg The message to write * @return true if the message was wrapped and written */ static bool gossip_store_append(int fd, struct routing_state *rstate, const u8 *gossip_msg) { int t = fromwire_peektype(gossip_msg); u32 msglen; beint32_t checksum, belen; const u8 *msg; if (t == WIRE_CHANNEL_ANNOUNCEMENT) msg = gossip_store_wrap_channel_announcement(tmpctx, rstate, gossip_msg); else if(t == WIRE_CHANNEL_UPDATE) msg = towire_gossip_store_channel_update(tmpctx, gossip_msg); else if(t == WIRE_NODE_ANNOUNCEMENT) msg = towire_gossip_store_node_announcement(tmpctx, gossip_msg); else if(t == WIRE_GOSSIPD_LOCAL_ADD_CHANNEL) msg = towire_gossip_store_local_add_channel(tmpctx, gossip_msg); else if(t == WIRE_GOSSIP_STORE_CHANNEL_DELETE) msg = gossip_msg; else { status_trace("Unexpected message passed to gossip_store: %s", wire_type_name(t)); return false; } msglen = tal_count(msg); belen = cpu_to_be32(msglen); checksum = cpu_to_be32(crc32c(0, msg, msglen)); return (write(fd, &belen, sizeof(belen)) == sizeof(belen) && write(fd, &checksum, sizeof(checksum)) == sizeof(checksum) && write(fd, msg, msglen) == msglen); } /* Local unannounced channels don't appear in broadcast map, but we need to * remember them anyway, so we manually append to the store. * * Note these do *not* add to gs->count, since that's compared with * the broadcast map count. */ static bool add_local_unnannounced(int fd, struct routing_state *rstate, struct node *self) { struct chan_map_iter i; struct chan *c; for (c = first_chan(self, &i); c; c = next_chan(self, &i)) { struct node *peer = other_node(self, c); const u8 *msg; /* Ignore already announced. */ if (c->channel_announce) continue; msg = towire_gossipd_local_add_channel(tmpctx, &c->scid, &peer->id, c->sat); if (!gossip_store_append(fd, rstate, msg)) return false; for (size_t i = 0; i < 2; i++) { msg = c->half[i].channel_update; if (!msg) continue; if (!gossip_store_append(fd, rstate, msg)) return false; } } return true; } /** * Rewrite the on-disk gossip store, compacting it along the way * * 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) { size_t count = 0; u64 index = 0; int fd; const u8 *msg; struct node *self; assert(gs->broadcast); status_trace( "Compacting gossip_store with %zu entries, %zu of which are stale", gs->count, gs->count - gs->broadcast->count); fd = open(GOSSIP_STORE_TEMP_FILENAME, O_RDWR|O_APPEND|O_CREAT, 0600); if (fd < 0) { status_broken( "Could not open file for gossip_store compaction"); goto disable; } if (write(fd, &gs->version, sizeof(gs->version)) != sizeof(gs->version)) { status_broken("Writing version to store: %s", strerror(errno)); goto unlink_disable; } while ((msg = next_broadcast(gs->broadcast, 0, UINT32_MAX, &index)) != NULL) { if (!gossip_store_append(fd, gs->rstate, msg)) { status_broken("Failed writing to gossip store: %s", strerror(errno)); goto unlink_disable; } count++; } /* 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)) { status_broken("Failed writing unannounced to gossip store: %s", strerror(errno)); goto unlink_disable; } if (rename(GOSSIP_STORE_TEMP_FILENAME, GOSSIP_STORE_FILENAME) == -1) { status_broken( "Error swapping compacted gossip_store into place: %s", strerror(errno)); goto unlink_disable; } status_trace( "Compaction completed: dropped %zu messages, new count %zu", gs->count - count, count); gs->count = count; close(gs->fd); gs->fd = fd; return true; unlink_disable: unlink(GOSSIP_STORE_TEMP_FILENAME); disable: status_trace("Encountered an error while compacting, disabling " "future compactions."); gs->disable_compaction = true; return false; } void gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg) { /* Only give error message once. */ if (gs->fd == -1) return; if (!gossip_store_append(gs->fd, gs->rstate, gossip_msg)) { status_broken("Failed writing to gossip store: %s", strerror(errno)); gs->fd = -1; } gs->count++; if (gs->count >= 1000 && gs->count > gs->broadcast->count * 1.25 && !gs->disable_compaction) gossip_store_compact(gs); } 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); tal_free(msg); } void gossip_store_load(struct routing_state *rstate, struct gossip_store *gs) { beint32_t belen, becsum; u32 msglen, checksum; 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) { status_unusual("gossip_store: lseek failure"); goto truncate_nomsg; } while (read(fd, &belen, sizeof(belen)) == sizeof(belen) && read(fd, &becsum, sizeof(becsum)) == sizeof(becsum)) { msglen = be32_to_cpu(belen); checksum = be32_to_cpu(becsum); msg = tal_arr(gs, u8, msglen); if (read(fd, msg, msglen) != msglen) { status_unusual("gossip_store: truncated file?"); goto truncate_nomsg; } if (checksum != crc32c(0, msg, msglen)) { bad = "Checksum verification failed"; goto truncate; } if (fromwire_gossip_store_channel_announcement(msg, msg, &gossip_msg, &satoshis)) { if (!routing_add_channel_announcement(rstate, take(gossip_msg), satoshis)) { bad = "Bad channel_announcement"; goto truncate; } stats[0]++; } else if (fromwire_gossip_store_channel_update(msg, msg, &gossip_msg)) { if (!routing_add_channel_update(rstate, take(gossip_msg))) { bad = "Bad channel_update"; goto truncate; } stats[1]++; } else if (fromwire_gossip_store_node_announcement(msg, msg, &gossip_msg)) { if (!routing_add_node_announcement(rstate, take(gossip_msg))) { bad = "Bad node_announcement"; goto truncate; } stats[2]++; } else if (fromwire_gossip_store_channel_delete(msg, &scid)) { struct chan *c = get_channel(rstate, &scid); if (!c) { bad = "Bad channel_delete"; goto truncate; } tal_free(c); stats[3]++; } else if (fromwire_gossip_store_local_add_channel( msg, msg, &gossip_msg)) { handle_local_add_channel(rstate, gossip_msg); } else { bad = "Unknown message"; goto truncate; } known_good += sizeof(belen) + msglen; gs->count++; tal_free(msg); } goto out; truncate: status_unusual("gossip_store: %s (%s) truncating to %"PRIu64, bad, tal_hex(msg, msg), (u64)1); truncate_nomsg: /* FIXME: We would like to truncate to known_good, except we would * miss channel_delete msgs. If we put block numbers into the store * as we process them, we can know how far we need to roll back if we * truncate the store */ if (ftruncate(fd, 1) != 0) status_failed(STATUS_FAIL_INTERNAL_ERROR, "Truncating store: %s", strerror(errno)); out: #if DEVELOPER 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); #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->fd = fd; }