gossipd: save gossip store writes, try them again (and fsync) if we get a read issue.

diff --git a/gossipd/gossip_store.c b/gossipd/gossip_store.c
index 3b61737a..15cae6bd 100644
--- a/gossipd/gossip_store.c
+++ b/gossipd/gossip_store.c
@@ -44,29 +44,11 @@ static void gossip_store_destroy(struct gossip_store *gs)
 	close(gs->fd);
 }
 
-#if HAVE_PWRITEV
-/* One fewer syscall for the win! */
-static ssize_t gossip_pwritev(int fd, const struct iovec *iov, int iovcnt,
-			      off_t offset)
+static bool append_msg(int fd, const u8 *msg, u32 timestamp, u64 *len,
+		       const u8 ***msgs)
 {
-	return pwritev(fd, iov, iovcnt, offset);
-}
-#else /* Hello MacOS! */
-static ssize_t gossip_pwritev(int fd, const struct iovec *iov, int iovcnt,
-			      off_t offset)
-{
-	if (lseek(fd, offset, SEEK_SET) != offset)
-		return -1;
-	return writev(fd, iov, iovcnt);
-}
-#endif /* !HAVE_PWRITEV */
-
-static bool append_msg(int fd, const u8 *msg, u32 timestamp, u64 *len)
-{
-	struct gossip_hdr hdr;
+	struct gossip_hdr *hdr;
 	u32 msglen;
-	struct iovec iov[2];
-	const u8 complete_byte = (GOSSIP_STORE_COMPLETED_BIT >> 8);
 
 	/* Don't ever overwrite the version header! */
 	assert(*len);
@@ -76,25 +58,25 @@ static bool append_msg(int fd, const u8 *msg, u32 timestamp, u64 *len)
 	msglen = tal_count(msg);
 	/* All messages begin with a 16-bit type */
 	assert(msglen >= 2);
-	hdr.len = cpu_to_be16(msglen);
-	hdr.flags = 0;
-	hdr.crc = cpu_to_be32(crc32c(timestamp, msg, msglen));
-	hdr.timestamp = cpu_to_be32(timestamp);
 
-	/* pwritev makes it more likely to appear at once, plus it's
-	 * exactly what we want. */
-	iov[0].iov_base = &hdr;
-	iov[0].iov_len = sizeof(hdr);
-	iov[1].iov_base = (void *)msg;
-	iov[1].iov_len = msglen;
-	if (gossip_pwritev(fd, iov, ARRAY_SIZE(iov), *len) != sizeof(hdr) + msglen)
+	hdr = (struct gossip_hdr *)tal_arr(tmpctx, u8, sizeof(*hdr) + msglen);
+	hdr->len = cpu_to_be16(msglen);
+	hdr->flags = 0;
+	hdr->crc = cpu_to_be32(crc32c(timestamp, msg, msglen));
+	hdr->timestamp = cpu_to_be32(timestamp);
+	memcpy(hdr + 1, msg, msglen);
+
+	if (pwrite(fd, hdr, sizeof(*hdr) + msglen, *len) != sizeof(*hdr) + msglen)
 		return false;
 
 	/* Update the hdr with the complete bit as a single-byte write */
-	if (pwrite(fd, &complete_byte, 1, *len) != 1)
+	hdr->flags = CPU_TO_BE16(GOSSIP_STORE_COMPLETED_BIT);
+	if (pwrite(fd, &hdr->flags, 1, *len) != 1)
 		return false;
 
-	*len += sizeof(hdr) + msglen;
+	*len += sizeof(*hdr) + msglen;
+	if (msgs)
+		tal_arr_expand(msgs, (const u8 *)tal_steal(*msgs, hdr));
 	return true;
 }
 
@@ -376,7 +358,7 @@ rename_new:
 	/* Create end marker now new file exists. */
 	if (old_fd != -1) {
 		append_msg(old_fd, towire_gossip_store_ended(tmpctx, *total_len),
-			   0, &old_len);
+			   0, &old_len, NULL);
 		close(old_fd);
 	}
 
@@ -426,7 +408,32 @@ void gossip_store_fsync(const struct gossip_store *gs)
 			      "gossmap fsync failed: %s", strerror(errno));
 }
 
