gossip_store: wait for completed bit on reading.

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

Author: rustyrussell

common/gossip_store.c

@@ -30,6 +30,8 @@ bool gossip_store_readhdr(int gossip_store_fd, size_t off,
 	r = pread(gossip_store_fd, &buf, HDR_AND_TYPE_SIZE, off);
 	if (r != HDR_AND_TYPE_SIZE)
 		return false;
+	if (!(buf.hdr.flags & CPU_TO_BE16(GOSSIP_STORE_COMPLETED_BIT)))
+		return false;
 	*len = be16_to_cpu(buf.hdr.len);
 	if (flags)
 		*flags = be16_to_cpu(buf.hdr.flags);

common/gossmap.c

@@ -694,10 +694,15 @@ static bool map_catchup(struct gossmap *map, bool must_be_clean, bool *changed)
 		reclen = msglen + sizeof(ghdr);
 
 		flags = be16_to_cpu(ghdr.flags);
+
+		/* Not finished, this can happen. */
+		if (!(flags & GOSSIP_STORE_COMPLETED_BIT))
+			break;
+
 		if (flags & GOSSIP_STORE_DELETED_BIT)
 			continue;
 
-		/* Partial write, this can happen. */
+		/* Partial write, should not happen with completed records. */
 		if (map->map_end + reclen > map->map_size)
 			break;
 

contrib/pyln-client/pyln/client/gossmap.py

@@ -392,8 +392,7 @@ class Gossmap(object):
     def __init__(self, store_filename: str = "gossip_store"):
         self.store_filename = store_filename
         self.store_file = open(store_filename, "rb")
-        self.store_buf = bytes()
-        self.bytes_read = 0
+        self.bytes_read = 1
         self.nodes: Dict[GossmapNodeId, GossmapNode] = {}
         self.channels: Dict[ShortChannelId, GossmapChannel] = {}
         self._last_scid: Optional[str] = None
@@ -592,24 +591,24 @@ def _remove_channel_by_deletemsg(self, rec: bytes):
         if scid in self.channels:
             self._del_channel(scid)
 
-    def _pull_bytes(self, length: int) -> bool:
-        """Pull bytes from file into our internal buffer"""
-        if len(self.store_buf) < length:
-            self.store_buf += self.store_file.read(length - len(self.store_buf))
-        self.bytes_read += len(self.store_buf)
-        return len(self.store_buf) >= length
-
     def _read_record(self) -> Optional[bytes]:
-        """If a whole record is not in the file, returns None.
-        If deleted, returns empty."""
-        off = self.bytes_read + 1
-        if not self._pull_bytes(12):
+        """If a whole record is not in the file, returns None, None."""
+        prev_off = self.bytes_read
+        hdr = self.store_file.read(12)
+        if len(hdr) != 12:
+            self.store_file.seek(prev_off)
+            return None, None
+        hdr = GossipStoreMsgHeader(hdr, prev_off)
+        rec = self.store_file.read(hdr.length)
+        if len(rec) != hdr.length:
+            self.store_file.seek(prev_off)
             return None, None
-        hdr = GossipStoreMsgHeader(self.store_buf[:12], off)
-        if not self._pull_bytes(12 + hdr.length):
-            return None, hdr
-        rec = self.store_buf[12:]
-        self.store_buf = bytes()
+        if (hdr.flags & GOSSIP_STORE_LEN_COMPLETE_BIT) == 0:
+            self.store_file.seek(prev_off)
+            return None, None
+
+        # Ok, we're digesting this one, so increment bytes_read.
+        self.bytes_read += 12 + hdr.length
         return rec, hdr
 
     def refresh(self):

devtools/dump-gossipstore.c

@@ -67,19 +67,21 @@ int main(int argc, char *argv[])
 		u16 flags = be16_to_cpu(hdr.flags);
 		u16 msglen = be16_to_cpu(hdr.len);
 		u8 *msg, *inner;
-		bool deleted, dying;
+		bool deleted, dying, complete;
 		u32 blockheight;
 
 		deleted = (flags & GOSSIP_STORE_DELETED_BIT);
 		dying = (flags & GOSSIP_STORE_DYING_BIT);
+		complete = (flags & GOSSIP_STORE_COMPLETED_BIT);
 
 		msg = tal_arr(NULL, u8, msglen);
 		if (read(fd, msg, msglen) != msglen)
 			errx(1, "%zu: Truncated file?", off);
 
-		printf("%zu: %s%s%s", off,
+		printf("%zu: %s%s%s%s", off,
 		       deleted ? "DELETED " : "",
 		       dying ? "DYING " : "",
+		       complete ? "" : "**INCOMPLETE** ",
 		       be32_to_cpu(hdr.crc) != crc32c(be32_to_cpu(hdr.timestamp), msg, msglen) ? "**BAD CHECKSUM** " : "");
 
 		if (print_timestamp)