cubelinux: read format v2 as well as v1

The store's write path is decided (append a log, fold it into the image at a
checkpoint; DESIGN-cubelinux-write-path.md), and that decision forces the log to
sit right after the image on the device. v1 had no extent and no count — it
walked records until it met zero padding — so a v1 reader would have walked
straight into the log's header and parsed it as a record.

v2 states the image's byte extent and its record count, and this teaches the
kernel reader both:

- v1: walk from a 6-byte header until the trailing zeros, as before.
- v2: walk from a 22-byte header, stop exactly at the declared count, and never
  read past the declared extent. A zero frame inside the count is a record, not
  padding — which is the ambiguity v1 could not resolve.
- a v2 header whose extent does not cover the header itself is refused rather
  than guessed at, and a count that is not met counts as an error instead of
  quietly returning a shorter list.

Gate, unchanged in method: the kernel's digest of /dev/vda must equal
cube-image's digest of the same bytes. Passing on all three:

  v1 curated   11 records       fnv1a64=161113085b1573b2
  v1 snapshot  35,318 records   fnv1a64=5e20f98455387b08
  v2 store     4 records        fnv1a64=20ecadb5cdc9c994

drivers/cube/cubelinux_store.rs

@@ -57,8 +57,12 @@ module! {
 /// `crates/cube-store-raw` is the source of truth; the digest comparison is what keeps
 /// this copy honest.
 const MAGIC: &[u8; 4] = b"CUBE";
-const VERSION: u8 = 1;
-const HEADER_LEN: usize = 6;
+/// The original format: magic, version, curve, then records until zero padding.
+const VERSION_V1: u8 = 1;
+/// The current format: the same, plus the image's byte extent and record count.
+const VERSION: u8 = 2;
+const HEADER_LEN_V1: usize = 6;
+const HEADER_LEN_V2: usize = 6 + 8 + 8;
 const SPACE_ID_LEN: usize = 32;
 const RAW_KEY_LEN: usize = 24;
 const RECORD_FIXED: usize = SPACE_ID_LEN + RAW_KEY_LEN + 8;
@@ -169,7 +173,8 @@ fn read_image(image: &mut KVVec<u8>) -> Result<()> {
     // only meaningful for locks that no one holds here.
     unsafe { bindings::filp_close(file, core::ptr::null_mut()) };
 
-    if result.is_ok() && image.len() < HEADER_LEN {
+    // Less than the shortest header is not an image at all; `digest` reports the rest.
+    if result.is_ok() && image.len() < HEADER_LEN_V1 {
         return Err(EINVAL);
     }
     result
@@ -182,29 +187,91 @@ fn read_image(image: &mut KVVec<u8>) -> Result<()> {
 /// value runs past the buffer is a truncated record and an error, and an all-zero frame is
 /// *end of records* only when every remaining byte is also zero — otherwise a real record
 /// at the origin, followed by padding, would be read as an empty one.
-fn digest(image: &[u8]) -> Result<Line> {
+/// What a header says, in both formats.
+struct Header {
+    version: u8,
+    curve: u8,
+    image_bytes: Option<u64>,
+    record_count: Option<u64>,
+}
+
+/// Read the header, or say what is wrong with it.
+fn parse_header(image: &[u8]) -> Result<Header, &'static str> {
+    if image.len() < 5 || &image[0..4] != MAGIC {
+        return Err("not-a-store");
+    }
+    let version = image[4];
+    let curve = image[5];
+    match version {
+        VERSION_V1 => Ok(Header {
+            version,
+            curve,
+            image_bytes: None,
+            record_count: None,
+        }),
+        VERSION => {
+            if image.len() < HEADER_LEN_V2 {
+                return Err("short-v2-header");
+            }
+            let mut word = [0u8; 8];
+            word.copy_from_slice(&image[6..14]);
+            let image_bytes = u64::from_le_bytes(word);
+            word.copy_from_slice(&image[14..22]);
+            let record_count = u64::from_le_bytes(word);
+            // A header that does not cover itself is corrupt, and guessing at the extent
+            // of an image means possibly reading somebody else's bytes.
+            if (image_bytes as usize) < HEADER_LEN_V2 {
+                return Err("bad-extent");
+            }
+            Ok(Header {
+                version,
+                curve,
+                image_bytes: Some(image_bytes),
+                record_count: Some(record_count),
+            })
+        }
+        _ => Err("unsupported-version"),
+    }
+}
+
+fn digest(image: &[u8]) -> Line {
     let mut line = Line::new();
 
-    if image.len() < HEADER_LEN || &image[0..4] != MAGIC {
-        let _ = write!(line, "error=not-a-store");
-        return Ok(line);
-    }
-    let curve = image[5];
-    if image[4] != VERSION {
-        let _ = write!(line, "error=version-{}", image[4]);
-        return Ok(line);
+    let header = match parse_header(image) {
+        Ok(h) => h,
+        Err(what) => {
+            let _ = write!(line, "error={what}");
+            return line;
         }
+    };
+    let end = match header.image_bytes {
+        Some(n) => core::cmp::min(n as usize, image.len()),
+        None => image.len(),
+    };
 
     let mut h = FNV_OFFSET;
     let mut count: u64 = 0;
     let mut value_bytes: u64 = 0;
     let mut errors: u64 = 0;
-    let mut off = HEADER_LEN;
+    let mut off = if header.version == VERSION_V1 {
+        HEADER_LEN_V1
+    } else {
+        HEADER_LEN_V2
+    };
+    let mut remaining = header.record_count;
 
-    while off + RECORD_FIXED <= image.len() {
+    while off + RECORD_FIXED <= end {
+        if remaining == Some(0) {
+            break;
+        }
         let frame = &image[off..off + RECORD_FIXED];
-        // Padding is not a record — but only if it is padding all the way down.
-        if frame.iter().all(|b| *b == 0) && image[off..].iter().all(|b| *b == 0) {
+        // Padding is not a record — but only if it is padding all the way down, and only
+        // in v1, where nothing else says where the records stop. A v2 image states its
+        // count, so a zero frame inside that count is a record like any other.
+        if remaining.is_none()
+            && frame.iter().all(|b| *b == 0)
+            && image[off..].iter().all(|b| *b == 0)
+        {
             break;
         }
 
@@ -215,7 +282,9 @@ fn digest(image: &[u8]) -> Result<Line> {
         let value_len = u64::from_le_bytes(len_bytes) as usize;
 
         let value_at = off + RECORD_FIXED;
-        if value_at + value_len > image.len() {
+        // A promised record that is not there is a truncated image, and saying so beats
+        // returning a shorter list that looks complete.
+        if value_at + value_len > end {
             errors += 1;
             break;
         }
@@ -228,20 +297,30 @@ fn digest(image: &[u8]) -> Result<Line> {
         count += 1;
         value_bytes += value_len as u64;
         off = value_at + value_len;
+        if let Some(n) = remaining.as_mut() {
+            *n -= 1;
+        }
+    }
+    // The count was a promise; not meeting it is an error the caller must see.
+    if let Some(n) = remaining {
+        if n > 0 {
+            errors += 1;
+        }
     }
 
     // Field order matches cube-image's `digest` line so the two diff directly.
     let _ = write!(
         line,
-        "digest curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors={}",
-        curve,
+        "digest version={} curve={} bytes={} records={} value_bytes={} fnv1a64={:016x} errors={}",
+        header.version,
+        header.curve,
         image.len(),
         count,
         value_bytes,
         h,
         errors
     );
-    Ok(line)
+    line
 }
 
 /// The module's registration; holds the misc device for as long as the module lives.
@@ -285,7 +364,7 @@ impl MiscDevice for CubeStore {
         // open this device, the block driver that provides the store is certainly up.
         let mut image = KVVec::<u8>::new();
         let line = match read_image(&mut image) {
-            Ok(()) => digest(&image)?,
+            Ok(()) => digest(&image),
             Err(e) => {
                 let mut line = Line::new();
                 let _ = write!(line, "error={:?}", e);