-u64 gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg, u32 timestamp)
+void gossip_store_rewrite_end(struct gossip_store *gs, const u8 **msgs)
+{
+	u64 offset = gs->len;
+
+	for (size_t i = 0; i < tal_count(msgs); i++) {
+		/* Don't overwrite version byte */
+		assert(tal_bytelen(msgs[i]) < gs->len);
+		offset -= tal_bytelen(msgs[i]);
+	}
+
+	for (size_t i = 0; i < tal_count(msgs); i++) {
+		if (pwrite(gs->fd, msgs[i], tal_bytelen(msgs[i]), offset) != tal_bytelen(msgs[i]))
+			status_failed(STATUS_FAIL_INTERNAL_ERROR,
+				      "Failed to re-write %s at offset %"PRIu64,
+				      tal_hex(tmpctx, msgs[i]), offset);
+		offset += tal_bytelen(msgs[i]);
+	}
+
+	/* Hit it harder. */
+	gossip_store_fsync(gs);
+}
+
+u64 gossip_store_add(struct gossip_store *gs,
+		     const u8 *gossip_msg,
+		     u32 timestamp,
+		     const u8 ***msgs)
 {
 	u64 off = gs->len, filelen;
 
@@ -446,7 +453,7 @@ u64 gossip_store_add(struct gossip_store *gs, const u8 *gossip_msg, u32 timestam
 				      filelen, off);
 	}
 
-	if (!append_msg(gs->fd, gossip_msg, timestamp, &gs->len)) {
+	if (!append_msg(gs->fd, gossip_msg, timestamp, &gs->len, msgs)) {
 		status_failed(STATUS_FAIL_INTERNAL_ERROR,
 			      "Failed writing to gossip store: %s",
 			      strerror(errno));
diff --git a/gossipd/gossip_store.h b/gossipd/gossip_store.h
index c75696f0..96ff0937 100644
--- a/gossipd/gossip_store.h
+++ b/gossipd/gossip_store.h
@@ -49,11 +49,14 @@ void gossip_store_corrupt(void);
  * @gs: gossip store
  * @gossip_msg: the gossip message to insert.
  * @timestamp: the timestamp for filtering of this messsage.
+ * @msgs: the option pointer to a u8 *array to append the written msgs to.
+ *
+ * Returns the offset (after the gossip_hdr).
  */
 u64 gossip_store_add(struct gossip_store *gs,
 		     const u8 *gossip_msg,
-		     u32 timestamp);
-
+		     u32 timestamp,
+		     const u8 ***msgs);
 
 /**
  * Delete the record at this offset (offset is that of
@@ -106,6 +109,11 @@ u32 gossip_store_get_timestamp(struct gossip_store *gs, u64 offset);
  */
 void gossip_store_set_timestamp(struct gossip_store *gs, u64 offset, u32 timestamp);
 
+/**
+ * We've seen (ZFS on Linux) writes not show up in the gossip store.
+ * This lets us rewrite the last bytes. */
+void gossip_store_rewrite_end(struct gossip_store *gs, const u8 **msgs);
+
 /**
  * For debugging.
  */
diff --git a/gossipd/gossmap_manage.c b/gossipd/gossmap_manage.c
index 25ea55ad..816c3f00 100644
--- a/gossipd/gossmap_manage.c
+++ b/gossipd/gossmap_manage.c
@@ -67,6 +67,10 @@ struct gossmap_manage {
 	/* gossip map itself (access via gossmap_manage_get_gossmap, so it's fresh!) */
 	struct gossmap *raw_gossmap;
 
+	/* Last writes to gossmap since previous sync, in case it
+	 * messes up and we need to force it. */
+	const u8 **last_writes;
+
 	/* The gossip_store, which writes to the gossip_store file */
 	struct gossip_store *gs;
 
@@ -266,7 +270,7 @@ static void remove_channel(struct gossmap_manage *gm,
 	/* Put in tombstone marker. */
 	gossip_store_add(gm->gs,
 			 towire_gossip_store_delete_chan(tmpctx, scid),
-			 0);
+			 0, &gm->last_writes);
 
 	/* Delete from store */
 	gossip_store_del(gm->gs, chan->cann_off, WIRE_CHANNEL_ANNOUNCEMENT);
@@ -307,7 +311,7 @@ static void remove_channel(struct gossmap_manage *gm,
 			timestamp = gossip_store_get_timestamp(gm->gs, node->nann_off);
 
 			gossip_store_del(gm->gs, node->nann_off, WIRE_NODE_ANNOUNCEMENT);
-			offset = gossip_store_add(gm->gs, nannounce, timestamp);
+			offset = gossip_store_add(gm->gs, nannounce, timestamp, &gm->last_writes);
 		} else {
 			/* Are all remaining channels dying but we weren't?
 			 * Can happen if we removed this channel immediately
@@ -467,6 +471,7 @@ static bool setup_gossmap(struct gossmap_manage *gm,
 		gm->gs = tal_free(gm->gs);
 		return false;
 	}
+	gm->last_writes = tal_arr(gm, const u8 *, 0);
 	return true;
 }
 
@@ -617,9 +622,10 @@ const char *gossmap_manage_channel_announcement(const tal_t *ctx,
 	 */
 	if (known_amount) {
 		/* Set with timestamp 0 (we will update once we have a channel_update) */
-		gossip_store_add(gm->gs, announce, 0);
+		gossip_store_add(gm->gs, announce, 0, &gm->last_writes);
 		gossip_store_add(gm->gs,
-				 towire_gossip_store_channel_amount(tmpctx, *known_amount), 0);
+				 towire_gossip_store_channel_amount(tmpctx, *known_amount), 0,
+				 &gm->last_writes);
 
 		node_announcements_not_dying(gm, gossmap, pca);
 		tal_free(pca);
@@ -743,9 +749,10 @@ void gossmap_manage_handle_get_txout_reply(struct gossmap_manage *gm, const u8 *
 	}
 
 	/* Set with timestamp 0 (we will update once we have a channel_update) */
-	gossip_store_add(gm->gs, pca->channel_announcement, 0);
+	gossip_store_add(gm->gs, pca->channel_announcement, 0, &gm->last_writes);
 	gossip_store_add(gm->gs,
-			 towire_gossip_store_channel_amount(tmpctx, sat), 0);
+			 towire_gossip_store_channel_amount(tmpctx, sat), 0,
+			 &gm->last_writes);
 
 	/* If we looking specifically for this, we no longer are. */
 	remove_unknown_scid(gm->daemon->seeker, &scid, true);
@@ -847,7 +854,7 @@ static const char *process_channel_update(const tal_t *ctx,
 	}
 
 	/* OK, apply the new one */
-	offset = gossip_store_add(gm->gs, update, timestamp);
+	offset = gossip_store_add(gm->gs, update, timestamp, &gm->last_writes);
 
 	/* If channel is dying, make sure update is also marked dying! */
 	if (gossmap_chan_is_dying(gossmap, chan)) {
@@ -1011,7 +1018,7 @@ static void process_node_announcement(struct gossmap_manage *gm,
 	}
 
 	/* OK, apply the new one */
-	offset = gossip_store_add(gm->gs, nannounce, timestamp);
+	offset = gossip_store_add(gm->gs, nannounce, timestamp, &gm->last_writes);
 	/* If all channels are dying, make sure this is marked too. */
 	if (all_node_channels_dying(gossmap, node, NULL)) {
 		gossip_store_set_flag(gm->gs, offset,
@@ -1347,7 +1354,7 @@ void gossmap_manage_channel_spent(struct gossmap_manage *gm,
 
 	/* Save to gossip_store in case we restart */
 	msg = towire_gossip_store_chan_dying(tmpctx, cd.scid, cd.deadline);
-	cd.gossmap_offset = gossip_store_add(gm->gs, msg, 0);
+	cd.gossmap_offset = gossip_store_add(gm->gs, msg, 0, &gm->last_writes);
 	tal_arr_expand(&gm->dying_channels, cd);
 
 	/* Mark it dying, so we don't gossip it */
@@ -1449,6 +1456,9 @@ struct gossmap *gossmap_manage_get_gossmap(struct gossmap_manage *gm)
 			      map_used, map_size,
 			      tal_hex(tmpctx, remainder_fd));
 		gossmap_disable_mmap(gm->raw_gossmap);
+
+		/* Try rewriting the last few records, syncing. */
+		gossip_store_rewrite_end(gm->gs, gm->last_writes);
 		gossmap_refresh(gm->raw_gossmap);
 
 		map_used = gossmap_lengths(gm->raw_gossmap, &map_size);
@@ -1459,6 +1469,9 @@ struct gossmap *gossmap_manage_get_gossmap(struct gossmap_manage *gm)
 		}
 	}
 
+	/* Free up last_writes, since we've seen it on disk */
+	tal_free(gm->last_writes);
+	gm->last_writes = tal_arr(gm, const u8 *, 0);
 	return gm->raw_gossmap;
 }